queue

package
v0.9.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 27, 2026 License: Apache-2.0 Imports: 3 Imported by: 0

Documentation

Overview

Package queue is the port for work that happens outside a request: the job queue, the handlers that run on it, and the leader that exists exactly once (ADR-0008).

The queue is a table in PostgreSQL rather than a broker, so that a self-hosted installation needs no second piece of infrastructure and a job can be created in the same transaction as the change that asks for it. The port is what keeps that a decision rather than a fact: a broker adapter later implements the same three interfaces.

Two guarantees run through everything here, and the rest of the design follows from them:

  • At-least-once. A process may die at any moment (observability-reliability.md §1), so a job that was picked up but not finished is picked up again once its lease expires. A handler that writes its effect in the transaction the runner opened therefore takes effect exactly once, because the effect and the job's completion commit together (test RT-3).
  • The queue is not a place to keep state. A job carries what it needs to run, and everything else is read from the database when it runs - a payload that has been waiting an hour is a snapshot of an hour ago.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Depth 

type Depth struct {
	Kind Kind
	// Pending counts what is waiting or overdue, not what is running: a job in flight is work
	// being done, and counting it would make a busy queue look like a stuck one.
	Pending int
}

Depth is the backlog of one kind, for the gauge an alert watches when processing stops keeping up (observability-reliability.md §10, alert A-06).

type Detached 

type Detached interface {
	Handler

	// OwnsItsTransactions is the assertion, and its value is never read: implementing the
	// interface is the declaration. A method rather than an empty interface, so that a handler
	// cannot satisfy it by accident.
	OwnsItsTransactions()
}

Detached is a handler the runner does not wrap in a transaction.

The narrow exception to the paragraph above, for the one kind of job that cannot live inside it: a pass that has to reach a bucket between two writes. Holding the transaction open across that call is exactly what observability-reliability.md §8 forbids, and doing the call after the commit is not available to a handler - the runner owns the boundary.

What an implementer gives up is the atomicity everybody else gets for free: its writes and the job's completion no longer commit together, so a process that dies between them leaves work done and a job that will be claimed again. That is only acceptable for a pass that is safe to run twice, and an implementer of this interface is asserting exactly that.

type Failure 

type Failure struct {
	Job Job
	// Code is why, machine readable: a detail code, never a sentence (rule 8) and never anything
	// the failing operation was working on (rule 10). It is what an operator sees on a dead
	// letter, so it has to be enough to act on and nothing more.
	Code string
	// RetryAt is when the next attempt is due. Zero sends the job to the dead letter instead:
	// the attempts are used up, or the error is one that will not read differently next time.
	RetryAt time.Time
	// Refused ends the job as FAILED rather than in the dead letter, at once: the job met a
	// refusal (Result.Refusal), which is an answer for whoever asked and nothing an operator has
	// to act on. RetryAt is ignored.
	Refused bool
	// Params are the refusal's parameters, kept beside Code so that the sentence a person reads
	// can say what the code alone cannot - when a spent budget renews. Codes and moments only,
	// never anything the job was working on (rule 10). Written only with Refused.
	Params map[string]string
}

Failure is a job that did not work out.

type Handler 

type Handler interface {
	Run(ctx context.Context, job Job) (Result, error)
}

Handler runs one kind of job.

It runs inside the transaction the runner opened, which is the whole reliability argument: what the handler writes and the job's own completion commit together, so a process that dies halfway leaves neither behind. A handler that needs to reach outside the database - a webhook, an email - does so through a job of its own rather than from in here, because an external call inside a transaction holds a database connection for as long as somebody else's server feels like taking (observability-reliability.md §8).

type Job 

type Job struct {
	ID       shared.ID
	TenantID shared.ID
	Kind     Kind
	Payload  map[string]any
	// Attempts counts this attempt, so the first run of a job reports 1. A handler that behaves
	// differently on a retry - a gentler timeout, a smaller batch - reads it here.
	Attempts    int
	MaxAttempts int
	// Lease is when this claim expires, and it is also the token that proves the claim. Every
	// statement that ends the attempt names it, so a worker that fell so far behind that somebody
	// else took the job over changes nothing and rolls back instead of applying its work twice.
	//
	// It is on the job rather than kept by the runner because the fence is only worth anything if
	// it cannot be forgotten: there is no way to call Complete without it.
	Lease time.Time
}

Job is one claimed unit of work, as the runner hands it to a handler.

func (Job) LastAttempt 

func (j Job) LastAttempt() bool

LastAttempt reports whether a failure now is final. The runner asks it to decide between a retry and the dead letter; a handler may ask it to decide how loudly to complain.

type Kind 

type Kind string

Kind names what a job does. It is the label of every job metric, so the set stays small and written by hand - a kind assembled from data would be an unbounded label (observability-reliability.md §3.2). 

const (
	// KindOutboxDispatch delivers one tenant's pending outbox events (ADR-0007). One job per
	// tenant, which is what "system jobs loop per tenant rather than running globally"
	// (multi-tenancy.md §2.1) looks like once it is a queue.
	KindOutboxDispatch Kind = "outbox.dispatch"

	// KindInvitationEmail tells somebody they have been invited. Queued in the transaction that
	// created the account, so the message and the seat exist together or neither does - and
	// delivered by a worker, so an unreachable mail server never fails the request that invited
	// them (observability-reliability.md §7).
	KindInvitationEmail Kind = "notification.invitation"

	// KindPasswordResetEmail carries a reset link to somebody who asked for one (ADR-0068 §6).
	//
	// A job rather than a send inside the request, for KindInvitationEmail's reason: an
	// unreachable mail server must never fail the request, and the request's answer is the same
	// `202` whether or not the address holds an account - a send that failed loudly would be the
	// difference a probe is looking for.
	//
	// And deliberately *not* a notification record. Every other message in this system is
	// something a person may switch off, has a read state, and belongs in a list; a reset link is
	// a credential on its way to a mailbox, with no preference to consult and nothing to list.
	// Giving it a category would add one that no preference may touch - which is a category that
	// exists only to be an exception.
	KindPasswordResetEmail Kind = "notification.password_reset"

	// KindPasswordResetRequest is what a forgotten-password request leaves behind, for every
	// address alike (T-02): the account the address named, or nobody. Its handler records the
	// request and queues KindPasswordResetEmail where there is an active account - work the request
	// itself used to do for a held address only, and whose time was the difference a probe could
	// read. Not detached: the entry and the mail job commit with the job's completion. Not
	// deduplicated: an empty account must not collapse into somebody else's request, and the mail
	// job is deduplicated per account already.
	KindPasswordResetRequest Kind = "identity.password_reset_request"

	// KindPasswordOpeningEmail tells one administrator of a workspace that an operator opened the
	// password there, or that the opening ended (ADR-0078 §3, identity.md §17.2). Queued in the
	// transaction that opened or closed it, one job per administrator, so the act and its notices
	// commit together. Not a notification record, for KindPasswordResetEmail's reason: it is the
	// one message about the way into a workspace that no preference may silence, and a category
	// that existed only to be an exception would be the wrong shape for it.
	KindPasswordOpeningEmail Kind = "notification.password_opening"

	// KindWayInChangedEmail tells a person that a way into their account changed - a password
	// changed or set through a reset, a provider connected or disconnected (ADR-0078 §6,
	// identity.md §6). Queued in the change's own transaction, one job per person, and only where
	// the installation has a mail server; not a notification record, for KindPasswordResetEmail's
	// reason.
	KindWayInChangedEmail Kind = "notification.way_in_changed"

	// KindAddressConfirmationEmail carries the link that confirms a managed account's new address
	// (ADR-0074 §6). Queued in the request's transaction, so the waiting address and its mail exist
	// together; not a notification record, for KindPasswordResetEmail's reason - a credential on its
	// way to a mailbox, with no preference to consult.
	KindAddressConfirmationEmail Kind = "notification.address_confirmation"

	// KindNotificationDeliver sends one notification. One job per record rather than one per
	// tenant, because the retry belongs to the message: an address the server refuses must not hold
	// up everybody else's mail, and the queue's own attempt budget and dead letter are exactly the
	// retry observability-reliability.md §7 promises for "the reminder arrives late".
	//
	// Queued by the outbox consumer in the dispatcher's transaction, so the record and the job to
	// send it commit together - and detached when it runs, because an SMTP server inside a
	// transaction holds a database connection for as long as somebody else's machine takes (§8).
	KindNotificationDeliver Kind = "notification.deliver"

	// KindEmergencyAccessEmail sends one notification of the category EMERGENCY_ACCESS: a member
	// told that the workspace's owner opened their private hub (ADR-0073 §4). The same delivery
	// as KindNotificationDeliver, under a kind of its own for the rolling update's sake: a process
	// of the previous release would claim a `notification.deliver` job for a category it does not
	// know and send it with the assignment's words, while a kind it has no handler for goes back to
	// the queue for a process that has one (presentation/worker.Runner).
	KindEmergencyAccessEmail Kind = "notification.emergency_access"

	// KindRoleGrantedEmail sends one MEMBERSHIP notification about a role somebody was given
	// (UC-NOT-01 check 9). The same delivery again, under its own kind for the same rolling update:
	// a process of the previous release does not know the record's grant, would read a membership
	// record without a subject as the trashed private hub's, and send that mail instead.
	KindRoleGrantedEmail Kind = "notification.role_granted"

	// RoleGrantedAttempts is that job's attempt budget, EmergencyAccessAttempts' for its reason.
	RoleGrantedAttempts = 20

	// EmergencyAccessAttempts is that job's attempt budget: more than the default, because a
	// rollout's old processes may claim and return it a few times before a new one runs it, and
	// each claim spends an attempt.
	EmergencyAccessAttempts = 20

	// KindReminderFire is the first job in this system that exists because of a stored future
	// timestamp rather than because something just happened.
	//
	// One job per tenant, seeded by the write that made something due - a reminder written, a due
	// date moved - with RunAt at the moment that write brought forward, because nothing may
	// enumerate tenants (multi-tenancy.md §2.1) and a scheduler therefore cannot create one job
	// per tenant even if it wanted to. It reschedules itself to the next moment the tenant owes
	// and completes when it owes none, which is what keeps a quiet tenant costing nothing: the
	// next write re-seeds it.
	KindReminderFire Kind = "reminder.fire"

	// KindMailboxPoll reads one connected mailbox into the workspace's mail intake (ADR-0094).
	//
	// The reminder's shape: one job per mailbox, seeded by the write that connected it, and the
	// pass reschedules itself a minute later for as long as the mailbox is there and switched on -
	// nothing enumerates tenants or mailboxes (multi-tenancy.md §2.1). Its dedupe key is the
	// mailbox, so a second seed while one is pending is the same job.
	KindMailboxPoll Kind = "mail.mailbox_poll"

	// KindRecurrenceMaterialize turns one tenant's series into the entries their rolling windows
	// owe (ADR-0008).
	//
	// The same shape as the reminder's wake-up and for the same reason: nothing may enumerate
	// tenants, so the write that made something owed seeds it - a rule written, an occurrence
	// completed - and the pass reschedules itself to the moment the horizon reaches the next
	// occurrence. A series in ON_COMPLETION owes nothing until somebody completes something, so
	// its pass finishes and the completion seeds the next one.
	KindRecurrenceMaterialize Kind = "recurrence.materialize"

	// KindAiSuggest asks the workspace's AI provider what one thing should become.
	//
	// A job rather than a call inside the request, because ai-first.md §2 forbids an AI call in
	// the critical path and because a provider's latency is somebody else's machine. One job per
	// *thing asked about* rather than per tenant, unlike the reminder's wake-up: what it does is
	// bounded and finite, and a failure belongs to that one question rather than holding up
	// everybody else's.
	//
	// Deliberately not deduplicated. Asking twice produces two proposals, and a duplicate
	// suggestion is something somebody dismisses - which is the whole safety of a suggestion
	// being a record rather than a change.
	KindAiSuggest Kind = "ai.suggest"

	// KindAiDraftRule drafts an automation rule from a person's sentence (UC-AUT-10). The same
	// question as KindAiSuggest, under a kind of its own for the rolling update's sake: a process of
	// the previous release would claim an `ai.suggest` job asking a prompt it does not know and
	// spend the retry ladder into the dead letter, while a kind it has no handler for goes back to
	// the queue for a process that has one (presentation/worker.Runner).
	KindAiDraftRule Kind = "ai.draft_rule"

	// AiDraftRuleAttempts is that job's attempt budget, EmergencyAccessAttempts' for its reason.
	AiDraftRuleAttempts = 20

	// KindAiMakeEntries turns a person's pasted text into a proposal of entries (UC-AI-09), under a
	// kind of its own for KindAiDraftRule's reason: a process of the previous release has no
	// handler for it and hands it back to the queue rather than spending its retry ladder.
	KindAiMakeEntries Kind = "ai.make_entries"

	// AiMakeEntriesAttempts is that job's attempt budget, AiDraftRuleAttempts' for its reason.
	AiMakeEntriesAttempts = 20
	// KindAiAnswer answers a person's question about the workspace (UC-AI-08), under a kind of its
	// own for KindAiDraftRule's reason, and with its attempt budget.
	KindAiAnswer Kind = "ai.answer_question"

	// KindAiSuggestArrival asks for a proposal the workspace prepares about one jumble arrival
	// (UC-JUM-07 check 10): the system acting for the workspace, not a person who asked. A kind of
	// its own rather than a flag on KindAiSuggest, for KindEmergencyAccessEmail's reason: a process
	// of the previous release would claim an `ai.suggest` job it cannot read as an arrival's and
	// refuse it for the person it names none of, while a kind it has no handler for goes back to
	// the queue for a process that has one. One job per arrival, deduplicated on the entry.
	KindAiSuggestArrival Kind = "ai.suggest_arrival"

	// AiSuggestArrivalAttempts is that job's attempt budget, EmergencyAccessAttempts' for its
	// reason: a rollout's old processes may claim and return it a few times first.
	AiSuggestArrivalAttempts = 20

	// KindAiEmbed brings one workspace's embeddings up to date with its entries.
	//
	// One job per tenant that reschedules itself, the retention sweep's shape rather than the
	// suggestion's - because what it does is a backlog rather than a question. An entry whose
	// title or notes changed owes a new vector, and a pass takes a batch of what is owed and comes
	// back for the rest.
	//
	// Seeded by the write that made something owed, like every per-tenant duty in this system:
	// nothing may enumerate tenants (multi-tenancy.md §2.1), so a scheduler cannot create one job
	// per workspace even if it wanted to. Deduplicated per tenant, because "this workspace owes
	// embeddings" is one fact however many entries changed.
	KindAiEmbed Kind = "ai.embed"

	// KindRetentionSweep removes what one tenant's retention periods say may go (ADR-0020).
	//
	// One job per tenant, which reschedules itself forever: a poller lives as one row rather than
	// as a new row per round (Result.Repeat). It is seeded by writes in the tenant rather than by a
	// scheduler enumerating tenants, because nothing in this system may enumerate them - the
	// `tenant` table is behind row level security with no bypass for the application role, and
	// tenant administration runs through the control plane (db/migrations/0001_init.sql). The writes
	// are a deletion, the workspace's provisioning and every session opened (RetentionSweep,
	// data-retention.md §5).
	KindRetentionSweep Kind = "retention.sweep"

	// KindMediaReconcile makes one tenant's media reference counts honest and reclaims what nothing
	// points at (data-protection.md §5).
	//
	// One job per tenant, rescheduling itself forever, seeded by a write in the tenant for the
	// reason the retention sweep is: nothing in this system may enumerate tenants, so a scheduler
	// cannot create one job per tenant even if it wanted to. A staging is what seeds it - an
	// upload is the first thing that can ever need reclaiming - and a deletion pulls it forward.
	KindMediaReconcile Kind = "media.reconcile"

	// KindBackupRun writes one archive to one target (backup-restore.md §5).
	//
	// One job per run rather than per tenant, because a run is a thing somebody asked for and can
	// cancel, and because the deduplication key is the target: two requests to back up the same
	// target collapse into the one that is already happening, which is the lock §5 asks for
	// expressed in the queue as well as in the table.
	//
	// It is the first job in this system that is long enough for "how far along is it" to be a
	// real question, which is why the job row grew a `progress` column with it.
	KindBackupRun Kind = "backup.run"

	// KindBackupVerify checks one archive at its target without restoring it.
	//
	// A job rather than a request, because verifying reads every member of an archive over
	// somebody else's network - which is minutes, and nothing a caller should hold a connection
	// open for.
	KindBackupVerify Kind = "backup.verify"

	// KindBackupRestore reads one archive back into a tenant (backup-restore.md §8.3).
	//
	// Detached and long, for the reasons a run is: it 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". Deduplicated on the restore
	// rather than on the tenant, because the tenant's lock lives in the table: two jobs for one
	// restore are the same work, and two restores in one tenant are refused where the caller can
	// read the refusal.
	KindBackupRestore Kind = "backup.restore"

	// KindBackupSchedule is one tenant's wake-up: what does this tenant owe now, and when does it
	// owe the next one (backup-restore.md §5).
	//
	// The same shape as the reminder's and the recurrence'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. The write that creates a schedule seeds it, and each round reschedules itself
	// to the next moment the tenant owes. Instance-wide schedules belong to no tenant and are the
	// leader's duty instead - see the scheduler.
	KindBackupSchedule Kind = "backup.schedule"

	// KindAuditAnchor is one tenant's daily anchoring (audit.md §3): the chain's end written to the
	// backup target the workspace named. One job per tenant, seeded by the configuration's write
	// and rescheduling itself to the next day, for the backup schedule's reason: nothing in this
	// system may enumerate tenants.
	KindAuditAnchor Kind = "audit.anchor"

	// KindAuditTrailRemoved records, in one workspace's own trail, that a month of it went for good
	// (audit.md §3). Queued by the leader in the transaction that dropped the month, one per
	// workspace that had entries in it: the partition holds every workspace, the entry belongs in
	// each one's own chain.
	KindAuditTrailRemoved Kind = "audit.trail_removed"

	// KindAuditExport writes a period of the audit trail to a backup target as a signed archive
	// (audit.md §5).
	//
	// A job rather than a request, and the reason is in the numbers: an export over four hundred
	// days is as large as the tenant was busy, and it is the second operation this system has that
	// cannot be bounded - which is why it reports through `/jobs/{id}`. Deduplicated on the export
	// rather than on the tenant: two jobs for one export are the same work, and two exports of
	// different periods in one tenant are two legitimate questions.
	KindAuditExport Kind = "audit.export"
	// KindImport applies a file somebody exported elsewhere as collections under a hub
	// (backup-restore.md §9): converted into archive records and landed through the restore's
	// applier. One per import, deduplicated by the import's identifier.
	KindImport Kind = "import.run"
	// KindImportIntoCollection is the same job for an import that lands in a collection of the
	// hub. A kind of its own so that a process that predates landing in a collection - and would
	// create collections from the file instead - leaves it for one that does not (rule 12).
	KindImportIntoCollection Kind = "import.into_collection"

	// KindPrivacyRequest carries out a data subject request that has been started: the archive an
	// access or portability case produces, or the erasure an erasure case is (data-protection.md
	// §4).
	//
	// A job because both are as large as the person's presence in the workspace, and because the
	// erasure serves every storage location in the data catalogue - rows, media, the search index,
	// the derived counters - which is minutes of work that no request should hold a connection
	// open for. Deduplicated on the case: two jobs for one case are the same work, and a case is
	// started once.
	KindPrivacyRequest Kind = "privacy.request"

	// KindPrivacyDeadlines watches the statutory deadlines of one tenant's open cases
	// (data-protection.md §4): "without deadline monitoring, the right gets violated in practice
	// even though the feature exists".
	//
	// The shape of the reminder poller, and for the same reason: 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, and
	// a tenant that owes nothing costs a row that is not there.
	KindPrivacyDeadlines Kind = "privacy.deadlines"

	// KindPrivacyExtensionEntry records an installation-wide case's extension in one workspace the
	// person is a member of (data-protection.md §4.1). One job per workspace, enqueued with the
	// extension, so that a failure in one workspace is retried there rather than lost; not
	// detached, so the entry and the job's completion are one transaction in that workspace.
	// Deduplicated on the case and the workspace.
	KindPrivacyExtensionEntry Kind = "privacy.extension_entry"

	// KindPrivacyErasureRemainder carries out the rest of an erasure a legal hold kept part of,
	// once that hold is lifted (data-protection.md §4.1). Seeded in the transaction that lifts the
	// hold, one per case, and by the retention pass for a remainder nothing seeded. Detached, like
	// the erasure it continues. Deduplicated on the case and the hold.
	KindPrivacyErasureRemainder Kind = "privacy.erasure_remainder"

	// KindWebhookDeliver sends one event to one subscription (automation.md §3.1).
	//
	// One job per delivery rather than one per event, and that is the whole retry discipline: a
	// target that is down must not hold up the events of every other subscriber, and eight
	// attempts with a backoff reaching a day is a schedule the queue already knows how to keep.
	// The job carries the subscription and the event; the body is rendered at each attempt from
	// the event as it was, so a retry sends what the first attempt would have.
	KindWebhookDeliver Kind = "webhook.deliver"

	// KindBusPublish puts one event on the optional message bus (ADR-0042).
	//
	// A job and not a step of the dispatch, for the reason this package states about every
	// subscriber: a subscriber runs inside the dispatcher's transaction and must not call the
	// outside world. A bus that is slow, restarting or gone would otherwise hold a database
	// transaction open across the network, which observability-reliability.md §8 forbids outright.
	//
	// One job per event rather than one per subject: there is one bus, so a failure isolates
	// nothing by being split further, and the queue's own ladder is the retry. The job carries the
	// event's identifier and reads the event back at each attempt, exactly as the webhook delivery
	// does - so a retry publishes what the first attempt would have.
	KindBusPublish Kind = "bus.publish"

	// KindAutomationRun is one rule's reaction to one event (automation.md §2).
	//
	// One job per matching rule rather than one per event: failure isolation per rule, the queue's
	// backoff per rule, and a dead letter naming which rule rather than which batch. An event
	// matching six rules that cost one job would make one rule's misconfiguration everybody else's
	// outage.
	//
	// The dedupe key is what collapses a storm. Without a rule's own expression it names the rule
	// and the event, so nothing collapses; with one it names the rule and the expression's value,
	// and the queue's existing uniqueness does the rest.
	KindAutomationRun Kind = "automation.run"

	// KindAutomationHTTP performs one HTTP_REQUEST action's call (automation.md §1.3).
	//
	// A job rather than a call inside the engine's transaction, for the webhook deliverer's
	// reason: an external call from inside one holds a database connection for as long as
	// somebody else's server feels like taking (observability-reliability.md §8). The job
	// carries the request - method, address, sealed secret, body template - and the sender
	// renders the body from the event at each attempt, so a retry sends what the first attempt
	// would have.
	KindAutomationHTTP Kind = "automation.http"

	// KindAutomationSchedule is one tenant's wake-up for its SCHEDULE rules (automation.md §1.1).
	//
	// The same shape as the reminders', the recurrence materialisation's and the backup schedules':
	// one job per tenant, rescheduling itself to the moment the tenant next owes something, seeded
	// by the write that made something owed. Nothing in this system may enumerate tenants
	// (multi-tenancy.md §2.1), so a scheduler cannot create one job per tenant even if it wanted
	// to - and a tenant with no scheduled rule has no row at all.
	KindAutomationSchedule Kind = "automation.schedule"

	// KindAutomationCheck is one workspace's check of its rules against what exists now
	// (ADR-0060): seeded by the deletion of anything a rule may name, one job per tenant however
	// many deletions arrive together, and never by a timer - the rules screen asks for the same
	// check on demand, which is what makes "after an update" true without enumerating tenants.
	KindAutomationCheck Kind = "automation.check"

	// KindTenantExport writes one workspace's complete, documented archive to a configured backup
	// target (tenant-export.md): the control plane's act, working for every lifecycle state - the
	// suspended and the leaving are exactly who needs it. One job per request; the §4 concurrency
	// quota is enforced where the job is created.
	KindTenantExport Kind = "tenant.export"

	// KindTenantHardDelete carries out multi-tenancy.md §5's final act after the 30-day grace: the
	// media bytes store-first, the tables the cascade cannot reach, the audit trail through the one
	// narrow purge, and the tenant row whose cascade takes the rest - each store counted into the
	// evidence entry that outlives them all.
	//
	// Seeded by the deletion request's own write (multi-tenancy.md §2.1: nothing enumerates
	// tenants, so there is no sweeper to find pending deletions - the request that created the debt
	// creates the job), with RunAt at the grace's end. The handler re-reads the two facts the grace
	// could have changed and deletes nothing if either moved.
	KindTenantHardDelete Kind = "tenant.hard_delete"

	// KindPasswordOpeningEnd records the end of an operator's opening of the password once its time
	// has passed (ADR-0078 §3, identity.md §17.2): the trail entry, the journal entry and the
	// notices. The password itself closed at that moment whether or not this has run - the end is
	// honoured where the opening is read - so the job is bookkeeping, not the end.
	//
	// Seeded by the opening's own write, with RunAt at its end and the tenant as the dedupe key
	// (per-tenant work is seeded by a write in that tenant, never by a sweep). An opening replaced
	// by a later one moves the pending job no later; the handler re-reads the row and comes back
	// when the opening it finds is still running.
	KindPasswordOpeningEnd Kind = "tenant.password_opening_end"

	// KindSecretReseal moves one workspace's sealed values under the current master key
	// (ADR-0045, security.md §8.1): the half of a rotation that lets a key leave the ring. One
	// job per tenant, enqueued by the operator's request through the control plane's one
	// legitimate enumerator - nothing else may enumerate tenants (multi-tenancy.md §2.1), and this
	// is an operator asking rather than a schedule. Deduplicated per tenant, so asking twice before
	// the first round has run queues nothing new.
	KindSecretReseal Kind = "secret.reseal"

	// KindSearchReindex rewrites the search documents of one workspace that were built under a
	// configuration the installation has since replaced or gained (ADR-0034): the rows whose
	// recorded configuration differs from what the resolver answers today, in batches, in the job's
	// own transactions. One job per tenant, enqueued by a workspace administrator's request - a
	// workspace's operation, because nothing may enumerate tenants - and deduplicated per tenant,
	// so asking twice before the walk has finished queues nothing new.
	KindSearchReindex Kind = "search.reindex"

	// KindPlatformSweep removes what the installation keeps for a purchase platform past its 30
	// days (UC-INS-14): its events and their deliveries. One installation job, without a tenant, seeded when a worker
	// starts and repeating; the dedupe key keeps it one however many workers start.
	KindPlatformSweep Kind = "platform.sweep"

	// KindPlatformDeliver sends one installation event to one platform subscription (UC-INS-14):
	// an installation job without a tenant, queued in the event's own transaction and again for
	// each retry and redelivery. A workspace's purge removes its jobs, never these.
	KindPlatformDeliver Kind = "platform.deliver"
)

func (Kind) String 

func (k Kind) String() string

type Leadership 

type Leadership interface {
	// Acquire tries to become the leader and reports whether it worked. A follower calls it again
	// on its next tick; being turned down is the normal state of a standby, not an error.
	Acquire(ctx context.Context) (bool, error)

	// Confirm reports whether leadership is still held. A leader asks before every tick: the
	// connection carrying the lock can be cut without anybody noticing, and a former leader that
	// keeps working is exactly the double execution the lock exists to prevent.
	Confirm(ctx context.Context) (bool, error)

	// Release gives leadership up. A graceful shutdown calls it so that the standby can take over
	// immediately rather than after the connection times out.
	Release(ctx context.Context) error
}

Leadership is the "exactly one" of ADR-0008: the scheduler role may run in several replicas so that one can fail, but only one of them may act, or every reminder fires twice.

It is not a lease in a table but a PostgreSQL advisory lock held on one connection. The difference matters when a process is killed rather than shut down: a table lease has to expire before anybody else may act, while a lock dies with the connection that held it - the operating system does the releasing, and the successor takes over in seconds (observability-reliability.md §9).

type Lease 

type Lease struct {
	// Now is the reading of the clock port, so that a test does not have to wait for time.
	Now time.Time
	// Until is when the claim expires. A job whose lease has run out is claimable again -
	// that is the only thing standing between a killed process and a job nobody ever finishes.
	// It has to outlast the job's own timeout, otherwise a second worker starts the same work
	// while the first is still doing it.
	Until time.Time
	// Batch is how many jobs are claimed at once.
	Batch int
}

Lease is the terms on which a batch of jobs is claimed.

type Queue 

type Queue interface {
	// Enqueue adds a job, or does nothing when one with the same dedupe key is already waiting
	// or running. It is deliberately not an error: the caller asked for work to happen, and work
	// that is already scheduled to happen satisfies that.
	//
	// It answers the identifier of the job that is now scheduled, which is not always a new one:
	// when a dedupe key collapses the request into a job that is already there, the answer is that
	// job's. A caller answering a 202 has to name something the caller of *that* can poll, and
	// naming a row that was never written would be a job resource answering 404 for work that is
	// happening. A caller that does not answer a 202 discards it.
	Enqueue(ctx context.Context, request Request) (shared.ID, error)

	// Claim takes the next batch and marks it running until the lease expires. Implementations
	// use FOR UPDATE SKIP LOCKED, so several workers claim disjoint batches without waiting for
	// each other (ADR-0008).
	Claim(ctx context.Context, lease Lease) ([]Job, error)

	// Hold takes the row lock on one claimed job for the rest of the caller's transaction.
	//
	// It exists for the one duty whose correctness depends on it, the reminders' wake-up. A job
	// that decides when it next runs reads that moment from the data, and a write committing
	// between that read and the reschedule would find the row RUNNING - where Enqueue's conflict
	// clause cannot pull a wake-up forward - and its reminder would wait for a wake-up nobody
	// scheduled. Held from the start of the pass, such a write instead waits for the pass to end
	// and then meets either a pending row it can pull forward or a finished one it may replace.
	//
	// Every other handler is welcome to ignore it: a poller that always comes back has nothing to
	// lose by missing a pull-forward.
	Hold(ctx context.Context, job Job) error

	// Complete finishes a job for good, with the outcome its handler answered (Result.Outcome,
	// empty for none). Called in the same transaction as the handler's effect, and refused when the
	// job's lease no longer holds.
	Complete(ctx context.Context, job Job, outcome string) error

	// Repeat returns a job to the queue for another round at runAt, with its attempt count
	// cleared: a poller's next round is not a retry of the last one.
	Repeat(ctx context.Context, job Job, runAt time.Time) error

	// Fail records an attempt that did not work: back to the queue when there is a retry left,
	// to the dead letter when there is not. It runs in a transaction of its own, because the
	// handler's has just been rolled back.
	Fail(ctx context.Context, failure Failure) error

	// Depth reports the backlog per kind, across every tenant. The job table has no tenant
	// boundary to respect here, which is what makes one global gauge possible at all.
	Depth(ctx context.Context) ([]Depth, error)
}

Queue is the job table. Like every repository it is called inside a unit of work and never opens a transaction of its own (project-structure.md §3).

type Refusal 

type Refusal struct {
	Code   string
	Params map[string]string
}

Refusal is the code and the parameters of a refusal, as a client renders any other: a message code and its parameters, never a sentence (rule 8).

type Releaser 

type Releaser interface {
	Handler

	// Release lets go of whatever the job's work holds beyond the job row itself. Best effort:
	// the runner logs a failure and does not retry it, so an implementation that can leave a
	// stale lock behind should say so where the lock is documented.
	Release(ctx context.Context, job Job)
}

Releaser is a handler that holds a lock outside the job table - a run row, a claim - and has to let it go when the queue gives up on the job.

The runner calls Release once, when a job goes to the dead letter. Without it, a lock whose row says RUNNING outlives the job that would have finished it: every later backup at that target is refused for ever, and nothing on any dashboard says why. Release is the reconciliation at the moment the queue's own account of the work ends - not a retry, not a second attempt at the job, only the honest closing of what the job left open.

It must be safe to call for a job that never took its lock (the row may not exist, or may be terminal already) and safe to call twice - the runner promises one call per dead-lettered job, but a lease that expired between the failure and its recording can hand the job to a second worker whose failure dead-letters it again.

type Reporter 

type Reporter interface {
	// Report writes the fraction, between 0 and 1, and is fenced on the job's lease like every
	// other statement a handler runs: a worker that fell so far behind that somebody else took the
	// job over writes nothing.
	//
	// A failure to report progress is not a failure of the job. The number is for whoever is
	// watching, and losing it is worth a log line rather than a backup.
	Report(ctx context.Context, job Job, fraction float64) error
}

Reporter is how a long job says how far along it is.

A second interface rather than a method on Queue, for the reason persistence.Snapshot is one: almost no job needs it. Most finish in milliseconds and the honest answer to "how far along" is the null api-guidelines.md §5 documents - a client renders an indeterminate bar for it rather than a number nobody measured. A method on Queue would put it on every double in the repository for the sake of the one handler that runs for minutes.

type Request 

type Request struct {
	Kind Kind
	// TenantID is whose work this is. Zero for work that belongs to no tenant: the job table is
	// deliberately the one table without row level security, because a queue that could only be
	// read under a tenant context could never be read by a worker that does not know the tenant
	// yet (db/migrations/0001_init.sql).
	TenantID shared.ID
	// Payload is what the handler needs in order to start. It is data, never a domain object:
	// the row outlives the process that wrote it, and a type that changed shape in between would
	// take the job with it.
	Payload map[string]any
	// DedupeKey collapses the same request made twice into one job. It is unique per kind while
	// a job is pending or running, so "make sure this tenant is being dispatched" costs one
	// insert that usually does nothing.
	DedupeKey string
	// RunAt is the earliest the job may start. Zero means now.
	RunAt time.Time
	// MaxAttempts is how often the job is tried before it goes to the dead letter. Zero leaves
	// the database default in place.
	MaxAttempts int
}

Request is a job somebody asks for.

func RetentionSweep 

func RetentionSweep(tenantID shared.ID, runAt time.Time) Request

RetentionSweep is the request every write that seeds a tenant's sweep makes: a deletion that starts a trash clock, a workspace provisioned, a session opened (data-retention.md §5). One constructor, so the kind and the dedupe key - the tenant - are written once.

The run-at is the caller's to choose, and the enqueue keeps the earlier of a pending job's and the request's: a deletion asks for now, a session for one interval later, and a later request never moves an earlier run.

type Result 

type Result struct {
	// Repeat keeps the job alive instead of finishing it: the same row returns to the queue and
	// runs again after RepeatAfter. That is how a poller lives - one row per tenant that is
	// rescheduled forever, rather than a new row per round that the deduplication of the running
	// one would swallow anyway.
	Repeat      bool
	RepeatAfter time.Duration
	// Counts is what a handler wants written down about a job that succeeded, by name: how much
	// of a provider's answer the narrowing dropped, say. The runner logs them with the job's kind
	// and identifier and nothing else; a handler puts no content here (rule 10).
	Counts map[string]int
	// Refusal is a refusal the job met where it would have done its work - a provider that may no
	// longer be asked - rather than a failure. The runner ends the job FAILED with the refusal's
	// code and parameters, without a retry and without counting a dead letter: asking again
	// would meet the same answer, and nobody has to repair anything.
	Refusal *Refusal
	// Outcome is how a job that succeeded ended, where its kind says more than that it finished: a
	// message code a caller polling the job reads (`ai.proposed`, `ai.nothing_to_change`, ...).
	// Stored on the job row, which has no row level security, so a code and nothing else (rule 10).
	// Empty where the kind has nothing to say.
	Outcome string
}

Result is what a handler leaves behind when it succeeds.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.