Documentation
¶
Overview ¶
Package outboundsend is Layer 3 of the outbound pipeline (docs/design/async-message-pipeline.md): the River execution stage that submits an accepted message to the upstream provider (SES) and records the terminal outcome. It mirrors internal/webhookdelivery — a River Worker on the shared `outbound` queue, with River owning claim / retry / rescue.
Delivery is at-least-once: River re-drives a crashed job, so the provider may receive a duplicate if the SMTP submit is accepted but the worker crashes before marking the message sent. That residual is narrowed by the X-E2A-Message-ID wire header + SNS reconciliation (async-send-contract §3.1): the SNS consumer records provider-accept evidence on the row, the re-driven claim then settles the message as sent instead of re-submitting, and the terminal-failure guard (here and in the terminal reconciler, via the store's guarded MarkFailed) never declares a provider-accepted row failed. A final attempt that fails ambiguously defers its terminal write to the reconciler's provider-evidence grace window rather than firing an immediate — possibly false — email.failed.
One SMTP attempt per job attempt — River owns the multi-attempt envelope via NextRetry, so Work() stays short (the deliverer does a single submit, not an internal retry loop). See the design's "claim + rescue, not a lease" note.
Index ¶
Constants ¶
const MaxSendAttempts = 6
MaxSendAttempts caps app/permanent-error retries (bounded 4xx/unknown tail).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeliverOutcome ¶
type DeliverOutcome struct {
ProviderMessageID string
SentAs string
Err error
// Permanent marks a non-retryable failure (validation / permanent 5xx): the
// worker fails the message terminally instead of retrying.
Permanent bool
// Outage marks a provider-connection failure (relay unreachable/misconfigured):
// the worker snoozes without burning an attempt (design §8), up to the retry
// horizon. Mutually exclusive with Permanent in practice.
Outage bool
}
DeliverOutcome is the result of one SMTP submit attempt.
type Deliverer ¶
type Deliverer interface {
Deliver(ctx context.Context, j *SendJob) DeliverOutcome
}
Deliverer performs a SINGLE SMTP submit — River owns re-attempts. Implemented in the binary over internal/outbound's single-attempt path.
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
Jobs is the outbound-send integration on the shared River client: a jobs.Registrar (contributes SendWorker + the terminal reconciler) plus the transactional enqueue entry point the accept-tx calls. The shared client is injected via SetEnqueuer after jobs.New builds it (two-phase wiring, same as webhookdelivery / senderidentity).
func NewJobs ¶
NewJobs builds the integration with its dependencies (no client yet). pool backs the periodic terminal-state reconciler's scan.
func (*Jobs) EnqueueSendTx ¶
EnqueueSendTx enqueues a send job WITHIN the caller's transaction — the outbox pattern: the accept-tx's messages-row insert and this job commit together, so an `accepted` message can never exist without a send job (or vice versa). The accept-tx stamps the returned river_job id on messages.send_job_id so the reconciler can find stranded rows (`accepted` with no job). Mirrors webhookdelivery.EnqueueDeliveryTx.
func (*Jobs) ReconcilePending ¶
ReconcilePending enqueues an outbound_send job for every accepted message that has no send job yet (send_job_id IS NULL). Run ONCE at startup as the cutover.
Because the accept-tx is a single transaction (message insert + job enqueue + send_job_id stamp all commit together), a committed `accepted` row in steady state ALWAYS has send_job_id set — so the send_job_id IS NULL set is normally empty. This exists to enqueue (a) any pre-async `accepted` rows at the moment the mode is first flipped on, and (b) rows from a future accept-tx variant that doesn't stamp atomically. Idempotent: the per-row FOR UPDATE + send_job_id IS NULL guard means a re-run (or concurrent replica) never double-enqueues. Mirrors webhookdelivery.ReconcilePending. Returns the count enqueued.
func (*Jobs) RegisterJobs ¶
func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the SendWorker and terminal-state safety net to the shared client's bundle. Implements jobs.Registrar.
func (*Jobs) SetEnqueuer ¶
SetEnqueuer injects the shared client so EnqueueSendTx can insert jobs.
type OutboundSendArgs ¶
type OutboundSendArgs struct {
MessageID string `json:"message_id"`
}
OutboundSendArgs drives one outbound send. Args carry only the message id; the worker re-reads the messages row (the source of truth) each attempt.
func (OutboundSendArgs) Kind ¶
func (OutboundSendArgs) Kind() string
type RampDecision ¶
type RampGate ¶
type RampGate interface {
Reserve(ctx context.Context, req RampRequest) (RampDecision, error)
Confirm(ctx context.Context, messageID string) error
Release(ctx context.Context, messageID string) error
Resolve(ctx context.Context, messageID string) error
}
RampGate reserves recipient capacity for an eligible custom-domain send. Implementations must make a same-message/day call idempotent.
type RampRequest ¶
type SendJob ¶
type SendJob struct {
MessageID string
// UserID is the owning account — the tenant scope for the pre-provider
// suppression guard (suppressions are per-account).
UserID string
AgentID string // exact sending agent for agent-scoped consent checks
Domain string // exact registered sender domain
MessageType string // send|reply|test; platform tests are ramp-exempt
Status string // messages.delivery_status
EnvelopeFrom string
Recipients []string
RawMessage []byte // composed MIME
SentAs string // From identity decided at accept ("own_address"|"relay")
// AcceptedAt is messages.created_at — the outage tail's clock, so a job that has
// been snoozing through an outage past sendRetryHorizon can be terminated.
AcceptedAt time.Time
// ProviderAccepted is set when authoritatively correlated provider-accept
// evidence (an SNS-verified, header- or provider-id-matched SES
// notification) has been recorded for this message: the provider already
// has it — an earlier attempt's submit landed in the SMTP-accept↔mark-sent
// crash window — so the worker settles the row as sent instead of
// re-submitting a duplicate.
ProviderAccepted bool
// ProviderMessageID is the evidence-repaired provider id accompanying
// ProviderAccepted (” when no evidence).
ProviderMessageID string
}
SendJob is the send payload the worker loads from the messages row (Store.LoadForSend).
type SendWorker ¶
type SendWorker struct {
river.WorkerDefaults[OutboundSendArgs]
// contains filtered or unexported fields
}
SendWorker submits an accepted message and records the terminal outcome. Mirrors webhookdelivery.DeliverWorker.
func NewSendWorker ¶
func NewSendWorker(store Store, deliverer Deliverer, ramp ...RampGate) *SendWorker
func (*SendWorker) NextRetry ¶
func (w *SendWorker) NextRetry(job *river.Job[OutboundSendArgs]) time.Time
NextRetry overrides River's default backoff with the decided send envelope.
func (*SendWorker) Work ¶
func (w *SendWorker) Work(ctx context.Context, job *river.Job[OutboundSendArgs]) error
Work intentionally has no Timeout() override — a single SES submit comfortably fits River's 60s default JobTimeout. (Contrast the maintenance/sweep workers, which override it because they can run for minutes.)
type Store ¶
type Store interface {
// ClaimSend returns nil when the message is gone, trashed, terminal, or owned
// by a different River job.
// (agent-delete cascade / TTL) — the worker treats that as a no-op.
ClaimSend(ctx context.Context, messageID string, jobID int64) (*SendJob, error)
// ReleaseSend clears a side-effect-free attempt before River backoff.
ReleaseSend(ctx context.Context, messageID string, jobID int64) error
// MarkSent records the provider outcome monotonically from a pre-terminal
// state, including when trash won after ClaimSend.
MarkSent(ctx context.Context, messageID, providerMessageID, sentAs string) error
// MarkFailed is the GUARDED terminal write (async-send-contract §3.1): if
// provider-accept evidence has reached the row it settles the message as
// sent (+ email.sent) instead; otherwise it sets delivery_status='failed'
// with the given failure provenance + detail and emits email.failed — all
// in one transaction. Callers therefore invoke it to "finalize a terminal
// state", not to unconditionally fail.
MarkFailed(ctx context.Context, messageID string, attempt int, detail string, source delivery.FailureSource) error
// DeferTerminalFailure records a final attempt's diagnostic + releases the
// I/O claim WITHOUT declaring failed: the terminal reconciler declares the
// outcome after the provider-evidence grace window (or settles the row as
// sent when evidence arrives first).
DeferTerminalFailure(ctx context.Context, messageID string, jobID int64, detail string) error
// SuppressedRecipients returns the effective account-wide + exact-agent
// subset — the last-line guard before provider I/O.
SuppressedRecipients(ctx context.Context, userID, agentID string, recipients []string) ([]string, error)
}
Store is the messages-store surface the worker needs. Implemented over internal/identity in the binary. ClaimSend atomically checks that the message and agent are live and persists delivery_status='sending' for the stamped River job before provider I/O begins.
type TerminalReconcileArgs ¶
type TerminalReconcileArgs struct{}
TerminalReconcileArgs drives the periodic safety net for outbound messages whose stamped send job reached a terminal state before recording delivery.
func (TerminalReconcileArgs) Kind ¶
func (TerminalReconcileArgs) Kind() string
type TerminalReconcileWorker ¶
type TerminalReconcileWorker struct {
river.WorkerDefaults[TerminalReconcileArgs]
// contains filtered or unexported fields
}
TerminalReconcileWorker settles accepted/sending outbound messages after their stamped River job is terminal or has already been pruned. SendWorker is still the primary owner; the compare-and-set store transitions make races safe. The store's guarded MarkFailed is the single terminal write: a row with provider-accept evidence is settled as sent (+ email.sent), a row without evidence — once past the providerEvidenceGrace window — is declared failed with provenance 'local' (correctable, §3.1) + exactly one email.failed.
func NewTerminalReconcileWorker ¶
func NewTerminalReconcileWorker(pool *pgxpool.Pool, store Store, ramps ...RampGate) *TerminalReconcileWorker
NewTerminalReconcileWorker builds the periodic safety-net worker.
func (*TerminalReconcileWorker) Work ¶
func (w *TerminalReconcileWorker) Work(ctx context.Context, _ *river.Job[TerminalReconcileArgs]) error