[n24q02m]

Motivation

When chunking is enabled, reserveSemaphore() acquires one permit per chunk up front from publishSemaphore, whose capacity is MaxPendingMessages (default 1000).

If totalChunks > MaxPendingMessages, the producer can never acquire all the permits it needs. It blocks in Acquire(ctx) holding every permit it will ever get, while those permits can only be released after the chunks are dispatched, which never happens because internalSend() is never reached. Send() only returns when the caller's context expires.

Reproduction: with MaxPendingMessages: 2, EnableChunking: true, ChunkMaxMessageSize: 1000 and a 10 KiB payload (11 chunks), Send() blocks until the context deadline.

Approach

This follows the throttle semantics suggested in the review (and in #1462): MaxPendingMessages limits how much is admitted, it is not a hard cap, exactly like MemoryLimitController. A message is never split between "sent" and "not sent".

• Semaphore gets AcquireN / TryAcquireN. A request is admitted while the permits already held are below the limit (>= is rejected, matching TryAcquire), even if the request itself overshoots the limit. Later requests wait until usage drops below the limit again.
• Release no longer blocks when nobody is waiting. The old unbuffered-channel implementation could block a releasing goroutine once permits were above the limit. Waiters are now kept in a FIFO list guarded by a mutex, and a cancelled waiter neither consumes nor leaks a permit.
• reserveSemaphore() takes all totalChunks permits in one call before the message enters the event loop, and records the number in reservedSemaphore. If the message fails before it is split (memory limit, context, closing producer), done() returns all of them. Once split, each chunk releases its own permit when it is acknowledged or fails.
• writeData() now puts the pending item in the pending queue before writing it to the connection, both under the pending queue lock. With more chunks in flight than the queue can hold, the broker could acknowledge a chunk before it had been queued; the receipt was then dropped (Got ack ... for timed out msg) and the callback of the chunk group never fired. This was the remaining hang after the semaphore change on its own.

Test evidence

• New semaphore tests: overshoot admission, TryAcquireN boundary, Release not blocking without waiters, FIFO wake-up, context expiry while waiting, and cancel racing with release. They pass with -race -count=20.
• New regression test TestSendChunkedMessageWithMoreChunksThanMaxPendingMessages (11 chunks, MaxPendingMessages: 2, bounded 30s context) completes instead of blocking until the deadline.
• TestChunkBlockIfQueueFull (MaxPendingMessages: 1) now expects the send to complete. It was the test that exposed the receipt-before-enqueue problem: it failed intermittently before the writeData() change and ran 800 iterations under -race without a failure after it.
• TestChunksEnqueueFailed changed: a single message larger than the limit is now admitted, so the test fills the limit with one pending SendAsync and expects ErrSendQueueIsFull for the next message.
• TestMemLimitRejectProducerMessagesWithChunking no longer needs the retryAssert added earlier in this PR. No chunk is dispatched when the memory reservation fails, so the memory is released synchronously.
• go build ./... and go vet are clean. The remaining failures in the broker-backed suite (TestChunkSize, TestChunkDLQWithNack, TestSendTimeout, TestTxn_AckChunkMessage, plus the TLS and token tests) fail identically on pristine master in the same environment, which I checked by running the same suite against a master worktree.

[geniusjoe]

Thanks for working on this fix! The deadlock issue is real and important. I have some concerns about the current per-chunk semaphore acquisition approach that I'd like to discuss.

Partial chunk delivery risk

The current implementation acquires semaphore permits one-by-one inside the internalSend chunk loop. This introduces a partial delivery scenario — some chunks may be successfully dispatched to the broker while subsequent chunks fail to acquire a permit:

Non-blocking mode (DisableBlockIfQueueFull=true):

MaxPendingMessages: 5, totalChunks: 11

chunk 0: uses the permit reserved by reserveSemaphore → sent ✅
chunk 1: TryAcquire() → ok (permits=2) → sent ✅
chunk 2: TryAcquire() → ok (permits=3) → sent ✅
chunk 3: TryAcquire() → ok (permits=4) → sent ✅
chunk 4: TryAcquire() → ok (permits=5) → sent ✅
chunk 5: TryAcquire() → fails (permits=5 >= maxPermits)
  → sr.done(nil, ErrSendQueueIsFull)
  → chunks 5–10 are never sent

Blocking mode (DisableBlockIfQueueFull=false): if the context times out while waiting for a permit mid-loop, the same partial delivery occurs.

In both cases, the broker has already persisted chunks 0–4, but chunks 5–10 will never arrive. On the consumer side, the ExpireTimeOfIncompleteChunk timer (default 60s) will eventually fire and discard the incomplete chunk group. If AutoAckIncompleteChunk is enabled, those orphaned chunks get auto-acknowledged — resulting in silent message loss. If it's disabled, the orphaned chunks remain unacked and cycle through redelivery indefinitely, never completing.

Suggested alternative: throttle semantics (allow one request to exceed the limit)

As discussed in #1462 (review), maxPendingMessages is fundamentally a throttle, not a hard limit. I think a cleaner approach would be to align the semaphore with the existing MemoryLimitController semantics — both in the Java and Go clients, tryReserveMemory / TryReserveMemory use this pattern:

// From memory_limit_controller.go — TryReserveMemory
if m.IsMemoryLimited() && current > m.limit {
    return false  // only reject when already over the limit
}
// otherwise allow, even if current + size > limit

The key idea: as long as the current usage hasn't exceeded the limit yet, allow the entire request (all chunks) to go through in one atomic operation, even if it temporarily pushes usage above the limit. Subsequent requests are blocked/rejected until usage drops back below the limit.

Concretely, this would mean adding batch acquire support to the Semaphore (e.g. AcquireN / TryAcquireN) with the same "allow one request to burst" semantics, and keeping reserveSemaphore as a single atomic call before entering the event loop:

func (s *semaphore) TryAcquireN(n int32) bool {
    for {
        current := atomic.LoadInt32(&s.permits)
        if current > s.maxPermits {
            return false // already over limit, reject
        }
        // Allow this batch to exceed maxPermits (same as MemoryLimitController)
        if atomic.CompareAndSwapInt32(&s.permits, current, current+n) {
            return true
        }
    }
}

[n24q02m]

Thanks for the detailed review — the partial-delivery argument convinced us, and we're not going to defend the per-chunk acquisition loop.

We implemented the batch acquire you proposed and measured it against the regression test. Short version: on its own it does not close #1448, so we'd like to check the direction with you before pushing another iteration.

What we measured

Implementation as described in your sketch: AcquireN / TryAcquireN on the internal semaphore, reserveSemaphore taking all totalChunks permits in one atomic call before the message enters the event loop, and the per-chunk acquire removed from internalSend. TestSendChunkedMessageWithMoreChunksThanMaxPendingMessages (11 chunks of 1000 bytes), Pulsar 3.0.11 standalone, DisableBatching: true:

MaxPendingMessages	result
2 (< totalChunks)	FAIL — context deadline exceeded after 30s
11 (= totalChunks)	PASS in 0.05s
1000	PASS in 0.05s

The same test against cc1033e (the per-chunk code in this branch) passes in 0.13s at MaxPendingMessages: 2 — the value the test itself sets.

So the batch acquire regresses this regression test at its own configured value, and only starts passing once MaxPendingMessages is raised to at least the chunk count. That is the opposite of the direction we need, and it's why we'd rather not push it as-is.

One plausible reading of the boundary: at MaxPendingMessages: 2 the per-chunk acquire keeps at most 2 chunks in flight, so chunk k+2 is only dispatched after chunk k is acked. With the batch acquire nothing throttles in-flight chunks, and something else becomes the bound at the same totalChunks threshold — i.e. the semaphore is not the only limit on a chunk group.

What we do know about the remaining hang

At timeout (goroutine dump plus producer warnings, MaxPendingMessages: 2):

• the event loop is parked in runEventsLoop's select — idle, not blocked in writeData, PutUnsafe, or the semaphore;
• no goroutine is blocked in the semaphore or the pending queue;
• no sequenceId mismatch (local < remote / local > remote) is logged;
• exactly one ack is dropped as Got ack ledgerId:11 entryId:2 for timed out msg, i.e. p.pendingQueue.Peek() returned nothing when it arrived.

That is consistent with every chunk being dispatched and acked while the chunk group's callback still never fires, but we haven't isolated it yet.

Two corrections to the sketch

1. The boundary should be >=, not >. With current > s.maxPermits and two concurrent callers at current == maxPermits, both pass the check and both CAS, so permits overshoots by 2n; repeated bursts have no bound. current >= s.maxPermits matches the existing TryAcquire in the same file and bounds the overshoot to roughly maxPermits + n under contention. That isn't a hard limit either — it's still a throttle, like TryReserveMemory.

2. The release side needs reservedSemaphore. sendRequest already carries a reservedSemaphore field; if reserveSemaphore acquires n permits, it must record n there and each chunk's completion path must decrement it, otherwise every error path that calls sr.done() without sending all chunks leaks n-1 permits permanently. In the batch design the parent's done() is the branch that historically doesn't run at all after dispatch.

Question

Is there prior art we should be following for throttling a chunk group as a unit — in particular, whether the bound should be the chunk count at all, or whether the chunks should be admitted against the connection's pending-message budget rather than the producer's publish semaphore? We'll keep the test at MaxPendingMessages: 2, its current value, since totalChunks > MaxPendingMessages is the case this PR is about, and iterate against it in whatever direction you prefer.

[n24q02m]

I pushed the redesign (0ace76c). It follows your throttle suggestion, and it also closes the hang I described in my last comment.

Batch acquire. reserveSemaphore now takes all totalChunks permits in one AcquireN / TryAcquireN call before the message enters the event loop, and the per-chunk acquire in internalSend is gone, so a message can no longer be half sent. As in my last comment, the check is >= rather than >, and the count is recorded in reservedSemaphore, so a message that fails before it is split gives back every permit. Each dispatched chunk still releases its own permit. I also rewrote the semaphore itself with a mutex and a FIFO list of waiters. The old Release could block on its unbuffered channel once permits were above the limit and nobody was waiting, which was one of the hangs I hit while testing the batch version.

The remaining hang was not the semaphore. It was an ordering problem in writeData. The chunk was written to the connection first and added to the pending queue second. With more chunks in flight than the queue can hold, the broker could ack a chunk before it had been queued. The receipt then found nothing at Peek() and was dropped as "Got ack ... for timed out msg", and the callback of the chunk group never fired. That matches the single dropped ack I saw in the goroutine dump. writeData now enqueues first and writes second, both under the pending queue lock. TestChunkBlockIfQueueFull with MaxPendingMessages: 1 used to hang intermittently (around the 70th run in a loop) and now ran 800 iterations under -race without a failure.

That also answers my question about the connection's pending budget: with the ordering fixed, the producer semaphore alone is enough, so I did not add a second limit.

Tests. TestSendChunkedMessageWithMoreChunksThanMaxPendingMessages stays at MaxPendingMessages: 2 and now passes. I added semaphore tests for overshoot, the boundary, FIFO wake-up, and cancellation racing with release (-race -count=20). Two existing tests changed:

• TestChunkBlockIfQueueFull now expects the send to succeed.
• TestChunksEnqueueFailed now sends one message to fill the limit and expects ErrSendQueueIsFull for the next one, because a single oversized message is admitted under these semantics.

The retryAssert I had added to TestMemLimitRejectProducerMessagesWithChunking is gone, since nothing is dispatched when the memory reservation fails.

The tests that still fail locally (TLS/token setup, the admin port, broker max message size, DLQ, send timeout) fail the same way on a clean master checkout in my environment, so they are not caused by this change. I updated the PR description to match the new approach.

If you see a case where a chunk group can still be dispatched partially, I would like to hear about it.

[n24q02m]

Hi @geniusjoe, gentle ping whenever you have a moment — commit 0ace76c switches the multi-chunk path to batch-acquire permits upfront as discussed, and workflows are currently awaiting approval to run. Thanks!

[geniusjoe]

Hi @n24q02m, thanks a lot for this work — especially for reworking the per-chunk approach into a batch acquire after the earlier review, and for digging into the timeout itself rather than only the symptom.

I should own that the AcquireN / throttle direction I suggested earlier was too simplistic. On a second look, turning internal.Semaphore from a hard limit into a throttle changes the core admission path for every producer, and the resulting churn (+341/-61, including a rewrite of the semaphore and its test suite) feels disproportionate for what is essentially an edge case: one message whose chunk count exceeds MaxPendingMessages. I don't think we should reshape the semaphore's contract to accommodate that case.

I'd also like to keep the Go client's underlying behavior aligned with the Java client as much as possible. Java keeps a hard, fair java.util.concurrent.Semaphore and never relaxes it for chunking; it changes when permits are acquired in ProducerImpl instead. To be fair, the FIFO waiter list you introduced is closer to the JDK implementation and is a genuine improvement — my concern is specifically the admission rule change, letting a single request overshoot the limit, which is a producer-wide contract change rather than a chunking fix.

So I'd like to step back and rethink the approach before asking you to iterate further — if I can find a viable direction that keeps MaxPendingMessages as a hard limit and preserves whole-message dispatch, I'll bring it back as an alternative. Let's hold this iteration for now — I'd rather not have you spend more time on a version that may change direction, and I'll hold off on the workflows as well.

Thanks again for the detailed analysis.