When NATS delivers twice: upsert, message ID, 24-hour window

The transcript processor turns a call’s transcript into a summary. Its first version stored two for the same call. A SIGKILL (an immediate stop with no cleanup) after the processor stored the summary and before it acknowledged the message made NATS deliver the message again, and the processor did the work twice. That happened in 10 of 10 runs stopped at that point.

This post is about the version that ships, in a video-call app with transcripts and recordings: what JetStream does there, how the services connect through it, and the changes that leave one summary when a message arrives twice.

What NATS gives the app

NATS is a message server. Services publish small messages to named subjects (transcript.processed), and other services listen. Plain NATS is fire and forget: a listener that is offline at publish time misses the message for good. JetStream adds storage on top, so an offline listener receives the message once it reconnects. A stream stores the messages published to a set of subjects, and a durable consumer keeps each reader’s position (the last message it acknowledged), even across restarts.

A message then travels like this:

1. A service publishes to a subject; the stream stores the message.
2. The consumer hands it to a reader.
3. The reader acknowledges (acks) when it is done. Until then the consumer waits for the ack wait, 30 seconds by default.
4. If the wait runs out, the consumer sends the message again (redelivery), up to a maximum number of attempts.

The app relies on redelivery. A call can end while the processor is restarting, and the summary must still be built. Redelivery covers that without a queue table in Mongo. The cost: a handler sometimes runs more than once for the same message, and a repeat must not add a second summary or notification.

The title’s 24-hour window is the duplicate window: how long a stream remembers message IDs (2 minutes by default). A publish with a remembered ID is dropped.

Services and streams

Two streams, with subjects named <domain>.<event>. TRANSCRIPTION holds transcription.segment, transcription.session-ended and transcript.processed. RECORDING holds recording.<event>.

The transcriber (the Go call-transcriber service) runs the speech-to-text path and publishes each finished segment. It now also publishes a session-ended event when the last WebSocket of a transcript closes. A new container, transcript-processor, reads that event, waits for late segments, builds a basic summary, stores it, and publishes transcript.processed.

The two services share no process. They share a subject and a Mongo collection. The transcriber saves each finished segment to Mongo, and session-ended includes publishedSegments, the number of segments it saved.

A segment can still reach Mongo after that event. The processor waits a quiet period and counts the stored segments, and repeats until two counts in a row match (at most 5 rounds). It then marks the summary complete if the count reached publishedSegments, and incomplete if segments are missing.

The backend publishes recording events to RECORDING; nothing reads that stream yet.

The transcriber and the backend skip publishing when NATS_URL is unset, so a local run or a test needs no NATS. If NATS is down, the backend logs the failed publish and carries on.

Implementation

The stream is defined once, in shared code

Both services call the same function at startup, so they cannot define the stream differently (simplified):

js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
    Name:       "TRANSCRIPTION",
    Subjects:   []string{"transcription.>", "transcript.>"},
    Duplicates: 24 * time.Hour,
})

> is the NATS wildcard for “this prefix and everything below it”, so one stream captures all three subjects. “Create or update” is idempotent (running it twice leaves the same result as running it once). It is also how the production stream got its 24-hour duplicate window without being recreated: the 73 messages it already held were still there after the update.

Consumer and redelivery settings

The processor reads transcription.session-ended through a durable consumer (simplified: the source starts from a 30-second ack wait and applies the processor’s values only when they are set):

cfg := jetstream.ConsumerConfig{
    Durable:       durable,
    FilterSubject: SessionEndedSubject,
    AckPolicy:     jetstream.AckExplicitPolicy, // done only on an explicit ack
    AckWait:       opts.AckWait,                // 6 s in the processor
    MaxDeliver:    opts.MaxDeliver,             // 5
}

For each message, the handler’s result picks the outcome:

var evt SessionEndedEvent
if err := json.Unmarshal(msg.Data(), &evt); err != nil {
    _ = msg.Term() // malformed: never redeliver
    return
}
err := handler(ctx, evt)
if err != nil {
    _ = msg.Nak() // failed: redeliver now
    return
}
_ = msg.Ack() // done

A nak (negative acknowledgement) asks for redelivery now, not after the wait. Term ends redelivery of that message. A retry cannot fix a malformed payload.

The processor can start over after any step

// error handling trimmed
func (p *Processor) Process(ctx context.Context, evt events.SessionEndedEvent) error {
    in := transcriptflow.Input{ConversationID: evt.ConversationID, SessionID: evt.SessionID, PublishedSegments: evt.PublishedSegments}
    count, err := waitStable(ctx, p.Quiet, countSegments) // wait for late segments
    complete := count >= in.PublishedSegments
    segs, err := p.Store.LoadSegments(ctx, in.SessionID)
    sum := p.Summarizer.Summarize(segs)
    err = p.Store.SaveSummary(ctx, in, sum, complete)     // upsert on conversationId:sessionId
    return p.Publisher.PublishProcessed(ctx, events.ProcessedEvent{ /* … */ }, in.Key()) // same key as Nats-Msg-Id
}

If any step fails, or the process dies between two of them, the message comes back and Process runs again from the top. The step order allows it: the two steps with side effects come last, and both are idempotent. The save is an upsert, and the publish carries a message ID.

The container runs with restart: always and starts after nats and mongo, so a processor that exits comes back and picks the message up again.

The Node side is a hook that cannot fail

The backend publishes from the function that logged recording state changes:

export function logRecordingEvent(data) {
	const event = { ts: Date.now(), ...data };
	logger.info(`RECORDING_EVENT ${JSON.stringify(event)}`);
	publishRecordingEvent(event); // new line
}

publishRecordingEvent sends to recording.<event> and does not wait. Inside, publish() catches every error and logs it, so a NATS outage cannot break a recording.

Upsert, message ID, ack wait

Upsert, not insert

The summary document’s _id is conversationId:sessionId. A second run overwrites it. This alone removes the duplicate summary.

A message ID on the notification

The upsert protects the database, not the transcript.processed message. Nats-Msg-Id is set to the same conversationId:sessionId, and JetStream drops a second publish with the same ID.

The message ID did not fail. It expired.

The test stopped the processor after it published the notification and before it acknowledged the message, and left it down for 130 seconds, longer than the 2-minute window. On restart JetStream redelivered the message, the window had already forgotten the ID, and the stream held a second notification. Two builds with the default window did this, in 2 of 2 runs each. The summary stayed at 1 because of the upsert. The change sets the window to 24 hours. It updates the production stream, so it is part of the release. After the release, the same 130-second outage against the shipped build gave: with the 2-minute window the stream stored two notifications, and with the 24-hour window one, in 3 of 3 runs each.

Idempotent writes were not enough; the duplicate window also had to be longer than a processor outage.

A short ack wait with a heartbeat

When the processor dies mid-message, JetStream redelivers once the wait runs out. At the default 30 seconds, a summary stuck behind a dead processor shows up 30 seconds late; at 6 seconds, after about 6. The catch is that a healthy run can last longer than 6 seconds, since the processor waits for late segments before summarizing, so a short wait alone would redeliver work still in progress.

The answer is a heartbeat. While the handler runs, a goroutine calls msg.InProgress() every 2 seconds (a third of the wait), and each call restarts the 6-second timer:

func heartbeat(msg jetstream.Msg, every time.Duration, done <-chan struct{}) {
    t := time.NewTicker(every)
    defer t.Stop()
    for {
        select {
        case <-t.C:
            _ = msg.InProgress() // resets the wait timer
        case <-done:
            return
        }
    }
}

The goroutine stops when the handler returns and done closes.

The short wait catches a dead processor; the heartbeat lets a long run finish. After 5 delivery attempts, JetStream stops redelivering that message.

Recording events

The backend’s logRecordingEvent is the function that logs recording state changes with a RECORDING_EVENT prefix.

That function now also publishes the same object to recording.<event>, reusing the event names from the logs: recording-confirmed, stop-requested, stored, deleted, and five more.

There is no recording.failed subject, since a new set of names would need a mapping from the ones in the logs.

A recording that is requested but never starts is published as recording.deleted, with the cause in its reason field.

A consumer that wants failures filters on that field.

How I tested it

The test that found defects stops the processor with SIGKILL at a chosen point, restarts it, and counts what ends up stored. At wait, the processor stops while it waits for late segments. At after-store, it stops after the summary is saved and before the notification is sent; a proxy between the processor and NATS discards the publish and the ack for that. At none, the control, it is never stopped. Each point ran 10 times with a 2-second quiet period on one machine. A fourth point, after-notify, lets the publish through, drops the ack, and then stops the processor; the long-outage runs use it (3 runs each with the 2-minute and the 24-hour window), with the processor kept down for 130 seconds before the restart.

case "after-store":
    // timeout and bookkeeping trimmed
    time.Sleep(quiet / 2)
    proxy.SetMode(modeFreeze)
    for summaries() == 0 { // wait until the summary is stored
        time.Sleep(time.Millisecond)
    }
    kill(cmd)              // crash here, before the notification goes out
    proxy.SetMode(modeOpen)

Recovery is the time from restart to the first transcript.processed. The version from the opening produced a duplicate summary in every after-store run. No stop point in this table comes after the notification is sent, so these runs say nothing about duplicate notifications; the after-notify runs described under the message ID cover that, at N=3 for each window.

I also tested the shipped code. An isolated compose project ran the actual transcript-processor image, and I sent session-ended twice, expected 1 summary and 1 notification, and read the 24-hour window off the stream. For recordings, I called the backend’s addRecording and deleteRecording and read the results from the RECORDING stream, including the reason on deleted. The production calls in the next section repeat this on live traffic.

Unit tests covered the publisher and its startup, how the summary is built, and the wait for a stable segment count, all without a NATS server. They would pass with a 2-minute duplicate window, the setting that let the second notification through.

Release

Before any call on the new build, all seven services were up. The processor had logged transcript processor started and the backend had logged its NATS connection.

The nats container had not been recreated. At that release it had no volume (a later change added one): a recreated container would have started with empty streams and no duplicate memory. The live TRANSCRIPTION stream showed the one-day duplicate window and its 73 old messages. RECORDING existed and was empty.

I ran the rest in a test group on production.

Recording. I started and stopped a recording. Seven events landed on RECORDING:

The events began with the request and ended with deleted. The backend’s RECORDING_EVENT log showed the same seven events in the same order. Their timestamps (ts) put the request to recording-confirmed at 12 seconds.

Transcript. A silent session tested the wiring, not the content: the transcriber published session-ended, and the processor wrote a summary with zero segments. Then I ran a call with speech. 12 segments arrived, and one transcript.processed came out: {"segments":12,"complete":true}. The consumer showed no redelivered and no outstanding messages.

What I would change

The summary is a placeholder. It counts words per speaker, keeps the three longest turns, and collects the lines that end in a question mark. Nothing reads the transcriptsummaries collection or transcript.processed yet. The wording of the output is untested. I will improve the summary once a reader needs more.