Add fast ingest publishing and atomic EOB to batch-publish

[scottf]

Adds the two publishers ADR-50 defines that this module did not have, and makes a pass over what was already here.

What is new

EobBatchPublisher ends an atomic batch without storing a final message, so a transaction does not have to hold a message back, or invent a filler one, just to carry the commit. The server discards the sentinel and reports a batch size that excludes it.

FastPublisher and EobFastPublisher implement fast ingest: server driven flow control, GapMode.Ok / GapMode.Fail, per message error reporting, and ping. Not atomic — messages persist as they arrive, and you choose up front what a gap means to you.

The fast control channel is read on a dispatcher of the publisher's own, and that thread does one thing: decide whether the server has ended the batch, and record it. The flow accounting, the counters, the terminal ack and every listener callback stay on the thread that called add, commit or ping, in arrival order. So a batch the server abandons while the application is idle is known to be over immediately, through isTerminal() and getEndReason(), without moving user callbacks onto a thread the application did not choose.

When a gap or a per message error ends a Fail batch, the server sends a final PublishAck saying how far it actually got. That ack is the only authoritative record of what was stored — ADR-50 says a gap report explicitly is not — so committing such a batch throws carrying it: catch (FastPublishException e) { e.getPublishAck(); }.

Changes to what was already here

Version is 0.x, so these are breaking where noted.

• Ack timeouts are milliseconds. ackTimeout(long) on both builders; the Duration overload is deprecated and converts, so existing code still compiles and links. A Duration under a millisecond used to mean wait forever on one path and time out immediately on the other.
• The per message ack settings are gone from BatchPublishOptions — breaking. They were accepted and never read. They belong to the batch and have always worked on the publisher's builder.
• Connection level rejections arrive as BatchPublishException rather than escaping add unchecked, and a send the connection refused before queueing gives its batch sequence back rather than leaving a hole for the server to reject.
• An acknowledged add reports the server's error rather than only Invalid ack returned from add with confirm, so atomic publish is disabled no longer gets dropped on the floor.
• commitAsync completes with the BatchPublishException itself rather than a RuntimeException wrapping it — breaking for anyone unwrapping two layers.
• A publisher level message ttl now applies to every message. It was silently ignored unless the message also carried a BatchPublishOptions.
• Four things the batch protocol does not allow are now refused locally, because each costs the whole batch: an expected last sequence after the first message, the batch protocol headers in user headers, Nats-Expected-Last-Msg-Id, and a batch id that is not a single subject token.
• The commit's PublishAck is checked against the client's own count and batch id, skipped on the fast side once a gap makes that count an upper bound rather than an equal.

Verification

66 tests, and all eight examples run end to end against nats-server v2.15.0-dev. Javadoc is clean.

[synadia-claude-reviewer]

Claude finished @scottf's task in 9m 19s —— View job

---

Code Review

• Gather context and understand request
• Read PR diff and key source files
• Analyze for correctness, thread safety, exception handling, API correctness
• Post inline comments for specific issues
• Write overall review summary

---

The implementation correctly covers the four publisher types ADR-50 defines, and the structural choices — dedicated dispatcher thread for terminal-state detection, BlockingQueue hand-off for happens-before, idempotent unsubscribe(), the NotSent sentinel for sequence bookkeeping, and the validateAck skip on gap/error batches — are all sound. The threading model is well-reasoned and documented.

A few issues found:

Medium — fast publisher re-committable after a timeout

FastPublisher.commit() and EobFastPublisher.commit() have no finally block to mark the batch terminal when awaitPubAck() throws (timeout or server error). Contrast with BatchPublisher.commit(), which always calls markClosed() in finally. Without that guard, a second call to commit() passes requireCommittable() and sends a duplicate commit op to the server for a batch that may already be finalized. See inline comments on FastPublisher.java:47-51 and EobFastPublisher.java:39-45.

Medium — first-reply timeout leaves publisher in non-terminal state

When awaitFirstReply() times out, terminal is not set, leaving the publisher with batchSeq = 1 and a non-null firstSubject. A subsequent add() call sends FAST_BATCH_OP_APPEND for a batch that was never acknowledged by the server. See inline comment on AbstractFastPublisher.java:319-324.

Low — stale comment in drain()

The Javadoc for drain() refers to a "pending count" AtomicLong that doesn't exist. See AbstractFastPublisher.java:661-663.

Low — momentary isTerminal()/getEndReason() inconsistency

end() writes terminal = true before CAS-ing endReason. A thread reading terminal via the awaitFlowWindow() loop condition and calling getEndReason() before the next drain() could transiently observe isTerminal() == true and getEndReason() == Open. Swapping the order (CAS first, volatile write second) makes the documented invariant precise. See AbstractFastPublisher.java:645-648.

Nit — duplicate import

PublishAck is imported twice in AbstractBatchPublisher.java:7-8.

[synadia-claude-reviewer]

PublishAck is imported twice (lines 7 and 8). Duplicate import; remove line 8.

Fix this →

[scottf]

Fixed in ae7f5ea. Mine, from adding validateAck — Java permits duplicate imports so nothing caught it.

[synadia-claude-reviewer]

When nextMessage() returns null (timeout — typically a server too old to support fast ingest), the exception propagates but terminal is never set. The publisher is left with batchSeq = 1 and firstSubject != null in a non-terminal state. A subsequent add() call will pass requireUsable() and send FAST_BATCH_OP_APPEND (since batchSeq != 0), which targets a batch that was never initialized on the server.

The fix is to mark the batch terminal before throwing:

private void awaitFirstReply() throws FastPublishException {
    Message m = nextMessage();
    if (m == null) {
        end(EndReason.Abandoned);  // prevent further use of this orphaned batch
        throw new FastPublishException(batchId,
            "No response to the first message of the batch. The server may not support fast ingest publish.");
    }
    process(m);
}

Users who use try-with-resources are already protected, but the publisher's own state should not allow re-use after this failure.

Fix this →

[scottf]

Fixed in ae7f5ea, going one step further than suggested: abandon() rather than end(EndReason.Abandoned), so the dispatcher and its thread are released too. If the first message was never answered the batch is unrecoverable, so there is no reason to keep the control channel alive.

Writing the test for this turned up a second problem you did not flag, and it was hiding this one. Publishing to a subject no stream captures makes the server answer 503 No Responders on the reply subject, and that status was being read as the terminal PublishAck — ending the batch as Committed and reporting Invalid JetStream ack. So the timeout path was not even reachable in that scenario. A status is now refused as itself, and the timeout test subscribes a core responder to the subject so it actually reaches a timeout.

[synadia-claude-reviewer]

If awaitPubAck() times out (or throws for any reason), the batch is not marked terminal. requireCommittable() would pass on a subsequent call, and a second FAST_BATCH_OP_COMMIT message would be sent to the server for a batch that may already be committed or in an unknown state.

This is the same problem BatchPublisher.commit() solves with finally { markClosed(); }. The fix for the fast side is analogous — call end(EndReason.Abandoned) in a finally block:

public PublishAck commit(@NonNull String subject, Headers userHeaders, byte[] data) throws FastPublishException {
    requireCommittable();
    _send(subject, userHeaders, data, FAST_BATCH_OP_COMMIT);
    try {
        return awaitPubAck();
    }
    finally {
        // awaitPubAck() either ends Committed or throws; either way,
        // a second commit would be invalid.
        if (!isTerminal()) {
            end(EndReason.Abandoned);
        }
    }
}

(Same applies to EobFastPublisher.commit().)

Fix this →

[scottf]

Fixed in ae7f5ea, via a shared abandonIfCommitDidNotFinish() in the base so both fast commits get it.

One narrowing worth recording: an acknowledgement carrying a server error already ended the batch, because awaitPubAck calls end(Committed) before constructing the PublishAck, so the JetStreamApiException is thrown from an already-terminal publisher. The hole was the timeout and interrupt paths only.

And retrying is not an option even in principle — a lost final ack cannot be recovered by pinging, because a committed batch is cleaned up server side and the ping comes back as an unknown batch id. So terminal is the only correct end state here, which is what the fix does.

[synadia-claude-reviewer]

Same issue as FastPublisher.commit(): if awaitPubAck() times out the batch is not marked terminal, allowing commit() to be called again and sending a second commit sentinel.

Fix this →

[scottf]

Fixed in ae7f5ea, same shared helper as FastPublisher.commit().

[synadia-claude-reviewer]

Stale comment: "The pending count comes first because it is an AtomicLong read while the take locks the queue" — there is no pending count or AtomicLong in this class. The implementation is just a no-wait control.poll(), which is correct and self-explanatory. The comment should be updated or dropped.

Fix this →

[scottf]

Fixed in ae7f5ea. Stale from the change that moved the control channel onto a dispatcher — the comment described the synchronous read it replaced. Rewritten to describe what the method now does and why the empty case is the common one: it runs twice per add, against a channel that is quiet by design at one acknowledgement per flow window.

[synadia-claude-reviewer]

Subtle ordering gap between terminal and endReason. The class Javadoc states "the reason is Open exactly while isTerminal() is false." Strictly, that invariant can be momentarily violated: the dispatcher writes terminal = true (volatile), then CAS's endReason. A publishing thread that reads terminal in the awaitFlowWindow() loop condition could see true and exit, then call getEndReason() before a drain() call delivers the queue message that provides the BlockingQueue happens-before edge carrying the CAS result — and see Open.

In practice this is harmless because:

1. drain() is always called next by add(), and the queue hand-off establishes the required happens-before.
2. The publishers are documented as single-threaded.

But the Javadoc claim is not quite accurate. Consider either reversing the order (CAS first, then the volatile write) or qualifying the claim to note the brief window, so the contract is precise.

Fix this →

[scottf]

Fixed in ae7f5ea, with one correction to the reasoning.

Swapping is right, and for the reason you give: a caller that sees isTerminal() true must never then see Open, because that is the pair it acts on, and the volatile write to terminal cannot be reordered before the compare-and-set.

But swapping does not make the javadoc claim precisely accurate — it moves the window to the other side, where the reason is set an instant before the batch reads as terminal. No ordering makes the biconditional true. So the claim was changed rather than preserved: the guarantee is now stated one-directionally, that once isTerminal() is true the reason is never Open. The reverse window is harmless and mildly useful, since serverEnded() noticing early only means the terminal PublishAck of a gap-abandoned batch is collected sooner.