Message lifecycle
Status: describes the system. This note is the design of how a message moves through MicroB2B: one routing decision made in the database, one record of progress per target, a lossy wake-up on the bus, claims with leases, a recovery sweep, and two operator actions. The operator’s view is in Routing and Troubleshooting; this note names the code.
The database is the one and only system of record for a message’s progress and the only place where routing is decided. A source plugin stores a message and dispatches it in the same transaction (or, for AS2, in a second transaction it controls); the dispatch matches the route table once, in Go, and inserts one queued delivery per target plugin instance. NATS carries nothing but a lossy wake-up per delivery on a subject that names only the target (delivery.<target>). A target claims a delivery with a compare-and-swap and a lease before it works on it, records the outcome on the delivery, and the message status users see is derived from its deliveries. A periodic sweep republishes wake-ups that were lost, a bounded number of times, after which a delivery is stalled and raises an alert.
Considered and rejected. Each target creating its delivery when it first handles the message, with routing staying the NATS subscription, was rejected because a message nobody handled would have no rows, so “nothing matched” could not be told from “lost”, which is the one distinction recovery needs; it also leaves the status of a message with several targets undecidable. JetStream was rejected because it would be a second place that holds message state and would still need the database claim, the sweep and idempotent handlers.
1. Vocabulary
1.1 Message stage
Stored on the message (messages.stage), one domain string persistence.MessageStage with ParseMessageStage, canonical slice and validating JSON.
| Stage | Meaning | Set by |
|---|---|---|
accepted | The row exists and its source plugin still owns it (AS2: being unpacked and checked); no deliveries | InsertMessage |
dispatched | The deliveries exist; at least one is not final | DispatchMessage / InsertAndDispatchMessage, and again by redispatch or redeliver |
settled | Every delivery is delivered, failed or stalled, or the message has none (no route matched, or the source refused it) | the persistence, in the transaction of the delivery transition that makes it true, or RefuseMessage |
1.2 Delivery
One delivery per (message, target plugin instance), persistence.DeliveryState (domain string, canonical slice, ParseDeliveryState, validating JSON, Final(), Redeliverable(), Label(), Tone() for the badge):
| State | Meaning | Final |
|---|---|---|
queued | Created or put back; nobody holds it. Woken by the dispatch and the sweep | no |
delivering | A target holds the claim (token and lease) | no |
retrying | An attempt failed in a way another may fix; next_dispatch_at is the next attempt | no |
awaiting_ack | Handed off; the partner’s end-to-end receipt (AS2 asynchronous MDN) has not arrived | no |
delivered | Done and, where a receipt exists, confirmed | yes |
failed | Ended for good, with the reason | yes |
stalled | Woken the maximum number of times and never claimed (section 5). Final until the target claims it at its start or an operator acts | yes (counts as settled) |
1.3 The status a user sees
persistence.MessageStatus is the one value of the list, the filter, the badge and REST, with this canonical set (labels in brackets):
Received (“Received”), InProgress (“In progress”), Retrying (“Retrying”), AwaitingReceipt (“Awaiting receipt”), Stalled (“Stalled”), Delivered (“Delivered”), NotRouted (“Not routed”), Refused (“Refused”), Error (“Error”).
The core names no protocol: “MDN” is AS2’s word, and only the AS2 message panel says it.
The derivation rule is one pure function, persistence.DeriveStatus(stage MessageStage, refusal Refusal, deliveries []DeliveryState) MessageStatus; the first match wins:
| Condition | Status |
|---|---|
stage accepted | Received |
no deliveries and refusal is RefusalPartner | Refused |
no deliveries and refusal is RefusalInternal | Error |
| no deliveries | NotRouted |
any failed | Error |
any stalled | Stalled |
any retrying | Retrying |
any queued or delivering | InProgress |
any awaiting_ack | AwaitingReceipt |
otherwise (all delivered) | Delivered |
Refusal (persistence.Refusal, ParseRefusal, validating JSON) is the reason class RefuseMessage takes (section 3.4). Work still to be done on our side (retrying, queued, delivering) ranks above waiting for a partner, so a message with one delivery awaiting a receipt and one still queued shows “In progress”.
MessageStatus.Settled() is true for Delivered, NotRouted, Refused, Error and Stalled. In flight (InFlightMessageStatuses, the dashboard card) is Received, InProgress, Retrying and AwaitingReceipt. The actions have their own rules (section 6).
1.4 Stored, not computed per query
The status is derived and stored: messages.status is written by the persistence, and only by it, in the same transaction as every stage or delivery transition, from DeriveStatus over the message’s deliveries (SELECT state FROM deliveries WHERE message_id = ?, at most a handful of rows, on the unique index). Nothing else writes it and no plugin names a message status.
Why stored: the message list, its filter and count, the dashboard cards (polled by the live dashboard) and the live list would otherwise each need a GROUP BY over deliveries joined to messages for every page. On SQLite’s single connection that serializes behind every delivery write; stored, they stay the indexed status = ?. The cost is one small read in each transition transaction. TestStoredStatusIsTheDerivedStatus (persistence) drives random transitions and compares the column with DeriveStatus after each.
The status transition is what logStateChange records (status_changed) and what raises message_failed (for Error only), so alerting stays in one place.
2. Data model
All in the core: deliveries names targets by instance name only, so the core names no module. The models are in internal/persistence/bun/models.go and CoreSchema (dbinit.go). There are no migrations: a changed model means resetting the database.
2.1 deliveries (Bun model delivery)
| Column | Type | Notes |
|---|---|---|
id | uuid, pk | the wake-up payload, the REST id, delivery_id in logs |
message_id | uuid, not null | no foreign key (plain id columns) |
target | string, not null | the route’s target_plugin |
state | persistence.DeliveryState, not null | validated on write and read |
attempts | int, not null, default 0 | incremented by each claim |
dispatch_count | int, not null, default 0 | wake-ups published since the last claim (section 5) |
requeues | int, not null, default 0 | times the delivery was put back in the queue after it had started (redeliver, redispatch); a target that sees a count above zero knows an earlier pass may have left its work behind |
next_dispatch_at | timestamp, nullzero | when the sweep looks at it next: queued the next wake-up, retrying the next attempt, delivering the end of the lease, awaiting_ack the receipt deadline; null for final states |
claim_token | uuid, nullzero | the current claim; every write by a claimant is conditional on it |
last_error | string, not null, default '' | bounded like history text; never a URL, header or payload |
created_at, updated_at | timestamp | |
finished_at | timestamp, nullzero | when it became delivered or failed |
trace | telemetry.TraceParent, not null, default '' | the W3C traceparent of the dispatch (or redispatch) that created it; its wake-ups, claims and attempts continue that trace (DeliveryClaim.Trace, not in the REST resource) |
Indexes (CoreSchema.Indexes):
| Index | Columns | For |
|---|---|---|
unique deliveries_message_target | message_id, target | one delivery per target; the derive read; idempotent dispatch |
deliveries_due | state, next_dispatch_at | the sweep (due queued/retrying, expired delivering, awaiting_ack deadlines) |
deliveries_target_due | target, state, next_dispatch_at | a target’s catch-up at start |
messages_status_created | status, created_at on messages | list filter, the in-flight count |
messages_created | created_at on messages | the dashboard’s and the statistics’ windows |
persistenceBun.Index has no partial index; the composite indexes are enough at this scale.
2.2 messages
stage (persistence.MessageStage, not null); status, written only by the derivation; error, the source’s refusal reason or the last_error of the first failed delivery (written with the status). persistence.Message carries Stage and Deliveries []Delivery (filled by GetMessage, empty in lists). A message holds no retry fields: retries belong to the delivery.
2.3 message_history
Columns delivery_id (uuid, nullzero), target, old_state, new_state (both DeliveryState, empty for other kinds). Kinds:
| Kind | Written when |
|---|---|
status_changed | the message status changes (old_status, new_status) |
delivery_changed | on every delivery transition; the text is the outcome text the target gave (a file written, an HTTP answer, a retry and why); the status change it causes repeats no text |
redispatch_requested, redeliver_requested | an operator’s action; the text names the operator’s subject |
mdn_received, mdn_refused, duplicate_received | AS2 receipt and repeat events |
2.4 Settings
persistence.Settings has Delivery DeliverySettings{MaxDispatches int} (JSON delivery.max_dispatches, default 5, 1 to 20), on the Settings page through internal/api/resource/settings.go and in /api/v1/settings. Like every setting it is read when a process starts. The back-off steps are constants (section 5).
2.5 AS2
as2_messages has delivery_id (outgoing rows only): Sender records it with the outgoing details, and MDNReceiver.locateSent reads it to settle the right delivery. The remote profile has receipt_timeout_hours (OverridableProfileParameters, default as2.DefaultReceiptTimeoutHours = 24, 1 to ReceiptTimeoutHoursCeiling = 720), the time the sender waits for an asynchronous receipt.
3. State machine
3.1 Message
| From | Event | To | Who | Condition |
|---|---|---|---|---|
| (none) | InsertMessage | accepted | AS2 receiver | |
| (none) | InsertAndDispatchMessage | dispatched, or settled with no deliveries | file poller | one transaction |
accepted | DispatchMessage | dispatched, or settled with no deliveries | AS2 receiver | after unpack, claim and unpacked stored |
accepted | RefuseMessage(refusal, reason) | settled (status Refused or Error) | AS2 receiver (unpack, claim, storage failure, repeat, replay); a source at start for abandoned rows (section 4.4) | only from accepted |
dispatched | last delivery becomes final | settled | persistence | in the transition’s transaction |
settled | redispatch adds or wakes deliveries, redeliver | dispatched | operator | section 6 |
settled | stalled delivery claimed at target start | dispatched | target | section 5 |
3.2 Delivery
| From | Event | To | Who | Condition / effect |
|---|---|---|---|---|
| (none) | dispatch | queued | persistence | dispatch_count = 1, next_dispatch_at = now + 10 s, wake-up after commit |
queued | ClaimDelivery | delivering | target worker | attempts + 1, new claim_token, next_dispatch_at = lease end, dispatch_count = 0 |
retrying | ClaimDelivery | delivering | target worker | only when next_dispatch_at <= now, or when the sweep has woken it (dispatch_count > 0: the wake-up moved next_dispatch_at on to the next back-off step) |
stalled | ClaimDelivery | delivering | target worker | only with fromStalled, from the target’s start-up catch-up |
delivering | CompleteDelivery delivered | delivered | target worker | token matches |
delivering | CompleteDelivery awaiting_ack | awaiting_ack | AS2 sender (async MDN requested) | token matches; next_dispatch_at = the receipt deadline; a token that was cleared (the MDN arrived first) is ErrClaimLost, logged at INFO |
delivering | CompleteDelivery retry | retrying | target worker | token matches; next_dispatch_at = At, dispatch_count = 0 |
delivering | CompleteDelivery failed | failed | target worker | token matches |
delivering | ReleaseDelivery (stop) | queued (attempt not counted) or retrying due now (the attempt had started its I/O) | target worker | token matches; written with a context that is not cancelled |
delivering | lease expired | retrying, next_dispatch_at = now | sweep | history “attempt interrupted: lease expired” |
queued, retrying | due, dispatch_count < MaxDispatches | same state | sweep | dispatch_count + 1, next_dispatch_at = now + DispatchBackoff(count), wake-up |
queued, retrying | due, dispatch_count = MaxDispatches | stalled | sweep | raises delivery_stalled |
awaiting_ack | deadline passed | failed | sweep | “no receipt within |
delivering, awaiting_ack, retrying | SettleDelivery delivered | delivered | AS2 MDN receiver | clears the claim |
delivering, awaiting_ack, retrying | SettleDelivery failed | failed | AS2 MDN receiver | never downgrades delivered; the first failure’s reason stands |
failed, stalled, retrying, delivered | redeliver | queued | operator | attempts = 0, dispatch_count = 0, requeues + 1, claim and finished_at cleared, wake-up |
queued, stalled | redispatch | queued | operator | dispatch_count = 0, requeues + 1, wake-up |
3.3 What each plugin returns
A target implements delivery.Deliverer (section 4.5) and returns an outcome; it never writes a state itself.
| Plugin | Situation | Outcome |
|---|---|---|
as2-sender | profile missing, build (compress/sign/encrypt), file name, MDN settings | Failed |
| payload not loadable, TLS client, POST error or non-2xx, async MICs not recorded | Retry{At: policy.delay(attempts)} while attempts < policy.maxRetries, else Failed("max retries exceeded: ...") (the resend policy of the partner profile) | |
| success, no MDN | Delivered | |
| success, sync MDN confirms / does not | Delivered / Failed("synchronous MDN: ...") (not retried) | |
| success, async MDN requested | AwaitingAck with the deadline now + receipt_timeout_hours; MDNReceiver later SettleDelivery(as2_messages.delivery_id, ...); a forged or unattributable MDN leaves it awaiting_ack | |
file-writer | written | Delivered (text FileWrittenText) |
| payload unavailable, I/O error | Retry on the writer’s fixed schedule (file.WriteRetryDelays = 30 s, 2 min, 10 min, then Failed) | |
| stage not recorded, pattern error, name exists without Overwrite | Failed | |
| re-attempt after an interrupted attempt or a requeue and the target file exists with exactly the payload’s bytes | Delivered (the write had completed) | |
http-poster | 2xx | Delivered |
retryable (httpout.Classify) and attempts < MaxAttempts | Retry{At: now + retryWait(attempts)} (the interval, doubled per attempt, up to an hour; Retry-After is not honored) | |
| otherwise | Failed; the message’s message_failed alert reports it |
3.4 AS2 inbound specifics
| Case | What happens |
|---|---|
| Unknown parties, receipt URL not allowed, too large | refused before a row exists |
| Unpack fails (signature, decryption, unsigned where signing is required), repeat or replay | RefuseMessage(RefusalPartner, ...): settled, status Refused, no deliveries, no alert; failed MDN. For a repeat the reason is “Repeat of message …” and the first message gets duplicate_received |
| Accepted | DispatchMessage commits, then CompleteInboundMessage, then the positive MDN (sync in the response, async queued). A message is accepted for the partner exactly when its dispatch has committed |
DispatchMessage or storage fails | RefuseMessage(RefusalInternal, ...): status Error, message_failed; failed MDN FailureReasonDatabase or its storage counterpart; the deferred ReleaseInboundMessage frees the key so the partner’s resend is accepted |
| Route matches nothing | NotRouted and one message_not_routed alert; the partner got a positive MDN |
4. Dispatch, routes and subjects
4.1 Dispatch API and transaction boundaries
Methods of persistence.Persistence (rpcgen generates client and server):
// InsertMessage stores a message in stage accepted with its first payload records.
InsertMessage(ctx, msg *Message, payloads []MessagePayload) error
// InsertAndDispatchMessage stores a message and dispatches it in one transaction.
InsertAndDispatchMessage(ctx, msg *Message, payloads []MessagePayload) (dispatch *Dispatch, err error)
// DispatchMessage dispatches an accepted message; on a message that is not accepted it changes nothing and returns its deliveries.
DispatchMessage(ctx, id uuid.UUID) (dispatch *Dispatch, err error)
// RefuseMessage settles an accepted message with no deliveries; the Refusal picks Refused or Error.
RefuseMessage(ctx, id uuid.UUID, refusal Refusal, reason string) errorDispatch{Stage, Status, Deliveries []Delivery}. The dispatch, in one transaction: read the source, from and to from the row (stage = accepted as the condition of the update), match routes (4.2), insert one queued delivery per distinct target, set stage and status, write the history lines (delivery_changed per delivery, or “No route matched (source, from, to)”). After commit the persistence publishes one wake-up per delivery through its delivery notifier (4.3) and, for a message nothing matched, raises message_not_routed. A route whose target is no usable instance name fails the dispatch with ErrInvalidTarget and leaves the message accepted.
- File poller:
InsertAndDispatchMessage: there is no moment when the row exists undispatched. A crash before commit leaves the file to be picked up again; a crash after commit and before the file is moved makes a second message at the next scan (risk R9). - AS2 receiver: the row must exist before unpacking (audit, refusal), so it is
InsertMessagethenDispatchMessage. A crash between the two cannot lose the message: the partner received no positive MDN and resends, and the claim’s lease (InboundClaimLease) lets the resend in. A crash afterDispatchMessageand beforeCompleteInboundMessagewould let a resend in after the lease; soClaimInboundMessagereports a stale incomplete holder (Stale), the receiver asksGetMessage(holder).Stage, and when that message is pastacceptedthe claim is completed and the resend is answered as a repeat; otherwiseTakeOverInboundMessagetakes the claim over. - Distributed mode: the source plugin’s process holds no route data. The rpcgen client sends
InsertAndDispatchMessageorDispatchMessageto the persistence process, which matches, writes and publishes. The poller’s dispatch costs no extra round trip (it replacesInsertMessage); the receiver’s costs one. - Publish after commit, never inside the transaction: a wake-up must never name a row the claimant cannot see yet. The window between commit and publish is covered by the sweep (
next_dispatch_at = now + 10 s).
4.2 Route matching and the route cache
persistence.MatchRoutes(routes []Route, source, from, to string) []string(route.go): the distinct targets of the routes withSourcePlugin == sourceand, in each position,patternCovers(pattern, name), the token ruleRoute.Coversuses. Deduplicated by target, so overlapping routes to one target make one delivery. This is the only routing decision in the system.- The cache lives in the core Bun persistence (
bunPersistence.routes,routes_cache.go), because every dispatch and every route write runs there: the standalone process, or the persistence process in distributed mode. It holds the whole route table (small), loaded lazily on the first dispatch (so fixtures and seeding written straight into the table at startup are seen). - Invalidation:
CreateRoute,UpdateRoute,DeleteRoutedrop the cache after commit, in the same persistence; they also publish apersistence.RouteChangeonevents.routes(WithRouteChanges(disp), wired in standalone and in the persistence process’s RPC server), and every Bun persistence subscribed to it drops its cache on a change from another instance (a second persistence process on PostgreSQL). A backstop TTL (RouteCacheTTL= 30 s) covers writes that bypass the persistence (the developer database console) and a lost event. - Staleness window: none for a route written through the persistence that dispatches; the NATS latency for another persistence process; at most 30 s for a write that bypasses the persistence. A route takes effect at the next dispatch, with no restart; a message dispatched before keeps its deliveries (redispatch adds the missing ones).
4.3 Subjects
| Subject | Built by | Carries |
|---|---|---|
delivery.<target instance> | persistence.DeliverySubject(target) (string, error), which refuses a name that is not one token of lower case letters, digits, _ and - (at most MaxInstanceNameLength) | persistence.Wakeup{DeliveryID, MessageID} (JSON), dispatcher.Event.MessageID set for traces and logs |
events.routes | persistence.RouteChangesSubject | RouteChange{Persistence instance id} |
- A fixed first token keeps the namespace apart from
events.*,notification.test.*and the RPC subjects, so no instance or party name can collide. An instance name is unique across plugin types (decision 1, section 11), so the name alone identifies the target. - Each target subscribes to exactly one static subject, its own, at
Start; nothing reads routes in a target.persistence.UnmarshalWakeupdecodes the payload. - The persistence publishes wake-ups through
WithDeliveryNotifier(persistence.PublishWakeups(disp)), the same pattern asWithMessageHistoryNotifier; the notifier never blocks (a failed publish is a DEBUG line and a metric; the sweep recovers it). - Sources publish nothing on NATS for messages.
| Plugin type | Subscribes | Publishes |
|---|---|---|
file-poller, as2-receiver | nothing for messages | nothing (the persistence publishes the wake-ups) |
as2-sender, file-writer, http-poster | delivery.<own name> through delivery.Worker | nothing |
| persistence (standalone or process) | events.routes | delivery.<target>, events.routes, events.message_history |
4.4 Abandoned accepted messages
The sweep never dispatches: only the source knows whether an accepted message is complete. A source plugin that leaves accepted rows behind settles its own at Start with RefuseMessage through ListAcceptedMessages(source, olderThan, limit). The AS2 receiver uses olderThan = now - InboundClaimLease (the partner’s resend is let in after the lease, so the old row is superseded), reason “interrupted before it was accepted; the partner resends it”. The poller never leaves one.
4.5 The delivery worker (internal/delivery, core)
One package, naming no module. The worker is generic over its jobs: a Store[T] keeps them and a Deliverer[T] attempts one. The deliveries of messages to target plugin instances are one kind of job; a module keeps jobs of its own in a store of its own (an AS2 receiver’s asynchronous MDNs, section 4.7; a notifier’s notifications, docs/notifications.md) and works them with the same worker.
type Deliverer[T any] interface {
// Deliver makes one attempt. ctx ends at the attempt timeout or when the worker stops.
Deliver(ctx context.Context, work T) Outcome
}
type Store[T any] interface {
Claim(ctx, id, now, leaseUntil, fromStalled) (Claim[T], error) // compare-and-swap, counts the attempt
Complete(ctx, id, token, outcome) error // conditional on the claim (ErrClaimLost)
Release(ctx, id, token, started, reason) error // a stop gives the claim back
ListDue(ctx, now, limit) ([]Due, error)
}
type Outcome // Delivered(text) | AwaitingAck(text, deadline) | Retry(at, reason) | Failed(reason)
func NewWorker[T](Config[T]{Name, Store, Deliverer, Wakeups, Annotate, Options}) *Worker[T]
// Options: MaxConcurrent, AttemptTimeout (the lease is 2*AttemptTimeout + 1 min), DrainGrace, PollInterval, Now
func (w *Worker[T]) Start(ctx context.Context) error // subscribe (if Wakeups), then look at the store; poll if PollInterval
func (w *Worker[T]) Poke() // a polling worker looks now
func (w *Worker[T]) RunOnce(ctx context.Context) error // work off what is due, return when it is recorded
func (w *Worker[T]) Quiesce() // unsubscribe, stop polling, claim nothing more
func (w *Worker[T]) Stop(ctx context.Context) error // drain, cancel, release (section 7)
// The worker of a target plugin instance: its deliveries, woken on delivery.<target>.
type Attempt struct { Delivery persistence.Delivery; Message persistence.Message }
func NewTargetWorker(target string, p persistence.Persistence, d dispatcher.Dispatcher, impl Deliverer[Attempt], opts Options) (*Worker[Attempt], error)Per wake-up: the subscription callback decodes the wake-up, acquires a slot of the semaphore (MaxConcurrent; selects on the lifecycle context), then starts a goroutine (tracked by a sync.WaitGroup) that calls Store.Claim (ClaimDelivery for a target); not claimed: release the slot and count a claim conflict; claimed: run Deliver with the attempt context (section 7), record the outcome with Store.Complete on a context that is not cancelled and has its own 10 s deadline. Backpressure: while every slot is taken the callback waits on the semaphore, so wake-ups queue in the subscription’s pending buffer (tiny messages; an overflow drops wake-ups, which the sweep recovers).
The worker keeps no timer. A recorded Retry is a row with a due time; the sweep wakes it when it is due, so a retry starts up to SweepInterval after its time.
The claim is the unit of concurrency: max_concurrent_deliveries is real for the AS2 sender, the file writer (default 1) and the HTTP poster (default 5). Deliveries of one target may complete out of order; the docs say so.
Catch-up at Start: after subscribing, the worker lists its due jobs (Store.ListDue; for a target ListDueDeliveries(target, now, limit): queued, retrying due or woken, stalled) and handles them like wake-ups, so a target that was down picks up its backlog without waiting for the sweep, and its stalled deliveries recover when it comes back.
Polling: a worker without wake-ups sets PollInterval and looks at its store at that interval and whenever it is poked. Its store is its own sweep: ListDue includes the jobs whose claim’s lease ran out, and Claim takes them over.
4.7 Asynchronous MDNs (AS2)
An asynchronous MDN the receiver owes a partner is not a delivery of a routed message: its outcome does not change the message’s status, and redispatch does not apply. It is a job of the receiver instance’s own worker over as2.AsyncMDNStore (table as2_pending_async_mdns: Pending, Delivering with a claim token and the lease in next_attempt_at, Retrying, Delivered, Failed), with as2.AsyncMDNSender as the Deliverer. The worker polls every mdn_poll_interval and the receiver pokes it when it queues an MDN, so an MDN is attempted at once. CompletePendingAsyncMDN refuses AwaitingAck and raises async_mdn_failed after the commit of a failure. The receiver plugin quiesces and stops the worker like a target’s.
4.6 Claim API and errors
| Method | Does |
|---|---|
ClaimDelivery(ctx, id uuid.UUID, target string, now, leaseUntil time.Time, fromStalled bool) (*DeliveryClaim, error) | conditional UPDATE ... WHERE id = ? AND target = ? AND (state = 'queued' OR (state = 'retrying' AND (next_dispatch_at <= now OR dispatch_count > 0)) OR (fromStalled AND state = 'stalled')); returns {Claimed, Token, Delivery, Message} (Delivery and Message are nil when not claimed) so the target needs no second round trip |
CompleteDelivery(ctx, id, token uuid.UUID, outcome DeliveryOutcome) error | conditional on state = 'delivering' AND claim_token = token; ErrClaimLost when not |
ReleaseDelivery(ctx, id, token uuid.UUID, started bool, reason string) error | stop path (section 7) |
SettleDelivery(ctx, id uuid.UUID, outcome DeliveryOutcome) (applied bool, err error) | async MDN receipt |
ListDueDeliveries(ctx, target string, now time.Time, limit int) ([]Delivery, error) | catch-up |
ListDeliveries(ctx, messageID uuid.UUID) ([]Delivery, error) | message page and REST |
ListAcceptedMessages(ctx, source string, olderThan time.Time, limit int) ([]Message, error) | section 4.4 |
Redispatch(ctx, messageID uuid.UUID) (message *Message, changed bool, err error), Redeliver(ctx, deliveryID uuid.UUID) (*Message, error) | section 6 |
DeliveryOutcome is a closed struct (Kind domain string delivered|awaiting_ack|retry|failed, At, Deadline, Text) with Check, validated like every domain type: an awaiting_ack has a Deadline, no other kind has. The sweep is not part of Persistence (it is not called over RPC): persistence.Sweeper (Sweep(ctx, now, SweepLimits) (SweepResult, error)) is implemented by the Bun persistence only (bun.NewSweeper).
Error vocabulary (persistence.ErrorVocabulary, so they cross the RPC with their identity; rpcerrtest.RoundTrip covers them):
| Code | Entry |
|---|---|
claim_lost | rpcerr.Sentinel(ErrClaimLost) |
not_redispatchable | rpcerr.Typed(&NotRedispatchableError{Stage}) |
not_redeliverable | rpcerr.Typed(&NotRedeliverableError{State}) |
not_accepted | rpcerr.Sentinel(ErrNotAccepted) (RefuseMessage on a dispatched message) |
invalid_target | rpcerr.Sentinel(ErrInvalidTarget) (a route row with an unusable name, surfaced by dispatch) |
5. The sweep and bounded dispatch
Who and where: a loop (delivery.StartSweep, over persistence.Sweeper) in the process that owns the database. Standalone starts it after manager.StartAll, so the first pass finds every target subscribed; the persistence process starts it with its RPC server. It is not a plugin and not on the Services page, like the core persistence. Interval SweepInterval = 5 s; its first pass runs inside StartSweep, before it returns. Two persistence processes on one PostgreSQL both sweep; every write is conditional, so a delivery is woken once per due time.
One pass, each step an indexed query plus one conditional update per row, in batches of SweepBatch = 500 with a short pause between batches:
- Expired leases:
state = 'delivering' AND next_dispatch_at <= now->retrying,next_dispatch_at = now,dispatch_count = 0,claim_tokencleared, history “attempt interrupted: lease expired”. - Due wake-ups:
state IN ('queued','retrying') AND next_dispatch_at <= now ORDER BY next_dispatch_at: withdispatch_count < MaxDispatches:dispatch_count + 1,next_dispatch_at = now + DispatchBackoff(count), wake-up after commit; withdispatch_count = MaxDispatches:stalled,last_error = "not claimed by <target> after <n> wake-ups", status recomputed, alert. - Receipt deadlines:
state = 'awaiting_ack' AND next_dispatch_at <= now->failed,last_error“no receipt within” (measured from the hand-off to the deadline), history line, the message status derived as usual. The deadline is set by the target when it completes the attempt as AwaitingAck(DeliveryOutcome.Deadline), so the core knows no protocol: the AS2 sender takes it from the partner’s profile (receipt_timeout_hours). A receipt that arrives later is refused as unattributable (mdn_refused), as a receipt for any settled delivery is. - Gauges: the number of deliveries per state (one
GROUP BY state, every state has an entry) and the due count, for metrics.
Back-off: DispatchBackoff = 10 s, 30 s, 2 min, 10 min (constants; the last step repeats when MaxDispatches is raised). With the default of 5 a never-claimed delivery is woken at 0 s, 10 s, 40 s, 2 min 40 s and 12 min 40 s and stalls at 22 min 40 s.
Rules that make it safe:
- Only unclaimed wake-ups count: a claim resets
dispatch_countto 0, and adeliveringdelivery is never woken (itsnext_dispatch_atis the lease end). - A retrying delivery the sweep woke is claimable although its
next_dispatch_atnow holds the next back-off step (dispatch_count > 0); the catch-up list includes it. - A message queued behind a busy target is claimed when a slot frees and is not stalled by being seen often; a backlog larger than a target can claim in 22 minutes does stall, which is the alert an operator needs (and the target’s next start claims it again).
stalleddiffers fromfailed: nothing was tried. An operator’s stop on the Services page stalls the target’s deliveries after the wake-ups run out; its next start claims them (catch-up includesstalled), so a maintenance window needs no manual action.- “No route matched” is decided at dispatch and never swept.
Alert: alert.CodeDeliveryStalled = delivery_stalled, critical, in alert.CodeTable, raised by the sweep once per delivery that stalls: subject the message, details target, wake_ups, from, to, link /ui/messages/<id>, dedupe key delivery_stalled:<target> (so the notifier’s cooldown and grouping fold a stopped target’s backlog into one notification). A delivery that fails for good is not an alert of its own: the message reaches Error and message_failed is raised through the status.
6. Redispatch and redeliver
| Action | On | Allowed when | Does |
|---|---|---|---|
| Redispatch | a message | stage is not accepted (persistence.CanRedispatch; its source owns an accepted message) | matches the routes again, inserts the deliveries of targets the message has none for, sets queued and stalled ones to queued with dispatch_count = 0, stage dispatched, status recomputed, history redispatch_requested, wake-ups after commit. Delivered, failed, retrying, delivering and awaiting deliveries are left alone. A call that changes nothing (no new target, nothing queued or stalled) succeeds, writes no history line and reports changed = false |
| Redeliver | one delivery | failed, stalled, retrying or delivered (DeliveryState.Redeliverable) | queued, attempts = 0, dispatch_count = 0, last_error cleared, claim and finished_at cleared, requeues + 1, history redeliver_requested, wake-up after commit; a delivered message is InProgress again until the delivery ends |
- One function each, the persistence methods
RedispatchandRedeliver(one transaction), called by the UI and REST. There is no permission check (an authenticated user is an administrator); the operator’s subject comes fromauth.IdentityFromand is the history text and theuserof the log line (operator action,action,message_id,delivery_id). They are not audit events: the audit log records configuration, and the message’s history names the operator beside what happened. - REST:
POST /api/v1/messages/{id}/redispatchandPOST /api/v1/messages/{id}/deliveries/{delivery_id}/redeliver(Content-Type: application/json, no body, underRequireContentType), 200 with the message resource; 404NOT_FOUND; 409NOT_REDISPATCHABLE/NOT_REDELIVERABLEwithdetails.stageordetails.stateanddetails.allowed, mapped inrest.WriteFailure(the one mapping).persistence.FindDeliverychecks that a delivery belongs to the message of the request. - UI:
POST /ui/messages/{id}/redispatch(a plain form with the CSRF field, 303 back withwebui.SetFlash),POST /ui/messages/{id}/deliveries/{delivery_id}/redeliver(a button withhx-post,hx-confirmnaming the target,hx-swap="none", answered with a flash andwebui.PageRefreshed, because it sends again). Both under CSRF; a refused redispatch is the page again with 409 and the reason as flash, a refused redeliver a flash with the reason andwebui.PageRefreshed. - Send again is Redeliver of a
delivereddelivery, for every target type: an operator may have lost the file or the downstream copy. The page words the button “Send again to” and its confirmation is the generic text (the target receives the payload a second time) unless the target’s module contributes its own ( webui.Extension.SendAgainWarning; the core names no module). For an AS2 sender the message goes out with the same Message-ID (decision 2), so the partner sees a repeat: it may answer with the receipt it stored or with none, and without a receipt the delivery fails at the receipt deadline and the message ends inError. The file writer treats a file with exactly the payload’s bytes as its write when the delivery has been requeued (Delivery.Requeues > 0), writes a missing file, and fails on a taken name with other content when Overwrite is off.awaiting_ack,queuedanddeliveringare refused.
7. Graceful stop
- Dispatcher:
nats.Dispatcherowns a base context, cancelled byClose(whichprocess.Runcalls last); the subscription handler’s context derives from it and keeps the extracted trace. It is a backstop only. - Manager:
StopAllfirst callsQuiesce()on every plugin that implements the optionalplugin.Quiescer(unsubscribe, claim nothing more; non-blocking), then stops the plugins in reverse start order (PluginStopTimeout10 s each insideprocess.ShutdownTimeout30 s). No target claims new work while an earlier one drains. - Worker
Stop(ctx): wait for in-flight attempts (theWaitGroup) forDrainGrace= 5 s or until 2 s beforectx’s deadline, whichever is first; then cancel the lifecycle context so stuck I/O aborts; then wait for the attempts to return. - Per attempt: the attempt context is
context.WithoutCancel(handlerCtx)(keeps message id, delivery id, trace), thencontext.WithCancelpluscontext.AfterFunc(lifecycle, cancel), thencontext.WithTimeout(AttemptTimeout).as2.sendMessageand every other I/O of a target use this context. - Cancelled attempts: the worker calls
ReleaseDeliverywith a context that is not cancelled and has its own 2 s deadline:queuedwhen the attempt had not started its I/O,retryingwithnext_dispatch_at = nowwhen it had (the partner may have received it; AS2’s duplicate detection answers a resend of the same Message-ID with the stored MDN). Nothing becomesfailedbecause of a stop. Attempts that finish within the grace period complete normally. - A process killed without a stop leaves
deliveringrows; the sweep turns them intoretryingat lease end.
8. UI, REST, observability
8.1 UI (internal/api/ui)
status_badge: classes fromMessageStatus.BadgeClass(); each colour pair is instyle_test.go(pairs, AA in both themes, the dark blocks kept identical).- Messages list filter: rendered from
MessageStatusOptions. Dashboard: the in-flight card’s footer is “Received, in progress, retrying or awaiting receipt”; the failed card countsErrorandStalledin 24 h (“Error or stalled”,MessageWindowCounts.AttentionoverAttentionMessageStatuses); a separate Refused card countsRefused(MessageWindowCounts.Refused). Every card’s figure, and/api/v1/stats, comes fromPersistence.CountMessageStatistics: the in-flight count onmessages_status_created, and one pass over the messages created since the earliest window (messages_created) with a conditional sum per figure. Nothing counts the whole table (the first card is the messages of the last 7 days). - Message page: a Deliveries table (target, state badge via
DeliveryState.Tone, attempts, next attempt for a retrying delivery, last error, updated;nowrap/wrap-anywhereby role; no row links) with an Actions column holding one icon button per redeliverable delivery (“Redeliver to”, “Send again to ” for a delivered one) and, above the table, the message-level Redispatch. A delivery_changedhistory line streamed on the page also carries the delivery’s table row out of band (hx-swap-oobonid="delivery-<id>"), so the table follows live. History rows showtarget: Old -> New: textfordelivery_changed. - Every link and form target is served (
TestEveryLinkInATemplateIsServed).
8.2 REST and OpenAPI (internal/api/handlers/messages.go, internal/api/openapi/openapi.yaml)
Message:statusenum = the canonical set;stage;deliveries(array ofDelivery:id,target,state,attempts,next_attempt_at,last_error,created_at,updated_at,finished_at) onGET /messages/{id}.GET /messages?status=accepts the canonical set (INVALID_STATUSlists it fromMessageStatusOptions).GET /stats:in_flight_messages;failed_last_24hcountsErrorandStalled.- History kinds enum (
/messages/{id}/history,/events): section 2.3;delivery_changedlines carrydelivery_id,target,old_state,new_state. RouteRequestdescription: matching happens at dispatch; a route applies to messages dispatched after it is saved; overlapping routes to one target deliver once.- The paths of section 6.
TestEveryRequestBodyIsDescribedWithTheTypeTheServerTakesandTestEveryStateChangeDescribesTheTypeItDeclarescover them.
8.3 Observability
| What | Where |
|---|---|
delivery_id log attribute | logctx.WithDelivery(ctx, id), set by the worker and the actions; carried over RPC by logctx.Headers; a forbidden hand-written key in .golangci.yml (sloglint) like message_id |
microb2b_delivery_transitions_total{state} | persistence, per transition |
microb2b_delivery_wakeups_total{source=dispatch,sweep,action,catch_up} and microb2b_delivery_wakeups_failed_total | notifier and sweep |
microb2b_delivery_claim_conflicts_total{target} | worker (double wake-ups; expected small) |
microb2b_deliveries{state}, microb2b_deliveries_due (gauges) | sweep |
microb2b_deliveries_stalled_total{target} | sweep |
| Spans | delivery.dispatch, delivery.claim, delivery.attempt, delivery.sweep |
| Log levels | an idle sweep or worker is silent at INFO; a stall, a lease expiry and an operator action are INFO/WARN; a wake-up publish is DEBUG |
telemetry names no module: labels are instance names and states.
9. Tests and risks
9.1 Test strategy
| Level | Tests |
|---|---|
| Pure | DeriveStatus table (every row of 1.3, precedence pairs); MatchRoutes (Any in each position, overlapping routes to one target give one target, other source, no match) checked against Route.Covers on the same fixtures; DeliverySubject refuses ., *, >, space, empty, over-long names; domain types parse/JSON; back-off schedule |
| Persistence (SQLite in memory, and PostgreSQL where the suite runs it) | dispatch writes rows, stage, status and history in one transaction; idempotent DispatchMessage; claim CAS (two claimants, one wins; retrying not due is refused; stalled only with fromStalled); CompleteDelivery with a stale token is ErrClaimLost; SettleDelivery never downgrades; TestStoredStatusIsTheDerivedStatus; sweep steps (lease expiry, back-off, stall at the limit, alert once, alerttest, receipt deadlines); route cache invalidated by own write, by events.routes, by TTL |
| RPC | rpcerrtest.RoundTrip for the vocabulary rows; internal/api/e2e_rpc_errors_test.go covers 409 for both actions |
| Worker | N deliveries run concurrently up to MaxConcurrent and never more; catch-up at start; a duplicate wake-up does one attempt; stop drains a short attempt to delivered, releases a stuck one to retrying (not failed); no attempt after Quiesce |
| Modules | each target’s outcome table of 3.3; AS2 async MDN before CompleteDelivery(AwaitingAck); MDN settles via as2_messages.delivery_id; inbound accepted only after dispatch; stale claim takeover of a dispatched message is a repeat; writer’s content-equal re-attempt |
| End to end, real in-process NATS | poller -> AS2 sender -> AS2 receiver -> writer reaches Delivered on both sides; a route saved while running applies to the next file without restart; no route -> NotRouted; target stopped -> stalled with an alert, start -> delivered; redispatch and redeliver through REST and form (parity: same outcome); a wake-up dropped on purpose (dispatcher wrapper) is recovered by the sweep |
| UI | TestEveryLinkInATemplateIsServed, TestHTMXRequestsDeclareWhereTheAnswerGoes, style pairs, TestEndToEnd_APageOfAMissingRecordIs404 for the delivery path |
Self load test (just test-load) | acceptance: SIGTERM to the sending process mid-run with a backlog (-sigterm-after N), restart: every staged file is one Delivered message on A and one Delivered (written) message on B, none in Received, InProgress, Retrying or Stalled, no AS2 Message-ID received twice by B’s writer, output equal to the input; throughput scales with max_concurrent_deliveries |
| Interop matrices | the validators wait for messages to leave Received, InProgress, Retrying and AwaitingReceipt; inbound success is Delivered (written) |
| Playwright | a message page shows its deliveries; Redeliver on a failed delivery asks for confirmation; Redispatch on a NotRouted message without a new target says nothing happened, and after adding a route routes it |
just test-shutdown | test/graceful-shutdown.sh checks that no delivery is failed after a stop |
9.2 Risks
| Risk | Mitigation |
|---|---|
| R1 Route cache staleness | own writes drop it synchronously; events.routes for other persistence processes; 30 s TTL for bypassing writes; redispatch adds missing deliveries |
| R2 Sweep load on SQLite’s single connection | two indexed queries per 5 s when idle, batch updates of 500 with a pause between batches; the derive read is on the unique index |
| R3 Double wake-ups (dispatch and sweep, catch-up and sweep) | the claim CAS; the loser does nothing and counts a conflict |
| R4 Subject token rules | DeliverySubject refuses anything that is not one identity token; instance names are already IdentityPattern; fixed delivery. prefix |
| R5 Distributed RPC latency on dispatch and claims | the poller’s dispatch replaces InsertMessage (no extra round trip), the claim returns the message; the rpcgen server handles one call per method at a time, which caps claims per second in distributed mode |
| R6 Dual write (commit, then publish) | publish after commit only; the sweep wakes anything whose wake-up was lost within 10 s |
| R7 A stopped or disabled target stalls its deliveries | that is the alert; its next start claims them; MaxDispatches is a setting |
| R8 Re-attempt after an interrupted attempt sends twice | AS2: same Message-ID, the partner’s duplicate detection; writer: content-equal check; poster: the endpoint must be idempotent (documented, as for any retry) |
| R9 Poller crash after commit, before the file moves | the file becomes a second message; listed in the poller’s docs |
| R10 Out-of-order completion with concurrency | documented; nothing in MicroB2B depends on order |
| R11 Cross-module check in the inbound claim takeover | through the core Persistence interface (GetMessage), not SQL across module tables |
10. Where it lives
| Concern | Code |
|---|---|
| Vocabulary: stage, state, status, outcome, refusal, wake-up, derivation, matching | internal/persistence/{delivery.go,message_stage.go,message_status.go,route.go,settings.go,actions.go,sweep.go} |
| Dispatch, claims, transitions, route cache, sweep, actions | internal/persistence/bun/{deliveries.go,persistence.go,routes_cache.go,sweep.go,actions.go} |
| Worker and sweep loop | internal/delivery/{worker.go,sweep.go} |
| Sources | internal/protocol/file/poller.go, internal/protocol/as2/receiver.go |
| Targets | internal/protocol/as2/sender.go (+ mdn_receiver.go), internal/protocol/file/writer.go, internal/protocol/httppost/poster.go |
| UI and REST | internal/api/ui/{routes_messages.go,message_actions.go,templates/message_detail.html}, internal/api/handlers/messages.go, internal/api/openapi/openapi.yaml |
| Graceful stop | internal/dispatcher/nats/nats.go, internal/plugin/manager.go, internal/delivery/worker.go |
| Acceptance | test/selftest-load (-sigterm-after), test/graceful-shutdown.sh |
11. Decisions of the maintainer
- An instance name is unique across plugin types. Routes, messages and deliveries name an instance by its name alone, and the delivery subject is
delivery.<instance>(section 4.3), so the plugin type travels nowhere. The rule is enforced where a name is chosen: saving a plugin instance (form and REST, one rule set) is refused when another plugin type holds the name, with a field error that names that type and instance. Each module lists the instance names it holds, enabled or not, andplugin.InstanceNamesasks them all, so the core names no module (the same shape as the endpoint vocabulary). Seeded and fixture names carry their type (as2_receiver_default,file_writer_default). - The AS2 Message-ID is per message. A delivery is a local concern; the Message-ID is globally unique and shared with the partner, so it is the message id.
as2_messageskeeps its key and the MDN lookup its rule;as2_messages.delivery_id(section 2.5) links the AS2 record to the delivery that sent it. awaiting_acktimes out, per partner, default 24 h. The AS2 remote profile has the receipt timeout (receipt_timeout_hours, 1 to 720, validated like every field of the profile form and REST, one rule set); the AS2 sender passes the deadline when it completes the attempt asAwaitingAck, and the sweep fails the delivery at the deadline (section 5, step 3) with a reason that names the duration.- A message that matches no route raises an alert. The
alert.CodeTablerowmessage_not_routed(warning, subject kind message, link to the message) is raised once by the dispatch step when it settles the message asNotRouted; the content rules apply (ids, party and instance names; no payload). - Repeats and refused inbound messages have their own status and raise no alert.
RefuseMessagetakes aRefusal: the partner’s message turned down on its merits (a repeat or replay, a failed verification, an unknown or unmatched party, over the size limit, no signature where one is required) settles the message asRefused; a failure on our side (storage, database) isErrorand raisesmessage_failed.Refusedis a message status of its own (label “Refused”, its own badge and filter value;logStateChangeraises no alert for it; the dashboard counts it apart from failures;telemetry.Statusknows it). The history line carries the reason.