Publisher and SyncInTxn

Package: publish

After events are appended, the repository hands them to a publisher. go-sequent has one default mode and one optional add-on.

SyncInTxn (default)

publish.SyncInTxn calls registered handlers inside the write transaction. For each event, in order, it runs the projectors registered for that event type, then the workflows. If any handler returns an error, the error aborts the transaction and nothing is stored.

type Handler interface {
    Handles() []string
    Handle(ctx context.Context, e event.Event) error
}

Register handlers through the façade:

app.RegisterProjector(balanceProjector) // included in projection rebuild
app.RegisterWorkflow(notifyWorkflow)    // excluded from projection rebuild

Why this is the default: a ledger balance must be correct as soon as the command returns. When the projector runs in the same transaction as the append, there is no window where the event exists and the balance does not. That property is read-your-writes. Ledger projections must use SyncInTxn.

The cost is that projector work is on the write path. Keep ledger projectors cheap and deterministic, and put heavy or external work elsewhere.

Ordering

For one Publish call, events arrive in aggregate load order, then sequence order within each aggregate. For each event, projectors run before workflows. Handlers for the same event run in registration order.

Failures

A projector error is wrapped as publish: projector for <EventType>: .... A workflow error is wrapped as publish: workflow for <EventType>: .... Either one rolls back the entire Execute.

PublishProjectorsOnly

PublishProjectorsOnly runs projectors and skips workflows. The projection rebuild pipeline uses it so that replaying history does not send notifications or enqueue commands a second time.

WithOutbox

For asynchronous fan-out, wrap the publisher with publish.WithOutbox:

ob := outbox.NewMemory() // or outbox.NewPostgres(pool)
pub := &publish.WithOutbox{
    Inner:  app.Publisher,
    Writer: ob,
    Topic:  "events",
    // EventTypes: map[string]bool{"bank.MoneyDeposited": true}, // empty = all events
}
app.Repo.Publisher = pub

Publish first runs the inner SyncInTxn handlers, then enqueues one outbox message per matching event, as JSON. With the Postgres writer the message row is inserted in the write transaction. So the event, the projections and the outbox row commit or roll back together.

The transport is a Postgres table that a poller drains. There is no LISTEN/NOTIFY, Kafka or other broker in the framework.

Use the outbox for:

• External systems that must see an event after commit.
• Cross-service integration.
• Projections that are too heavy for the write latency budget and are not ledger projections.

Do not use it for ledger balances.

Draining the outbox

poller := &outbox.Poller{
    Store:      ob, // outbox.Store: Claim, Ack, Nack, Depth
    Handler:    func(ctx context.Context, msg outbox.Message) error { /* deliver */ return nil },
    BatchSize:  10,
    Interval:   time.Second,
    LockFor:    30 * time.Second,
    RetryAfter: 5 * time.Second,
}
err := poller.Run(ctx) // or RunOnce(ctx)

Claim locks a batch, the handler delivers each message, and the poller Acks on success or Nacks with a retry delay and the error text. Delivery is at-least-once. Make consumers idempotent. Message carries Topic, Payload, TenantID, AggregateID, EventType, Attempts and timestamps.

The CLI has a polling command:

go run ./cmd/go-sequent outbox -dsn "$GO_SEQUENT_PG_DSN" -once
go run ./cmd/go-sequent outbox -dsn "$DSN" -interval 1s -batch 10

The bundled CLI handler only prints each message. It does not know your delivery target. For real delivery, run your own outbox.Poller in your service with a handler that calls the target system.

Store.Depth returns the pending count. Feed it to obs.Config.OutboxDepth for the sequent.outbox.depth gauge. See OpenTelemetry.

Choosing between them

Related

• Projector
• Workflow
• Projection rebuild

Docs track the go-sequent library. When APIs drift, the source wins.