Lesson 6 · Publishing, consuming & delivery

Publishing

Every message leaves wrapped in an envelope — carrying not just the payload, but the tenant and the trace — and stamped with an id that makes duplicates harmless.

Your win: explain what happens when a service publishes — the DataInMessage envelope, the dedup id, and the sync/async/traced variants — and read a real publish.

Four ways to publish, one envelope

The wrapper exposes four publish methods — sync vs async, plain vs traced — but they all do the same core thing: wrap your bytes in the DataInMessage envelope before sending:

internal/golibs/nats/jetstream.go:507-528 (PublishContext, condensed)

userInfo := golibs.UserInfoFromCtx(ctx)          // pull tenant + user from the context
p := npb.DataInMessage{
    Payload:      data,                          // ← YOUR proto bytes go here
    ResourcePath: userInfo.ResourcePath,         // ← the tenant, riding alongside
    UserId:       userInfo.UserID,
}
payload, _ := proto.Marshal(&p)
opts = append(opts, nats.MsgId(idutil.ULIDNow()))   // ← a dedup id on EVERY message
js.Publish(subject, payload, opts...)

The envelope is the whole trick Your business event goes in Payload, but the envelope also carries the ResourcePath (the tenant) and UserId, pulled from the request context. Why? Because NATS is asynchronous — the subscriber runs later, with no request context of its own. By packing the tenant into the message, the consumer can rebuild it (Lesson 7) — which is how tenant-isolated (RLS) queries keep working across the hop. The traced variants add one more field, TraceInfo (the B3 trace context), so distributed traces span the async gap too.

The dedup id — publish-side idempotency

Notice nats.MsgId(idutil.ULIDNow()) on every publish. That sets the Nats-Msg-Id header to a fresh ULID. JetStream keeps a deduplication window (default 2 minutes): if it sees a message whose Nats-Msg-Id it already stored, it drops the duplicate and re-acks the original.¹ So if a publish is retried after a flaky network (below), the same id means the stream won't store it twice. Idempotency at the source.

Sync vs async vs traced

Method	What it does
`PublishContext`	synchronous — waits for the server's `PubAck` before returning (safe, slower)
`PublishAsyncContext`	fire-and-buffer — returns immediately; up to 200k acks in flight (fast, throughput)
`TracedPublish` / `…Async`	the above + an OpenTelemetry producer span + `TraceInfo` in the envelope

The async variant returns before the server confirms — great for the activity-log firehose (Lesson 11) where throughput matters more than a per-message ack. The traced async even ends its span later, when the PubAck finally arrives on a channel (jetstream.go:496-503).

A real publish + retry

Here's conversationmgmt telling notification to send a push — the cross-service link between your two services:

internal/conversationmgmt/modules/conversation/infrastructure/nats/notification.go:32-89 (condensed)

msg, _ := proto.Marshal(&ypb.NatsCreateNotificationRequest{ … })
try.Do(func(attempt int) (bool, error) {              // retry the publish, up to 5×
    _, err := jsm.TracedPublish(ctx, "Conversation.PushNotificaton",
                                constants.SubjectNotificationCreated, msg)   // → Notification.Created
    return attempt < 5, err
})

Callers wrap publishes in try.Do for retry — and because every attempt carries the same… wait, actually a fresh ULID per call, so publish retries rely on try.Do succeeding rather than on dedup; the dedup window protects against JetStream's own internal redelivery of a publish. Either way, the consumer's app-level idempotency (Lesson 8) is the real backstop.

JetStream publishing & deduplication

The publish path, the PubAck, and how the Nats-Msg-Id dedup window works.

→ docs.nats.io — model deep dive (dedup)
→ Developing with JetStream · in-repo internal/golibs/nats/jetstream.go:444-554

Check yourself (from memory)

Q1. What does the DataInMessage envelope carry besides the payload?

Tenancy + user + (traced) B3 trace ride alongside Payload, so the async consumer can rebuild the request context.

Q2. What does nats.MsgId(idutil.ULIDNow()) on every publish enable?

The Nats-Msg-Id header feeds JetStream's dedup window (default 2 min) — a repeat id is dropped.

Q3. When would you use PublishAsyncContext over the synchronous publish?

Async returns immediately with up to 200k acks in flight — used by the high-volume activity-log firehose.

Recall: publishing — the envelope, the dedup id, the variants.

envelope + MsgId + sync/async/traced, then reveal

Every publish wraps the payload in npb.DataInMessage{Payload, ResourcePath, UserId} (tenant + user pulled from ctx via UserInfoFromCtx), proto-marshals, and appends nats.MsgId(ULID) → the Nats-Msg-Id dedup header (JetStream dedup window ~2 min drops repeats). Traced variants add an OTel producer span + TraceInfo (B3) in the envelope → tracing spans the async hop. Four methods: PublishContext (sync, waits for PubAck), PublishAsyncContext (fire-and-buffer, ≤200k in flight — the firehose), TracedPublish/…Async (+ span). Callers wrap in try.Do (5×). The tenant-in-envelope is why RLS survives the hop (Lesson 7 unpacks it).

🎯 Interview one-liner "What happens when you publish an event?" → "We wrap the payload in a proto envelope that also carries the tenant and the trace context, marshal it, and stamp a unique message id so JetStream can dedup. There's a synchronous variant that waits for the ack and an async one that buffers for throughput; the traced versions add a producer span. Putting the tenant in the envelope is what lets the async consumer keep RLS working."

Next: the other end — subscribing and the handler, where the envelope is unpacked and the tenant re-injected so your handler runs exactly like a synchronous request. The flagship lesson. Ask me about the dedup window if it's new.

1. docs.nats.io — dedup. In-repo: internal/golibs/nats/jetstream.go:444-554, …/conversation/infrastructure/nats/notification.go:32-89.