Patterns

Everything above publishing and consuming lives in one package:

import "github.com/AceMQ-Company/acemq-go-amqp/patterns"

It is separate because none of it is needed to send a message, and a core that carries everything is a core nobody can read.

Request and reply

The short version. Request and reply is the whole page: timeouts against context cancellation, what a responder failure does to the request, the reply queue, and what is not measured.

requester, err := patterns.NewRequester[PriceRequest, PriceResponse](ctx, mq, "", "prices")
defer requester.Close()

response, err := requester.Do(ctx, PriceRequest{SKU: "abc"})

And the other side:

responder, err := patterns.Serve(ctx, mq, "prices",
    func(ctx context.Context, m acemq.Message[PriceRequest]) (PriceResponse, error) {
        return price(ctx, m.Payload)
    })
defer responder.Close()

Replies are paired with requests by correlation identifier, so many can be in flight at once. There is a test that runs twenty concurrently and checks each caller gets its own answer.

This is a synchronous shape drawn on an asynchronous system, and that costs something. A caller blocked on a reply holds a goroutine, a connection and a deadline; a queue that backs up turns into a service that stops responding. Reach for it where a caller genuinely cannot proceed without the answer, and publish an event otherwise.

A responder that returns an error sends it back rather than swallowing it, so a blocked caller learns it failed instead of waiting out the timeout:

acemq: the responder failed: no such product

ErrRequestTimedOut says nothing about whether the request was handled. A timeout is the absence of an answer, and the work may well have been done — which is why a request that changes anything should be idempotent.

The reply is published without mandatory, deliberately. A generated reply queue is auto-deleting and goes away with a requester that gave up, and a reply nobody is waiting for is not a message worth keeping. A reply that reaches no queue is therefore dropped, the request is accepted, and the requester — if it is still there — sees ErrRequestTimedOut.

By default replies come back on an exclusive, auto-deleting queue that goes away with the process. It is a classic queue and has to be: RabbitMQ allows a quorum queue to be neither exclusive nor auto-deleting. patterns.ReplyTo("queue") names one that survives a restart, and a named one is durable, so it is quorum like any other durable queue — see quorum, classic and streams.

Where the reply address travels

Write both, read either. A request carries the reply queue twice, in the same two places in all five AceMQ libraries:

patterns.Serve reads the header first and falls back to the property when the header is absent, so it answers a request from any of the five. Requester.Do writes both, to the same queue, so any of the five can answer one of ours.

Header first because it is the half that survives a rebuild: a service that reconstructed the message kept the headers and lost the properties, so where the two disagree the header is the more recent of the two.

Publishing a request by hand rather than through a Requester means writing them yourself:

err := acemq.NewPublisher[PriceRequest](mq, "", "prices").
    Send(ctx, request,
        acemq.ReplyTo("my-replies"),
        acemq.Header(patterns.HeaderReplyTo, "my-replies"))

A request carrying neither is dead-lettered rather than retried: retrying cannot make a reply address appear.

Idempotency

store := patterns.NewInMemoryIdempotencyStore(time.Hour)

sub, err := acemq.Consume(ctx, mq, "orders",
    patterns.Idempotent(store, handle))

A duplicate is accepted, not rejected: the work was done, so the message has been handled, and dead-lettering it would raise an alarm about something that went right.

When the handler fails the key is forgotten, so the retry actually runs. Remembering a message that then failed would mean the retry silently does nothing and the message is dropped after appearing to succeed.

Claimed, duplicate, or in progress

A redelivery finds the key in one of three states, and each gets a different answer. This is the same rule in all five AceMQ libraries:

In progress is a handler still running elsewhere, or one that failed and could not release its claim because the store was unreachable. Until 0.9.5 Go treated it as a duplicate and acknowledged it, and in the second case the message was lost: the claim outlived the failure, and the retry was accepted as a duplicate of work that never happened.

Now Idempotent answers acemq.InProgress(delay). The consumer waits delay (five seconds by default), republishes the original body and envelope to the same queue, and acknowledges the original. If the republish is not routed, the original is nacked and requeued instead. The attempt is not incremented and the retry policy is not consulted, so a message is never dead-lettered for being in progress. Once the lease ends, the next look claims it and runs the handler. It is counted as acemq.consume.total{outcome="in_progress"}, not as retried or dead-lettered. A stream consumer refuses it as it refuses a retry.

patterns.Idempotent(store, handle, patterns.InProgressDelay(2*time.Second))

Keep the delay well under the store's claim timeout. store.TryClaim(ctx, key) returns the three-way patterns.ClaimResult. Both shipped stores implement it through the patterns.ClaimReporter interface, and Claim keeps its bool. A custom store that implements only IdempotencyStore behaves as before: a refused claim is read as a duplicate.

This is a guard against duplicates, not a guarantee of exactly-once. Between the handler finishing and the acknowledgement reaching the broker, a crash still leaves a message that will be delivered again. Only a store written in the same transaction as the work closes that gap.

NewInMemoryIdempotencyStore is for tests and for one worker. With two, each has its own memory and both will believe they are first.

patterns.IdempotentBy takes a key from the payload, for when two different messages are about the same thing.

The outbox

A service that writes to a database and then publishes has two things that can fail independently, and the gap between them is where messages are lost or invented. Writing the message into the same transaction as the work removes the gap:

tx, _ := db.Begin()
placeOrder(tx, order)

record, _ := patterns.Record(mq, "orders-events", "order.placed", event)
store.Add(ctx, record)      // the same tx

tx.Commit()

A relay publishes what was committed:

relay := patterns.NewOutboxRelay(mq, store)
relay.Start(ctx)
defer relay.Close()

A record is removed only after the broker confirms, so a crash in between republishes it. That is at-least-once by construction — which is why consumers of anything sent this way need to be idempotent. Removing first would lose messages instead, and a lost message is worse than a repeated one.

An OutboxStore is only worth having if Add can join the caller's transaction. InMemoryOutboxStore cannot, and so has none of the property the pattern exists for; it is there for tests and for seeing the shape.

The relay publishes the bytes that were recorded rather than re-encoding, because a record outlives the process that wrote it and the Go type may not survive a deployment.

It publishes them mandatory, as Java's relay does. A record nothing is bound to receive — the service that wants the event has not declared its queue yet — is a failed publish and stays in the outbox, rather than being dropped by the broker and marked as sent. Up to 0.9.5 it was the second, and the event was gone. The same holds on a connection dialled WithoutConfirms: the relay publishes on the library's own confirm channel, so a record is marked sent only once the broker has confirmed it reached a queue.

A record the broker will never take

A sweep stops at the first record that fails and leaves it in the outbox. That is right — the writer chose the order, and skipping ahead would deliver a later message before an earlier one — but it means a record that can never be published holds up everything behind it on every tick, for ever. An exchange somebody deleted, or a payload a policy will always refuse: one row, and the outbox stops.

So a failure is counted against the record. Both stores keep an Attempts and a LastError on it, and once Attempts reaches the store's maximum — ten by default, the same ten Python and Ruby use — Pending stops offering the record and the queue behind it moves again.

store.SetMaxAttempts(10)          // the default; shorten it in a test
...
stuck, _ := store.Retired(ctx)    // the in-memory store's takes no ctx
for _, record := range stuck {
    log.Printf("nobody could publish %s: %s", record.ID, record.LastError)
}

The record is kept, not deleted. One nothing could publish is evidence: somebody has to read it, fix whatever refuses it, and release it by putting its attempts back to zero. Deleting it would be the silent loss this pattern exists to prevent, reached from another direction. Nothing but Retired shows these rows, because Pending exists to skip them — so graph acemq.outbox.total{outcome=failed} and read Retired when it climbs.

RecordFailure is deliberately not part of the OutboxStore interface: adding a method would stop every custom store compiling. The relay type-asserts for it, so a store without one keeps the behaviour it has — blocking at a permanently failing record. A custom store should add RecordFailure(ctx, id, cause string) error and leave records at the limit out of Pending.

The SQL table gained two columns in 0.9.0. A table created before it needs the migration, and Pending fails on the missing column until it has run. Both are additive with a default, so the old code is unaffected by them and the migration can — and should — go out before the deployment does:

ALTER TABLE acemq_outbox ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0;
ALTER TABLE acemq_outbox ADD COLUMN last_error TEXT;

What a relay reports

Every sweep counts what it did through the connection's acemq.Observer: acemq.outbox.total tagged published or failed, and acemq.outbox.lag — how long a record waited between being committed and being published, measured from the commit rather than from the sweep, because what a lag answers is how long somebody has been owed this message. The tracing adapter's outbox.publish_failed event is not written from here and cannot be: a sweep runs on a goroutine of the relay's own with no span open. See observability.

Ordering

A queue delivers in order; acemq.Concurrency above one stops honouring that. Usually the right trade, and the wrong one where later messages about the same entity must not overtake earlier ones.

sub, err := acemq.Consume(ctx, mq, "orders",
    patterns.Ordered(patterns.ByHeader[OrderEvent]("x-order-id"), handle),
    acemq.Concurrency(16))

Messages sharing a key are handled one at a time; different keys still run at once. Both halves are tested — ordering per key that became ordering overall would just be concurrency turned off.

What it does not do. It orders messages already delivered to this process. It cannot reorder what the broker delivered out of order, and with several consumers on one queue it orders only within each. Ordering across consumers needs the messages to reach the same consumer, which is a routing decision:

key := patterns.PartitionedRoutingKey("orders", order.ID, 8)  // "orders.3"

Pipelines

Declaring one

A flow with named steps, each its own queue and its own consumers:

pipeline, err := patterns.NewPipeline[Order](ctx, mq, "fulfilment",
    patterns.PipelineStep("validate", validate),
    patterns.PipelineStep("reserve", reserve,
        patterns.StepConsumers(4),
        patterns.StepRetry(acemq.ExponentialRetry(4, time.Second, time.Minute)),
        patterns.StepDescribedAs("hold stock for 15 minutes so payment cannot oversell")),
    patterns.PipelineStep("dispatch", dispatch))
if err != nil {
    return err
}
defer pipeline.Close()

runID, err := pipeline.Send(ctx, Order{ID: "A-1"})

A step's handler takes what the previous one produced and returns what the next one consumes:

func reserve(ctx context.Context, m acemq.Message[Order]) (Reservation, bool, error) {
    if m.Payload.Digital {
        return Reservation{}, false, nil    // nothing to reserve; the run stops here
    }
    return stock.Reserve(ctx, m.Payload)
}

NewPipeline declares the exchange, one queue per step and the bindings, then starts the consumers — so a pipeline that returns without an error is one whose topology exists and whose steps are running. Declaring is idempotent and every service that declares the same pipeline agrees about it, which is what lets two services run different steps of one flow.

Each step gets {pipeline}.{step}, so a slow stage is a deep queue you can see and scale on its own with StepConsumers. A single consumer doing all four in sequence gives you none of that.

The topology is Java's, to the letter: exchange named for the pipeline and declared direct, queue {pipeline}.{step}, binding key the step's name, and the declared route carried on the message as x-acemq-route plus a position. So a Go-declared pipeline can be continued by a Java, .NET, Python or Ruby service, and a Go step can be one hop of a pipeline any of them declared — see Being one step of a Java pipeline for that direction, which needs no Pipeline at all.

What Go cannot do, and says instead. Java's builder threads the types through at compile time: a step producing a Reservation can only be followed by one consuming a Reservation. Go's generics cannot express that — a method cannot introduce type parameters, so a fluent chain cannot carry the previous step's output type forward. Each step is built on its own and NewPipeline checks the chain with reflection when it assembles it:

acemq: pipeline "fulfilment" step "reserve" produces patterns_test.Reservation
but the next step "dispatch" consumes patterns_test.Order

A worse guarantee than Java's, and the honest one: it fails at start-up rather than on the first message, and it names both steps and both types.

Nothing here logs. pipeline.Describe() returns the one-line summary — steps and their descriptions — for you to log where you log things, because a library that picks a logger picks it for every application that imports it.

Wrapping one handler

handler := patterns.Chain(handle,
    patterns.WithLogging[OrderPlaced](log.Printf),
    patterns.WithTimeout[OrderPlaced](10*time.Second),
    patterns.WithIdempotency[OrderPlaced](store))

The first named is outermost, so logging records what the others decided.

WithTimeout reports a retry whenever the deadline passed, whatever the handler says about itself. Between retrying work that may have succeeded and accepting work that may have failed, duplicates are a problem you can solve and a lost message is not. It cancels the context, so a cooperative handler stops promptly; one that ignores its context will not stop, and no wrapper can make it.

patterns.Then publishes the result onwards, carrying the correlation and recording causation:

step := patterns.Then(
    acemq.NewPublisher[Shipment](mq, "shipping-events", "shipment.requested"),
    func(ctx context.Context, m acemq.Message[OrderPlaced]) (Shipment, bool, error) {
        if m.Payload.Digital {
            return Shipment{}, false, nil    // nothing to ship
        }
        return Shipment{OrderID: m.Payload.OrderID}, true, nil
    })

The input is accepted only once the output is published. If publishing fails the input is retried and the work runs again, so a step that changes anything should be idempotent.

Then's hop is always published mandatory, whatever the publisher you give it was built with, as are a declared pipeline's and FollowSlip's hops: a next step nothing is bound to is a failed hop, retried and then dead-lettered, never accepted. The publisher itself is not changed; your own Send calls on it keep the mandatory setting you gave it.

Every hop the library makes for you — Then's included — is also published with publisher confirms, even on a connection dialled WithoutConfirms, because a return is only known once the broker has confirmed. See Publisher confirms.

Naming a step, so a finished run is visible

patterns.InPipeline and patterns.AtStep name a Then or a FollowSlip, and that is what makes it write a pipeline.run_finished event onto the delivery's span:

step := patterns.Then(shipments, plan,
    patterns.InPipeline("fulfilment"), patterns.AtStep("pick"))

A step that returns false reports ended_early — this message does not continue, a decision rather than a failure, and otherwise a thing only a log could tell you. FollowSlip reports completed when it finishes the itinerary, and takes the step's name off the slip so it usually needs only InPipeline. An unnamed step reports nothing at all, rather than an event tagged with two empty strings.

Replay

What somebody actually does at three in the morning: a dead-letter queue has two thousand messages, the bug is fixed, and they need to go back — but not all of them, and not silently.

result, err := patterns.Replay(ctx, mq, patterns.ReplayFrom{
    Queue:    "orders-dead",
    Exchange: "orders-events",
    Limit:    500,
    Filter: func(env acemq.Envelope, body []byte) bool {
        return strings.Contains(env.Error, "timeout")
    },
})
// moved 500, skipped 31 (limit)

The reason is part of the answer: "moved 500" means something quite different when the limit was 500.

Messages the filter declines stay where they are. Each replayed message is stamped acemq-replayed-from, acemq-replayed-at and acemq-replay-count, so a consumer that needs to treat them differently can.

Those are the names Java, Python and Ruby write too, and none of them begins x-acemq-. That namespace is the engine's, and a header in it that the engine does not put on the envelope is dropped before a handler sees it — so a replay stamp written there would arrive on the wire and vanish. Unprefixed, the three come through in Envelope.Headers. acemq-replayed-at is RFC 3339, not the epoch milliseconds x-acemq-first-seen uses.

.NET is the outlier here. It writes x-acemq-replayed-from, x-acemq-replayed-at and x-acemq-replay-count, and a Go consumer will not see them: the reserved prefix means this library drops all three on the way in. A message replayed by a .NET operator arrives, and it arrives without its stamps. That costs an audit trail rather than a message, and only for a replay somebody did by hand — but do not write a Go consumer that expects to be able to tell a .NET-replayed message from an original.

Attempts start again. A replayed message goes back on attempt one with the reason it was dead-lettered cleared, because a message dead-lettered on the last attempt of a five-attempt policy would otherwise be dead-lettered again before any handler saw it — two thousand messages moved from one queue to the same queue. KeepAttempts: true puts back exactly what was there, for an audit or for a queue read by something that counts attempts itself.

How it knows when to stop. It reads the queue a message at a time and stops when the broker has nothing left to hand over.

Messages the filter declines are held unacknowledged until the pass ends, and returned together at the end. Returning one immediately does not work: RabbitMQ puts a returned message back where it was, at the head of the queue, so the next read is handed the same message again and everything behind it is never looked at — a selective replay that silently stops at the first message it declines. Holding them takes them out of the way for the length of the pass, and the broker still has them, so a replay that dies half way through returns them rather than losing them.

A message is acknowledged only after the broker has confirmed the new copy. A crash in that gap replays it twice, which is the right way round for a dead-letter queue.

The new copy is published mandatory. A destination nothing is bound to is confirmed by the broker and then dropped, so until 0.9.5 a replay into it acknowledged the last copy of the message it was asked to recover. Now that message is returned to the queue it came from, the replay stops with Reason failed, and the error is a *acemq.PublishFailedError with Unroutable set.

Bodies are read as bytes, not through the connection's codec. A body that will not decode is exactly the kind that ends up in a dead-letter queue.

Routing slips

An itinerary the message carries, instead of a central orchestrator:

slip := patterns.NewRoutingSlip().
    Then("orders-events", "order.validate", "validate").
    Then("orders-events", "order.charge", "charge").
    Then("orders-events", "order.ship", "ship")

err := patterns.Start(ctx, mq, slip, order)

Each service does its part and the message goes to the next stop:

sub, err := acemq.Consume(ctx, mq, "charge-queue",
    patterns.FollowSlip(mq, func(ctx context.Context, m acemq.Message[Order]) (Order, error) {
        return charge(ctx, m.Payload)
    }))

The route is decided once, by whoever started the work, and travels with the message rather than living in a component every service has to talk to.

What it costs: no single place knows the whole route at runtime, so a route that is wrong is discovered one hop at a time. Worth it when the steps vary per message; not worth it when every message goes the same way, where a fixed chain of consumers is simpler to follow.

Two forms on the wire, and both are read

The family writes a slip in two shapes. Go reads either and writes the JSON one unless asked otherwise.

The JSON slip is self-describing: a message carries its whole itinerary, so any consumer can send it onwards with nothing declared anywhere, an operator reading a dead-letter queue can see where it was going, and the route can differ per message — which is the reason to use a slip rather than a fixed chain at all.

The declared form is smaller and stays readable in a management console — validate,charge,ship at position 1 says where a message is without anybody decoding anything — at the cost of meaning nothing without the declaration.

A slip keeps the form it arrived in. A Go step in a Java-declared pipeline answers in the shape the next Java step is looking for, without being told to.

Being one step of a Java pipeline

Java's Pipeline names its steps and resolves them against a declaration: the pipeline's name is the exchange, a step's name is the routing key, and the queue behind a step is {pipeline}.{step}. Declare the same route in Go and a Go consumer is one of its steps:

route := patterns.NewRoute("orders", "validate", "charge", "ship")

sub, err := acemq.Consume(ctx, mq, route.QueueFor("charge"),
    patterns.FollowSlip(mq, func(ctx context.Context, m acemq.Message[Order]) (Order, error) {
        return charge(ctx, m.Payload)
    }, patterns.AlongRoute(route)))

The Go step publishes to the orders exchange keyed on the next step's name with the position advanced and the run identifier carried through — exactly what the Java step after it is waiting for.

patterns.AlongRoute also names the pipeline for run reporting, so patterns.InPipeline is only needed to call it something other than the route's own name.

Without AlongRoute, a declared route is refused rather than followed. The names carry no destinations, so publishing with an empty exchange would send the message to a queue named for the step instead of to the pipeline's — quietly, and to the wrong place. The rejection is fatal, because a missing declaration will still be missing on the fourth attempt, and it names AlongRoute in the error.

To start a run that Java steps will follow, use the route:

slip := route.Start()                  // at validate, with a fresh run id
err := patterns.Start(ctx, mq, slip, order)

slip.AsSteps(route) converts a hand-built slip to the declared form, and slip.AsJSON() converts one back — neither is a one-way door.

Claim check

A payload too large for a broker goes to a store, and the message carries the key:

store := patterns.NewFilesystemClaimCheckStore("/mnt/claims")
checked := patterns.ClaimCheck(acemq.JSONCodec{}, store)

mq, err := acemq.Connect(ctx, url, acemq.WithCodec(checked))

A scanned medical report is tens of megabytes. Putting it on a queue is possible and is a mistake: it fills the broker's memory, it is copied to every bound queue, it makes a dead-letter queue impossible to inspect, and it turns a broker into a filesystem with worse tools.

Only above the threshold

Below patterns.DefaultClaimCheckThreshold — 64 KiB — the payload travels inline, exactly as it would without the codec. That matters more than it sounds: offloading a two-hundred-byte message turns one broker round trip into a store round trip and a broker round trip, so an unconditional claim check makes the common case slower to fix the rare one.

patterns.OffloadAbove(n) changes it; zero offloads everything.

What is on the wire

0xAC  0x01  0x00  payload   inline, and identical to what the delegate wrote
0xAC  0x01  0x01  key       a claim check

Three bytes, and the third is how a consumer decides whether it is holding a payload or a reference to one. The framing rather than a header is the contract: a header can be stripped by a shovel or a federation link, and the body cannot — and a header that is present or absent cannot say whether a payload travelled inline, which is what makes the threshold changeable without a flag day.

These are the same three bytes Java, Python and Ruby write, and the key is the store's key as bare UTF-8 — not a URI, not a scheme, nothing wrapped around it. A Go consumer pointed at the same store reads a document a Java publisher checked in.

A body with no framing is read as the delegate would read it, which is what makes adding this codec to a live queue safe.

The content type is the delegate's, unchanged. A claim-checked message is still a document; it is a document that is somewhere else.

x-acemq-claim is not written by this codec. That header is reserved for an application that wants an operator reading a dead-letter queue to see where a payload went. Python and Ruby reserve it the same way. To read the key out of a body without fetching it — which object does this need, and is it still there? — use patterns.ClaimKeyOf(body).

The store

type ClaimCheckStore interface {
    Put(content []byte) (string, error)
    Get(key string) (content []byte, found bool, err error)
    Delete(key string) error
}

Three methods, so a store in front of S3 or Azure Blob Storage is a small type. Nothing in it knows about messaging.

Get tells not found from failed, and the codec treats them differently: a store that timed out may well answer next time, so that is retryable; a key the store does not hold never will be, so that is fatal and the message stops rather than circling. The store failure comes back wrapped in acemq.Retryable, and the consumer retries such a decode error through the retry policy. Until 0.9.5 every decode error was parked, the retryable one included.

Two are supplied. patterns.NewInMemoryClaimCheckStore() is for tests — the payloads are in the publisher's own memory, so a consumer in another process finds nothing, which is the pattern failing rather than working. patterns.NewFilesystemClaimCheckStore(dir) is a real store where the filesystem is shared and durable; writes go to a temporary file and are renamed into place, because a consumer fast enough to read the key before the writer finished would otherwise get a truncated payload.

There is no context on the store: acemq.Codec has none to pass on. A store doing network I/O must carry its own timeout, because without one a hung object store hangs a consumer's whole delivery.

Retention is the part that goes wrong

The store and the queue have different lifetimes and nothing enforces a relationship between them. A message replayed a month later carries a key, and if the store expired that key the replay produces a message nobody can read — worse than a lost message, because it looks like a message and fails deep inside a consumer rather than visibly.

The store's retention must exceed every retention that could bring a message back: queue TTLs, dead-letter queues, and however long somebody might sit on a message before replaying it by hand. When in doubt, longer.

Delete is never called by the codec. Deleting on read would break a second consumer of the same message and deleting on acknowledgement would break a replay, so when a payload may be removed is a retention decision — and those belong to whoever owns the data.

A key becomes a path segment in the filesystem store, and a key arriving from a message is whatever a publisher put there. Every key the store issues is a UUID, so one that is not is refused rather than followed; ../../etc/passwd is a key too.

Sagas

Work that spans services, where a database transaction is not available and the alternative — doing half of it and hoping — is how a customer ends up charged for an order that was never placed:

saga, err := patterns.NewSaga("place-order", []patterns.SagaStep[*Order]{
    {Name: "reserve-stock", Do: reserveStock, Undo: releaseStock},
    {Name: "take-payment", Do: takePayment, Undo: refund},
    {Name: "confirm", Do: confirm},
})

result := saga.Run(ctx, order)

Steps run in order. When one fails, the steps that already completed are undone in reverse order, because that is the order the world was changed in and a compensation often depends on state a later step has not yet altered.

Undo is optional, and a completed step without one is skipped rather than treated as an error — a step that only read something needs no undo. It is a mistake on a step that changed something, and nothing here can tell the two apart, which is the argument for writing Undo first and Do second.

Run returns a result rather than an error. A failed saga is not an exceptional condition to a caller that has to decide what happens next, and the interesting part is not the failure anyway:

if result.HasUnresolved() {
    alert("could not undo: %v", result.Unresolved)
}

Unresolved is the list to alert on. A compensation that fails does not stop the others — stopping would leave more undone than continuing — so it is reported, its step collected, and the rest still run. What ends up in Unresolved is real-world effects that happened, were meant to be undone, and were not. Nothing else in the system knows about them and no retry resolves them; a person has to.

The compensations run with the context Run was given. If the failure you expect is a cancellation, hand them one that outlives it with context.WithoutCancel, or every compensation lands in Unresolved.

A step that panics is a step that failed, and everything before it is still compensated: unwinding past the compensation is the one outcome this pattern exists to prevent.

Nothing here touches a broker. It is in-process coordination, so a Go saga and a Java one are the same idea rather than two ends of one conversation. What matches across the libraries is the behaviour, not a wire format.

Consumer groups

group, err := patterns.NewConsumerGroup(ctx, mq, "orders", 4, handle)
defer group.Close()

Several consumers over one queue, stopped together.

Concurrency or a group? acemq.Concurrency runs several handlers on one consumer and one channel. A group runs several consumers, each with its own channel and prefetch. Use a group when handlers are slow enough that one channel's prefetch is the limit, or when a fair share across processes matters — the broker round-robins between consumers, so four here compete evenly with four in another instance.

A group that cannot start closes what it already started, because a half-started group holds messages nothing is going to handle.

Schema registry

registry := patterns.NewInMemorySchemaRegistry()
schema, err := registry.Register(ctx, "order.placed", "avro", definition)

Messages carry a small identifier and consumers look the schema up, which is what lets a producer add a field without every consumer being redeployed the same afternoon. The Avro codec uses it directly.

Definitions are fingerprinted, so a service registering on start-up does not add a version per restart. A lookup that finds nothing returns ErrSchemaNotFound rather than a zero value — a consumer that cannot find its schema has a real problem and must not carry on with an empty definition.

NewInMemorySchemaRegistry shares nothing between processes, which is the whole point of a registry. Use the SQL one, or Confluent's, for anything real.

Stores that survive a restart

The in-memory idempotency store and outbox are for tests and for one worker. The SQL ones are the point:

tx, _ := db.BeginTx(ctx, nil)
placeOrder(ctx, tx, order)

record, _ := patterns.Record(mq, "orders-events", "order.placed", event)
patterns.NewSQLOutboxStore(tx, patterns.PostgresDialect).Add(ctx, record)

tx.Commit()

They take anything with ExecContext and QueryContext, so *sql.Tx satisfies them and the record commits with the work. There is a test that rolls a transaction back and proves neither the order nor the message survived.

Schema() returns the DDL rather than running it. A library that runs migrations against your database on start-up is a library fighting your migration tool — put it in a migration.

Dialects cover Postgres, MySQL and SQLite: placeholder syntax and the spelling of "insert unless it is already there" genuinely differ, and getting either wrong fails at runtime on one database and not another.

MySQL also reserves key, the idempotency table's column, so MySQLDialect quotes it with backticks (Dialect.Quote) and its no-op upsert names that column (Dialect.InsertIgnoreOn). Its Schema() declares the index inside the table and uses DATETIME(6). The Postgres and SQLite schemas are unchanged, so no migration is needed there; on MySQL the store did not work before 0.9.5, so there is no older table to migrate.

Streams

A stream keeps its messages after they are read, so several consumers can each read from their own position:

err := patterns.DeclareStream(ctx, mq, "events", patterns.StreamRetention{
    MaxAge: 7 * 24 * time.Hour,
})

sub, err := patterns.ReadStream(ctx, mq, "events", handle,
    patterns.StreamOptions{Offset: patterns.FromFirst(), Prefetch: 100})

It is not a queue with a longer memory. Acknowledging advances this consumer's position rather than removing anything, and rejecting dead-letters nothing, because there is nothing to remove it from. A message that cannot be handled has to be dealt with by the handler — logged, copied elsewhere, counted — and the stream moves on regardless. Nothing is lost, and nothing is retried for you: ReadStream refuses acemq.Retry, because on a stream a retry appends a second copy of the message to the log.

Retention is unbounded by default, which for a stream means "until the disk is full". Set MaxAge or MaxBytes on anything that runs for long.

Streams is the whole page: where to start reading, checkpointing, why acemq.Retry is refused on a stream, and what the in-memory transport will not do.

Delayed delivery

Deliver a message later:

scheduler, err := patterns.NewScheduler(ctx, mq)
defer scheduler.Close()

scheduler.In(ctx, 4*time.Hour, "billing", "invoice.due", invoice)
scheduler.At(ctx, renewal, "policies", "policy.renew", policy)

Why not a per-message time to live

The obvious implementation is to set an expiration on the message, drop it in a queue nobody consumes, and let it dead-letter to its destination. It is what most articles suggest and it is wrong for anything but a single fixed delay, because a classic queue expires messages only at its head. Put a four-hour message in, then a one-minute message behind it, and the one-minute message is delivered in four hours. Nothing reports it: the queue looks healthy, the message is not lost, it is simply late by a factor nobody predicted — and it fails in production under mixed load rather than in testing under uniform load.

What this does instead

A ladder of queues, each with a uniform time to live, and a message hops through them until it is due:

Every message in a rung has the same delay, so the head is always the one due soonest and head-of-line expiry is harmless. Each rung is classic and carries exactly three arguments — the time to live above, x-dead-letter-exchange of acemq.schedule and x-dead-letter-routing-key of acemq.schedule.due — and is bound to the acemq.schedule direct exchange under its own name. An expired message lands on acemq.schedule.due, where the scheduler either delivers it or puts it on the largest rung that does not overshoot what is left. A one-day delay is twenty-four hops and a one-minute delay is one, which is the right way round.

Both hops — onto a rung, and out to the destination — are published mandatory. A rung somebody deleted, or a destination nothing is bound to, fails the hop rather than being confirmed and dropped. Up to 0.9.5 such a message was acknowledged and gone.

A message that is due but reaches no queue — the target has nothing bound for its routing key yet — is kept: it goes back through the ten-second rung and is tried again every ten seconds until it is delivered, with no attempt limit, as the Java, .NET and Ruby schedulers do. At or In for a moment already past keeps it the same way and returns no error. Delivered() counts only messages that arrived, so Hops() climbing while Delivered() stands still is the sign of a target nobody has bound. Only when the message can be put on no rung at all — a rung somebody deleted — does the control consumer set it aside on acemq.schedule.due.dlq with the reason, and At or In return the error. Up to 0.9.6 an unroutable due message went to acemq.schedule.due.dlq on the first attempt.

patterns.ScheduleTopology() is the whole thing, exported so a deployment can declare it up front or compare it with what another AceMQ library declares:

fmt.Println(patterns.ScheduleTopology())

Those names and arguments are a cross-language contract, not a preference. Java, .NET, Python, Ruby and this library declare the same objects, and a queue redeclared with a different argument table is answered PRECONDITION_FAILED — the second service to start cannot consume at all.

What travels with the message

Four headers, and deliberately not in the reserved x-acemq- namespace: that one is the engine's, and a header carrying it is dropped from the application's view on the way in, so a scheduler header using it would be written on publish and gone on consume.

The content type is carried because the scheduler republishes bytes rather than values, and a consumer picks its codec from it. Publishing pre-encoded bytes under application/octet-stream produces a message the intended consumer cannot decode: it arrives, it is the right bytes, and nothing can read it.

None of the four reaches the consumer. They are bookkeeping, and a consumer depending on them would be depending on how a message got to it.

What it costs

A long delay is several broker round trips rather than one, and delivery is accurate to about the smallest rung rather than to the second — the last hop is skipped when under a second is left, because another hop would cost more than the accuracy it buys. Something that must fire at 09:00:00.000 exactly is a scheduler, not a message broker.

The alternative is RabbitMQ's delayed-message-exchange plugin, which does this properly and is a plugin — so it is not available everywhere, and a library that silently required it would be a library that works on your laptop.