worker

type AddressConfirmationMessage ¶

type AddressConfirmationMessage struct {
	Send notification.SendAddressConfirmation
}

AddressConfirmationMessage is the queue's way into sending the link that confirms a managed account's new address (ADR-0074 §6). Detached, PasswordResetMessage's reasoning: it mints a credential, then reaches an SMTP server, and no transaction is held across somebody else's machine. A retry mints again, and the newest mail's link is the one that works.

func (h AddressConfirmationMessage) OwnsItsTransactions()

func (h AddressConfirmationMessage) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AiEmbedding ¶

type AiEmbedding struct {
	Embed work.EmbedItems
	// Interval is the wait after a pass that found nothing left. What a quiet workspace pays for
	// having the machinery at all.
	Interval time.Duration
	// Continuation is the wait after a pass that filled its batch: there is known work left, and
	// the only reason not to do it now is that a batch is where one provider call ends.
	Continuation time.Duration
}

AiEmbedding keeps one workspace's vectors in step with its entries.

The retention sweep's shape rather than the suggestion's: a batch, then a decision about whether to come straight back, and a job that is never finished for good - the next entry somebody renames is always coming, and a row that removed itself would leave the workspace with no pass until its next write.

Detached, because the provider call inside it reaches somebody else's machine and a transaction held across that is what observability-reliability.md §8 forbids. The pass owns its two transactions: one to read what is owed, one to write what came back.

**It stops rather than fails when there is nothing to do**, and there are three ways to have nothing to do: no embedding store, no provider that can embed, and nothing owed. Semantic search is optional at every one of those levels, and a handler that treated any of them as an error would fill a dead letter queue with a feature somebody switched off.

func (h AiEmbedding) OwnsItsTransactions()

func (h AiEmbedding) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AiSuggestion ¶

type AiSuggestion struct {
	Produce suggestion.Produce
	// Jobs is the fence: the answer is recorded holding the job's row, so a question stopped while
	// the provider answered records nothing (UC-AI-04 check 9). Nil holds nothing.
	Jobs JobHolder
}

AiSuggestion is the queue's way into asking a provider what something should become: an inbound adapter, like every other handler, translating a job into a call on the application layer.

Detached, and it has to be. The call reaches a provider between two reads, and a transaction held open across it is what observability-reliability.md §8 forbids - the notification delivery's reasoning, for the same kind of dependency. What is given up is smaller here than there: a process that dies after the provider answered loses the answer, and the retry asks again. A duplicate suggestion is a record somebody dismisses.

func (h AiSuggestion) OwnsItsTransactions()

func (h AiSuggestion) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AiSuggestionArrival ¶

type AiSuggestionArrival struct {
	Produce suggestion.Produce
}

AiSuggestionArrival is the queue's way into a proposal the workspace prepares about one jumble arrival (UC-JUM-07 check 10). Detached for AiSuggestion's reason: a provider sits between two reads.

The actor is the system acting for the workspace - a tenant and no account - because nobody asked: the workspace switched it on (UC-JUM-07 check 10), and the embedding pass is the other work done that way.

func (h AiSuggestionArrival) OwnsItsTransactions()

func (h AiSuggestionArrival) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AuditAnchoring ¶

type AuditAnchoring struct {
	Anchoring service.Anchoring
	// Fallback bounds the wait: the next moment is tomorrow's, and a lease longer than a day
	// deserves nothing.
	Fallback time.Duration
}

AuditAnchoring is one tenant's daily anchoring (audit.md §3): the chain's end written to the backup target the workspace named, then the next day's moment.

The backup schedule's shape - one job per tenant, seeded by the configuration's write, rescheduling itself, finishing when the workspace names no target - and Detached for the export's reason: the write goes to somebody else's machine, and the row that records it is written in a short transaction of its own afterwards.

func (h AuditAnchoring) OwnsItsTransactions()

func (h AuditAnchoring) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AuditExport ¶

type AuditExport struct {
	Archivist service.Archivist
}

AuditExport is the queue's way into writing a period of the trail to a backup target (audit.md §5).

Detached, for the reason a backup run is: it streams as much evidence as the tenant produced to somebody else's machine, and doing that inside the runner's own transaction would hold one open for minutes on the pool the API shares.

Safe to repeat, which is what makes it a job at all. An export is a pure read followed by a write under a name derived from the tenant and the period, so an attempt that takes over after a worker died writes the same archive over the same keys rather than a second one somewhere else.

func (h AuditExport) OwnsItsTransactions()

func (h AuditExport) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AuditPartitions ¶

type AuditPartitions interface {
	Ensure(ctx context.Context, month time.Time) (string, error)
}

AuditPartitions keeps the audit trail's partitions conforming (audit.md §3).

type AuditTrailRemoved ¶

type AuditTrailRemoved struct {
	Record service.RecordTrailRemoval
}

AuditTrailRemoved records in one workspace's trail that a month of it went (audit.md §3).

Not detached: the entry and the job's completion are one transaction in that workspace's scope, so a run that fails after the entry is rolled back whole and its retry stores exactly one.

func (h AuditTrailRemoved) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AutomationCheck ¶

type AutomationCheck struct {
	Check automation.CheckRules
}

AutomationCheck is one workspace's check of its rules, as a deletion seeded it (ADR-0060).

Not Detached: everything the check does is a write - findings recorded, a broken rule switched off with its audit entry and its notification - and those belong in the runner's transaction with the job's own completion, so that a process that dies halfway leaves the rules as they were.

func (h AutomationCheck) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AutomationRun ¶

type AutomationRun struct {
	Engine automation.RunRule
	// Rules answers what to do about a failure. Read here rather than returned by the engine
	// because it is the *queue's* question - the engine records what happened, and this layer
	// decides whether the job comes back.
	Rules automation.RuleReader
}

AutomationRun is the queue's way into the engine.

An inbound adapter like every other handler: it translates a job into a call on the application layer and the answer into the queue's vocabulary. What is particular to it is `on_error: RETRY` - the one of the three values that is not the engine's to implement, because "retry with exponential backoff, and the dead letter after the budget" is what the queue already is (automation.md §2). A second backoff inside the engine would be a second answer to a question this system has answered.

It runs inside the runner's transaction, which is what makes at-least-once delivery safe to build on: the run row, the effects its actions had, the idempotency records and the job's completion commit together.

func (a AutomationRun) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type AutomationScheduling ¶

type AutomationScheduling struct {
	Pass automation.SchedulePass
	// Fallback is when to come back if the tenant owes nothing at all. It is not a poll: the pass
	// reschedules itself to the moment it actually owes, and this is only the cap on a lease - a
	// rule whose next moment is a year away does not deserve one that long.
	Fallback time.Duration
}

AutomationScheduling is one tenant's wake-up for its SCHEDULE rules.

The same shape as the backup scheduling and the reminders, and for the same reason: nothing in this system may enumerate tenants, so a scheduler cannot create one job per tenant even if it wanted to (multi-tenancy.md §2.1). The write that creates or enables a scheduled rule seeds this job for its own tenant, each round reschedules itself to the moment the tenant next owes something, and a tenant that owes nothing lets its poller finish - the next write re-seeds it.

It is *not* Detached. Everything it does is a write - runs queued and rules moved on - and those belong in the runner's transaction with the job's own completion, so that a process that dies halfway leaves neither a queued run nor a rule advanced past a moment nothing acted on.

func (h AutomationScheduling) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type BackupExpiry ¶

type BackupExpiry struct {
	Performer service.Performer
	Schedules ScheduleSource
}

BackupExpiry applies the generation plan after a run that worked (backup-restore.md §6, BK-8).

**A run with no schedule behind it deletes nothing**, and that is a decision rather than an omission. The plan lives on a schedule; a backup somebody asked for by hand has none, and inventing a default for it would mean a manual run - the thing people do before something risky - could delete the archives they were making it alongside.

Its failures are logged and not returned. The archive is written and recorded by the time this runs; failing the job would retry the whole backup in order to retry a deletion, which is the wrong operation by several orders of magnitude. The next successful run meets the same archives and tries again.

func (e BackupExpiry) After(ctx context.Context, run domain.Run)

type BackupFreshness ¶

type BackupFreshness interface {
	LastSuccessPerTarget(ctx context.Context) (map[shared.ID]time.Time, error)
}

BackupFreshness answers when each target last had a backup that worked.

type BackupRestore ¶

type BackupRestore struct {
	Applier service.Applier
	// Progress is how the restore says how far along it is. Optional: losing a reading is worth a
	// log line rather than a restore.
	Progress queue.Reporter
}

BackupRestore is the queue's way into reading an archive back (backup-restore.md §8.3).

Detached, and it has to be. The restore streams an archive from somebody else's machine and writes it in batches, each of which is its own transaction - which is what §8.3 step 5 means by "rollback within a transaction per batch size". Doing that inside the runner's own transaction would mean the runner's staying open for the length of the restore, on the connection pool the API shares.

What it gives up is the atomicity of its own completion, which it can afford because the run is safe to repeat: the run row is claimed by identity, and the progress of each batch is recorded in the transaction that wrote it, so the attempt that takes over continues where the last one stopped rather than starting a second restore (BK-7).

func (h BackupRestore) OwnsItsTransactions()

func (h BackupRestore) Release(ctx context.Context, job queue.Job)

func (h BackupRestore) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type BackupRun ¶

type BackupRun struct {
	Performer service.Performer
	// Progress is how the run says how far along it is. Optional: losing a reading is worth a log
	// line rather than a backup.
	Progress queue.Reporter
	// Expiry applies the generation plan after a run that worked. It is here rather than inside
	// the run because §6 makes the ordering a rule: a failed run deletes nothing, so the plan is
	// something that happens *after* a success rather than part of one.
	Expiry BackupExpiry
}

BackupRun is the queue's way into writing an archive (backup-restore.md §5).

Detached, and it has to be. The run holds a REPEATABLE READ snapshot while it streams to somebody else's machine, and doing that inside the runner's own transaction would mean two open at once - the runner's for minutes, on the connection pool the API shares. What it gives up is the atomicity of its own completion, which it can afford because the run is safe to repeat: the run row is claimed by identity, so the attempt that takes over after a worker died continues its own run rather than starting a second one (BK-7).

func (h BackupRun) OwnsItsTransactions()

func (h BackupRun) Release(ctx context.Context, job queue.Job)

func (h BackupRun) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type BackupScheduling ¶

type BackupScheduling struct {
	Pass service.SchedulePass
	// Fallback is when to come back if the tenant owes nothing at all. It is not a poll: the pass
	// reschedules itself to the moment it actually owes, and this is only the answer to "there is
	// nothing on the books", where finishing would be correct and coming back once a day is
	// cheaper than being wrong about a schedule that was written by another process.
	Fallback time.Duration
}

BackupScheduling is one tenant's wake-up: what does this tenant owe now, and when is the next moment it owes anything (backup-restore.md §5).

The same shape as the reminders' and the recurrence materialisation's, and for the same reason: nothing in this system may enumerate tenants, so a scheduler cannot create one job per tenant even if it wanted to (multi-tenancy.md §2.1). The write that creates a schedule seeds it, each round reschedules itself to the next moment the tenant owes, and a tenant that owes nothing lets its poller finish - the next write re-seeds it. A quiet tenant costs nothing.

It is *not* Detached. Everything it does is a write - a job enqueued and a schedule moved on - and those belong in the runner's transaction with the job's own completion, so that a process that dies half way leaves neither an enqueued run nor a schedule that has been advanced past a moment nothing acted on.

func (h BackupScheduling) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type BackupVerify ¶

type BackupVerify struct {
	Performer service.Performer
}

BackupVerify is the queue's way into checking an archive at its target.

Detached for the same reason and with less at stake: it reads every member of an archive over somebody else's network, which is minutes, and it changes one row at the end.

func (h BackupVerify) OwnsItsTransactions()

func (h BackupVerify) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type EnvironmentOperators ¶

type EnvironmentOperators interface {
	Apply(ctx context.Context) ([]env.OperatorEntry, error)
}

EnvironmentOperators is the operator list the installation's environment names.

type FinishedJobs ¶

type FinishedJobs interface {
	PurgeFinished(ctx context.Context, before time.Time, batch int) (map[string]int, error)
}

FinishedJobs removes the jobs whose catalogue period has run out. The queue's own table, read and written in the installation's scope: it has no tenant to scope by (multi-tenancy.md §2.1).

type Import ¶

type Import struct {
	Runner service.Runner
	// Progress records how far the applier got on the job row, for whoever polls it.
	Progress queue.Reporter
}

Import is the queue's way into landing a file somebody exported elsewhere.

Detached, for the reason a restore is: it reads the file back from the object store and applies its records in batches of their own, and doing that inside the runner's transaction would hold one open for as long as the file is long. Safe to repeat, which is what makes it a job: the run records how far the applier got, a resumed attempt continues there, and the identities are derived from the source, so an attempt taking over after a worker died writes what the first one did not rather than a second copy of what it did.

func (h Import) OwnsItsTransactions()

func (h Import) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type InstanceBackups ¶

type InstanceBackups interface {
	Run(ctx context.Context, scope persistence.Scope) (backupservice.PassResult, error)
}

InstanceBackups is the slice of the schedule pass the leader runs: the instance-wide schedules, which belong to no tenant and which nothing else can fire.

type InvitationMessage ¶

type InvitationMessage struct {
	Invitation notification.RecordInvitation
}

InvitationMessage is the queue's way into the invitation queued when the account was created.

Not detached, unlike the delivery beside it, and that is the whole design: this handler writes a record and queues the send, both inside the transaction the runner opened, so a process that dies halfway leaves neither. The reaching-outwards is the next job's problem, which is where the retries and the dead letter already live.

func (h InvitationMessage) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type JobHolder ¶

type JobHolder interface {
	Hold(ctx context.Context, job queue.Job) error
}

JobHolder takes a job's row for the caller's transaction, and fails when the job is no longer the pass's own (queue.Queue.Hold).

type MediaReconciliation ¶

type MediaReconciliation struct {
	Reconciliation media.ReconcileMedia
	// Interval is the wait after a pass that found nothing left to reclaim. It is what a quiet
	// tenant pays for having the machinery at all.
	Interval time.Duration
	// Continuation is the wait after a pass that filled its batch, and is short: there is known
	// work left, and the only reason not to do it now is that a batch is where one pass ends.
	Continuation time.Duration
}

MediaReconciliation is the queue's way into the reclamation of uploaded files: an inbound adapter, like every other handler, that translates a job into a call on the application layer and its answer into the queue's vocabulary.

It reschedules itself rather than being rescheduled, for the reason the retention sweep does: nothing in this system may enumerate tenants, so a scheduler cannot create one job per tenant even if it wanted to. The first job is created by a staging - an upload is the first thing that can ever need reclaiming - and a deletion pulls the next round forward.

The one handler that runs outside the runner's transaction (queue.Detached). It has to: the pass deletes bytes from a bucket between two writes, and a transaction held open across that call is exactly what observability-reliability.md §8 forbids. What it gives up in return is the atomicity of its own completion, which it can afford because the pass is safe to run twice - the argument is written out at media.ReconcileMedia.

func (r MediaReconciliation) OwnsItsTransactions()

func (r MediaReconciliation) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type NotificationDelivery ¶

type NotificationDelivery struct {
	Delivery notification.DeliverNotification
}

NotificationDelivery is the queue's way into sending one notification: an inbound adapter, like every other handler, translating a job into a call on the application layer.

Detached, and it has to be. The delivery reaches an SMTP server between two writes, and a transaction held open across that call is what observability-reliability.md §8 forbids - the same reason the media reconciliation is detached, for the same kind of dependency. What is given up is the atomicity everybody else gets for free: a process that dies between the send and the write-back leaves a message sent and a record that still says pending, and the retry sends it again. That is the right side of the trade for email - a duplicate is a nuisance, a lost one is the thing somebody needed to know.

func (h NotificationDelivery) OwnsItsTransactions()

func (h NotificationDelivery) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PasswordOpeningEnd ¶

type PasswordOpeningEnd struct {
	End admin.EndPasswordOpening
}

PasswordOpeningEnd is the queue's way into recording the end of an operator's opening of the password (ADR-0078 §3). Not detached: it reads and writes one workspace's rows and nothing outside the database, so the runner's transaction in that workspace is the right one. While the opening it finds is still running - a later one replaced the one it was seeded for - it comes back at that opening's end.

func (h PasswordOpeningEnd) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PasswordOpeningMessage ¶

type PasswordOpeningMessage struct {
	Send notification.SendPasswordOpening
}

PasswordOpeningMessage sends one administrator the notice of an opening or of its end.

Detached, PasswordResetMessage's reasoning: it reaches an SMTP server, and a transaction held open across somebody else's machine is what observability-reliability.md §8 forbids.

func (PasswordOpeningMessage) OwnsItsTransactions()

func (h PasswordOpeningMessage) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PasswordResetMessage ¶

type PasswordResetMessage struct {
	Reset notification.SendPasswordReset
}

PasswordResetMessage is the queue's way into sending a reset link (ADR-0068 §6): an inbound adapter, like every other handler, translating a job into a call on the application layer.

Detached, NotificationDelivery's reasoning verbatim: it mints a credential, then reaches an SMTP server, and a transaction held open across somebody else's machine is what observability-reliability.md §8 forbids. What is given up is the same atomicity and the trade is the same one - a retry that mints a second token invalidates nothing, because the newest token is the one the mail that arrived carries.

func (h PasswordResetMessage) OwnsItsTransactions()

func (h PasswordResetMessage) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PasswordResetRequest ¶

type PasswordResetRequest struct {
	Record identity.RecordResetRequest
}

PasswordResetRequest is the queue's way into a forgotten-password request: the job every request leaves behind, whether or not its address held an account (T-02).

Not detached, InvitationMessage's reasoning: it records the request and queues the mail inside the transaction the runner opened, so a process that dies halfway leaves neither. Reaching the mail server is the next job's.

func (h PasswordResetRequest) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PlatformDeadLetters ¶

type PlatformDeadLetters interface {
	DeadLetters(ctx context.Context) (int64, error)
}

PlatformDeadLetters answers how many deliveries to a purchase platform wait to be redelivered, read from the installation's own table under the system scope.

type PlatformSweep ¶

type PlatformSweep struct {
	Sweep adminservice.SweepPlatformEvents
}

PlatformSweep runs the installation's sweep of what it keeps for a purchase platform, and comes back. Not detached: the runner opens the system scope's transaction the sweep runs in.

func (h PlatformSweep) Run(ctx context.Context, _ queue.Job) (queue.Result, error)

type PrivacyDeadlines ¶

type PrivacyDeadlines struct {
	Watch service.WatchDeadlines
	// Queue is here for one statement: the row lock on this job, taken before anything is read.
	// A case recorded between this pass's read and its reschedule would otherwise find the row
	// RUNNING, where Enqueue cannot pull a wake-up forward, and its deadline would wait for a
	// wake-up nobody scheduled.
	Queue queue.Queue
	Clock clock.Clock
}

PrivacyDeadlines watches one tenant's statutory deadlines (alert A-19).

The reminder poller's shape, and for the reason that one has it: nothing in this system may enumerate tenants, so a scheduler cannot create one of these per tenant. The write that records a case seeds it; each pass reschedules itself while the tenant has an open case; a tenant that owes nothing finishes, and its next case brings the job back.

func (d PrivacyDeadlines) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PrivacyErasureRemainder ¶

type PrivacyErasureRemainder struct {
	Resume service.ResumeErasure
}

PrivacyErasureRemainder carries out the rest of an erasure a legal hold kept part of, once the hold is lifted (data-protection.md §4.1).

Detached, for the erasure's own reason: it serves every storage location in the data catalogue, and the bytes go outside any transaction. Safe to repeat: it runs the same erasure against the holds in force, and what is already gone is not there to remove twice.

func (h PrivacyErasureRemainder) OwnsItsTransactions()

func (h PrivacyErasureRemainder) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PrivacyExtensionEntry ¶

type PrivacyExtensionEntry struct {
	Record service.RecordExtensionEntry
}

PrivacyExtensionEntry records an installation-wide case's extension in one workspace the person is a member of (data-protection.md §4.1).

Not detached: the entry and the job's completion are one transaction in that workspace's scope, so a run that fails after the entry is rolled back whole and its retry stores exactly one.

func (h PrivacyExtensionEntry) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type PrivacyRequest ¶

type PrivacyRequest struct {
	Performer service.Performer
}

PrivacyRequest carries out a data subject request that has been started (data-protection.md §4): the archive an access or portability case produces, or the erasure an erasure case is.

Detached, for the reason the backup run is: an erasure serves every storage location in the data catalogue and an export streams a person's whole presence to somebody else's machine, and doing either inside the runner's own transaction would hold one open for minutes on the pool the API shares.

Safe to repeat, which is what makes it a job. Each step of an erasure is idempotent on its own terms - a credential removed twice is removed once, a pseudonym is recorded once - and an export writes the same archive under a name derived from the case. A case that has already been answered is left alone.

func (h PrivacyRequest) OwnsItsTransactions()

func (h PrivacyRequest) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type RecurrenceMaterialisation ¶

type RecurrenceMaterialisation struct {
	Materialisation work.MaterializeOccurrences
	Queue           queue.Queue
	Clock           clock.Clock
	// Continuation is the wait after a pass whose batch filled: there is known work left, and the
	// only reason not to do it now is that a batch is where one transaction ends.
	Continuation time.Duration
	// MinimumWait keeps a wake-up that is due now from becoming a spin.
	MinimumWait time.Duration
}

RecurrenceMaterialisation is the queue's way into what a series owes: an inbound adapter like every other handler, translating a job into a call on the application layer and its answer into the queue's vocabulary.

It holds its own job row for the pass, for the reason the reminder's firing does: a pass that decides its own next wake-up must not lose a write that arrives while it is running (queue.Queue.Hold).

func (m RecurrenceMaterialisation) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type ReminderFiring ¶

type ReminderFiring struct {
	Firing work.FireReminders
	// Queue is here for one statement: the row lock on this job, taken before anything is read.
	// Without it a write committing between the pass's last read and its reschedule would find
	// the row RUNNING, where Enqueue cannot pull a wake-up forward, and its reminder would wait
	// for a wake-up nobody scheduled (queue.Queue.Hold).
	Queue queue.Queue
	Clock clock.Clock
	// Continuation is the wait after a pass that filled its batch: there is known work left, and
	// the only reason not to do it now is that a batch is where one transaction ends.
	Continuation time.Duration
	// MinimumWait keeps a wake-up that is due now from becoming a spin: a reminder whose moment
	// has just passed is fired by this pass, and the next round is never closer than this.
	MinimumWait time.Duration
}

ReminderFiring is the queue's way into the first duty this system has that is caused by a stored future timestamp rather than by something somebody just did.

An inbound adapter like every other handler: it translates a job into a call on the application layer and the answer into the queue's vocabulary. What is particular to it is the shape of the answer - it reschedules itself to the moment the tenant next owes something and finishes when it owes nothing, rather than polling forever like the retention sweep. A quiet tenant therefore costs one row that is not there, and the next write re-seeds it.

func (f ReminderFiring) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type RetentionSweep ¶

type RetentionSweep struct {
	Retention lifecycle.RunRetention
	// Interval is the wait after a pass that reached the end of the trash. It is what a quiet tenant
	// pays for having the machinery at all.
	Interval time.Duration
	// Continuation is the wait after a pass that filled its batch, and is short: there is known work
	// left, and the only reason not to do it now is that a batch is where one transaction ends.
	Continuation time.Duration
}

RetentionSweep is the queue's way into the retention run: an inbound adapter, like every other handler, that translates a job into a call on the application layer and its answer into the queue's vocabulary.

It reschedules itself rather than being rescheduled. One row per tenant that comes back forever is what a poller looks like once it is a queue (queue.Result.Repeat), and it is the only shape available here: nothing in this system may enumerate tenants, so a scheduler cannot create one job per tenant even if it wanted to. The first job is created by a deletion, which is also the more honest statement of what has to happen - something was deleted, so its cleanup is now due.

func (s RetentionSweep) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type RetryBacklog ¶

type RetryBacklog interface {
	ScheduledBacklog(ctx context.Context, kind queue.Kind) (int, error)
}

RetryBacklog answers how many jobs of one kind wait for a future moment. One series reads it today: a webhook retry is exactly one scheduled job - one attempt, one row - so the queue is where hubtask_webhook_retry_backlog is honestly counted, and no tenant-bounded table is crossed.

type Runner ¶

type Runner struct {
	Queue      queue.Queue
	UnitOfWork persistence.UnitOfWork
	// Handlers is what this process can run. A kind with no handler here is not an error at
	// claim time: during a rolling update the old pods have not learned the new kinds yet, and
	// the job simply waits for a pod that has.
	Handlers map[queue.Kind]queue.Handler
	Clock    clock.Clock
	Signals  Signals

	// Batch is how many jobs one round claims.
	Batch int
	// PollInterval is the wait after a round that found nothing. A full batch is followed by the
	// next round at once - a backlog is not a reason to wait.
	PollInterval time.Duration
	// JobTimeout bounds one job. It has to be shorter than Lease, and the constructor refuses a
	// configuration where it is not: a job still running when its lease expires is a job two
	// workers are doing.
	JobTimeout time.Duration
	// Lease is how long a claim holds. Long enough to outlast a job, short enough that a killed
	// worker's jobs do not sit still for minutes.
	Lease time.Duration
	// NextAttempt is the backoff between attempts, given the number of attempts made so far. It
	// is injected rather than computed here because the policy belongs to the resilience adapter
	// and this layer does not know it (observability-reliability.md §6).
	NextAttempt func(attempt int) time.Duration
	// Observe wraps one job run in its trace span. Injected as a function for the same reason as
	// NextAttempt: this layer knows no adapter, and a tracer is one. Nil leaves the run
	// untraced - which is what a test wants and what a process without tracing configured gets
	// anyway, since the no-op tracer costs nothing.
	Observe func(ctx context.Context, kind string, fn func(context.Context) error) error
	// Woken is the doorbell an enqueue rings, so that a job does not wait out the poll interval
	// before anybody looks at it (ADR-0007: "an adaptive polling interval plus LISTEN/NOTIFY as a
	// wake-up"). A channel rather than a listener type, because this layer knows no adapter.
	//
	// Nil is a supported configuration and not a degraded one in any way that matters: the loop
	// then waits out its interval exactly as it always did. That is deliberate - the doorbell is
	// latency, never delivery, and a runner that stopped working when the notification channel
	// did would have turned an optimisation into a dependency.
	Woken <-chan struct{}
	// Diagnose names what a failure's cause was, in attributes safe for a log line: a database
	// error's SQLSTATE and the constraint it broke, never its message, which quotes the row.
	// Injected, because the cause is an adapter's error type and this layer knows no adapter;
	// nil logs the code alone, which is what the rule-10-safe line always carried. What it buys
	// is the difference between "postgres.query_failed" and knowing which statement.
	Diagnose func(error) []slog.Attr
	// SchemaBehind reports whether the database lacks a migration this build embeds. While it does,
	// nothing is claimed: a job run against the older schema may need a column that is not there
	// yet, and the readiness probe already keeps the API from serving (deployment.md §5). The jobs
	// wait for a process whose schema is complete - this one, once its migration has run. Nil
	// claims as before.
	SchemaBehind func(ctx context.Context) bool
}

Runner claims jobs and runs them until its context ends.

func (r Runner) Run(ctx context.Context)

type ScheduleSource ¶

type ScheduleSource interface {
	Plan(ctx context.Context, tenantID, scheduleID shared.ID) (domain.Retention, string, error)
}

ScheduleSource is the slice of the schedules an expiry pass needs: the plan a run was made under.

type Scheduler ¶

type Scheduler struct {
	Leadership queue.Leadership
	Queue      queue.Queue
	UnitOfWork persistence.UnitOfWork
	Clock      clock.Clock
	Signals    SchedulerSignals
	// PlatformDeliveries counts the dead letters waiting for a purchase platform. Nil samples
	// nothing: an installation without the feed has none.
	PlatformDeliveries PlatformDeadLetters
	// Kinds are the job kinds this installation knows. They are published at zero when the queue
	// holds none of them, because a gauge that has never been written has no series - and an
	// alert on a backlog that never appears is an alert that reads "no data" and is believed
	// (observability-reliability.md §4, alert A-06). The same reasoning seeds the panic counter.
	Kinds []queue.Kind

	// Backlog is where the webhook retry ladder is counted. Optional, like everything
	// below: a build without it publishes no backlog series rather than a wrong one.
	Backlog RetryBacklog

	// InstanceBackups fires the schedules that belong to no tenant. Optional: an installation
	// without it simply never has one, which is the state every installation is in today.
	InstanceBackups InstanceBackups
	// BackupFreshness is the reading behind alert A-12.
	BackupFreshness BackupFreshness
	// AuditPartitions makes sure next month's partition of `audit_log` exists before the first
	// entry of it does, and carries its own policy and its own revokes. Optional, like the two
	// above: an installation without it keeps writing into the default partition, which has both.
	AuditPartitions AuditPartitions
	// StreamPartitions is the same duty for the four monthly streams: activity
	// entries, outbox events, rule runs and the change log. Nil skips, ensureAuditPartitions'
	// contract.
	StreamPartitions streams.Partitions
	// OfflineWindow is the installation's offline window (RetentionConfig.TombstoneWindow), the
	// floor of the change log's drop: a month of it falls only once every device that was offline
	// could have pulled it (offline-sync.md §7). Zero leaves the catalogue's default as the
	// floor.
	OfflineWindow time.Duration
	// StreamEvidence records a dropped partition where a per-tenant trail cannot: a partition
	// holds every tenant's rows, so the evidence is the instance journal's (audit.md §6). Nil
	// records nothing.
	StreamEvidence StreamEvidence

	// FinishedJobs removes finished jobs past FinishedJobPeriod, on the first tick of every hour.
	// Nil skips.
	FinishedJobs FinishedJobs
	// Usage removes the use counts older than UsageMonths, on the first tick of every day. Nil
	// skips.
	Usage UsageSweep
	// AuditTrail lets the trail's months past their period go, on the first tick of every hour,
	// and queues the entry each workspace's own trail gets. Nil skips.
	AuditTrail auditrepo.TrailRetention

	// Operators applies HUBTASK_OPERATORS, on the first tick of every minute: an entry whose account
	// did not exist at start is put into the register once it does (UC-INS-02 check 4). Nil skips.
	Operators EnvironmentOperators

	// SchemaBehind is the runner's: while the database lacks a migration this build embeds, this
	// process neither takes leadership nor keeps it, so the duties fall to an instance whose
	// schema is complete. Nil acts as before.
	SchemaBehind func(ctx context.Context) bool

	// TickInterval is how often the leader looks at the clock. It is also how quickly a standby
	// notices that the leader is gone, because a standby tries the lock on every tick of its own.
	TickInterval time.Duration
}

Scheduler is the role that may run in several replicas but act in only one (ADR-0008).

It does no work itself beyond deciding what has to happen: everything it decides becomes a job, which is what lets the workers scale while the deciding stays single. That split is also what keeps the tick short - a leader holding a lock for a long-running task is a leader nobody can replace while it holds it.

In 0.1.0 the tick has one duty, and the rest of the schedule - reminders, recurrence, retention - registers here as those arrive. The duty it has is not a placeholder: the queue depth is the number alert A-06 watches, and it is a property of the installation, so exactly one process should be reporting it.

func (s Scheduler) Run(ctx context.Context)

type SchedulerSignals ¶

type SchedulerSignals interface {
	QueueDepth(ctx context.Context, kind string, pending int64)
	SchedulerTickLag(ctx context.Context, seconds float64)
	// BackupLastSuccess is when a target last had a backup that worked - alert A-12's number
	// (observability-reliability.md §10).
	BackupLastSuccess(ctx context.Context, targetID string, at time.Time)
	// WebhookRetryBacklog is the deliveries waiting out their backoff
	// (observability-reliability.md §4).
	WebhookRetryBacklog(ctx context.Context, waiting int64)
	// PlatformDeadLetters is the installation's deliveries to a purchase platform waiting to be
	// redelivered - alert A-21's number (UC-INS-14).
	PlatformDeadLetters(ctx context.Context, waiting int64)
	// JobsPurged counts the finished jobs the leader removed, by terminal state - never a kind's
	// payload or a tenant (rule 10).
	JobsPurged(ctx context.Context, state string, count int64)
	// AuditPartitionsHeld is how many of the trail's months past their period a legal hold keeps
	// whole - for every workspace in them (data-retention.md §3).
	AuditPartitionsHeld(ctx context.Context, months int64)
	// AuditDefaultAgedRows is how many entries in the trail's default partition are past their
	// period: rows no partition drop can reach, which an operator moves out (data-retention.md §3).
	AuditDefaultAgedRows(ctx context.Context, rows int64)
}

SchedulerSignals is the slice of the metrics adapter the scheduler uses.

Both of these are measurements of the installation rather than of a process, which is why the leader takes them: every replica reporting the same queue depth would leave a dashboard summing one number over N instances and calling it a backlog.

type SearchReindex ¶

type SearchReindex struct {
	Rebuild  work.RebuildSearchIndex
	Progress queue.Reporter
	// Continuation is the wait between two batches. Short: there is known work left, and the
	// only reason not to do it now is that a batch is where one transaction ends.
	Continuation time.Duration
}

SearchReindex rewrites one workspace's stale search documents, batch by batch.

The embedding pass's shape without the provider: a batch in its own transaction, then a decision about whether to come straight back - and unlike the embedding pass it *finishes*, because what it does is a backlog with an end. Detached, so that each batch commits on its own and a walk over a large workspace holds no row longer than one statement.

Progress is what a client polls for: the rows still stale over the rows that were stale when the administrator asked, which the job's payload carries.

func (h SearchReindex) OwnsItsTransactions()

func (h SearchReindex) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type SecretReseal ¶

type SecretReseal struct {
	Round        sealing.RunReseal
	Installation sealing.RunInstallationReseal
}

SecretReseal is the queue's way into one workspace's re-sealing round (ADR-0045): the job the operator's request seeded, translated into one call on the application layer inside the transaction the runner opened for the tenant. Not detached - every write is a row of the tenant's, and a round that fails halfway rolls back whole and runs again.

A job without a tenant is the installation's own round (UC-INS-16 check 2): the runner opened it under the scope with none, and it moves the secrets that belong to no workspace.

func (h SecretReseal) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type Signals ¶

type Signals interface {
	JobFinished(ctx context.Context, kind string, seconds float64)
	JobFailed(ctx context.Context, kind, attemptClass string)
	JobDeadLettered(ctx context.Context, kind string)
}

Signals is the slice of the metrics adapter this package uses. An interface rather than the adapter, so the presentation layer keeps pointing inwards (project-structure.md §2). Nil is allowed - a test that only cares about what happened to a job says so by leaving it out.

type StreamEvidence ¶

type StreamEvidence interface {
	PartitionDropped(ctx context.Context, table, partition string, rows int64) error
}

StreamEvidence is the instance journal's slice the drop duty writes through.

type TenantExport ¶

type TenantExport struct {
	Archivist admin.TenantExportArchivist
	// Told tells a purchase platform the export is written (UC-INS-14).
	Told admin.TellExportFinished
}

TenantExport is the queue's way into the workspace export: an inbound adapter that translates the job the control plane seeded into one call on the archivist.

Detached, the audit export's reason: the pass streams an archive to a target between reads, and a transaction held open across that is what observability-reliability.md §8 forbids. The archivist opens its own snapshot transaction for the read and nothing else.

func (TenantExport) OwnsItsTransactions()

func (h TenantExport) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type TenantHardDelete ¶

type TenantHardDelete struct {
	Deletion admin.HardDeleteTenant
}

TenantHardDelete is the grace job's way into multi-tenancy.md §5's final act: seeded by the deletion request's own write, due when the 30-day deadline the row carries has passed, and translated here into one call on the application layer.

Detached, for MediaReconciliation's reason: the pass deletes bytes from a bucket between two writes, and a transaction held open across that call is exactly what observability-reliability.md §8 forbids. What it gives up in return it can afford - the pass is safe to run twice, and the argument is written out at admin.HardDeleteTenant.

func (TenantHardDelete) OwnsItsTransactions()

func (h TenantHardDelete) Run(ctx context.Context, job queue.Job) (queue.Result, error)

type UsageSweep ¶

type UsageSweep interface {
	SweepUsage(ctx context.Context, before time.Time) (int64, error)
}

UsageSweep removes the use counts older than the day it is given, inside workspaces and in the installation's ledger (multi-tenancy.md §4.3).

type WayInChangedMessage ¶

type WayInChangedMessage struct {
	Send notification.SendWayInChanged
}

WayInChangedMessage tells a person that a way into their account changed (ADR-0078 §6).

Detached, PasswordResetMessage's reasoning: it reaches an SMTP server, and a transaction held open across somebody else's machine is what observability-reliability.md §8 forbids.

func (WayInChangedMessage) OwnsItsTransactions()

func (h WayInChangedMessage) Run(ctx context.Context, job queue.Job) (queue.Result, error)