[PureGo] Generalize stream durability model

[zlata-stefanovic-db]

Summary

Generalizes the internal stream core so one implementation can serve both the existing atomic protocols (Protocol Buffers, JSON) and a future record-count protocol, with no public API change.

• Durability units — acknowledgments are tracked as cumulative unit counts over submitted ranges rather than one ack per item, so a protocol can acknowledge part of a batch.
• Partial replay slicing — on reconnect, only the unacknowledged suffix of a partially acknowledged item is replayed.
• Submission receipts — a transport can report how much of a multi-frame send actually reached the server, so a failure part-way through a batch keeps the acknowledged prefix instead of discarding it.
• Extension seams — EncoderHooks, AckModelHooks, OpenFunc, and an exported WireStream let a protocol instantiate the core over its own payload type.

Atomic proto/JSON behavior is unchanged: those encoders report one unit per item, never produce partial acknowledgments, and continue to resolve wire offsets through the existing path.

Behavior fixes found while reviewing this change

• An acknowledgment that arrives while its Send is still outstanding, where that Send then fails, is no longer converted into a non-retryable protocol error. The core cannot know how much of that send reached the server, which is not a server violation — so the unusable ack is discarded and the retryable send failure drives recovery. Without this, a transient failure killed the stream permanently instead of reconnecting, and blamed the server for a violation it never committed.
• Rejecting an impossible submission receipt now also clears the buffered-ack state. Otherwise teardown waits forever for a completion event that was already consumed, hanging the receiver, the supervisor, and every Close()/Terminate() caller.
• An impossible submission receipt is now terminal rather than retryable. A receipt larger than the payload, or a short receipt from a Send that reported success, is a deterministic transport defect: reconnecting replays the same payload into the same defect, so the stream burned its whole retry budget before failing. Accepting a short receipt from a successful send would also record unsubmitted units as submitted, so it has to fail loudly instead.
• Durable progress refreshes the head item's lack-of-ack budget, including when a full ack promotes the next item to head. Previously only a partial ack landing inside the head refreshed it, so a boundary-aligned ack promoted an item still carrying the deadline it was given on entering flight. Once partial acks had carried the head past that deadline, completing it tore down a healthy stream on the very ack that proved the server alive. Acks are ordered, so a promoted item could not have been made durable sooner; a duplicate or stale ack still makes no progress, so a stalled server cannot postpone recovery.
• A definitive server rejection now survives the prefix-ACK teardown path. When a partial submission's prefix ACK ended the durability wait, the receiver was reaped without its result being read, so the server's status was dropped and the earlier send error was returned instead — recovery replayed against a server that rejects every attempt, and an auth rejection never invalidated its credential. The sender-first arbitration now also runs after the reap, restricted to definitive rejections so the forced abort's own retryable error cannot replace the cause.
• An acknowledgment that advances the watermark but resolves to no offset (landing in a gap below every submitted range) is rejected rather than silently raising the connection's ack watermark with no durable progress behind it.

Performance

Resolving an acknowledgment while a Send is in flight no longer allocates or copies the outstanding window. AckState carries the active range beside the submitted slice (Active/HasActive, read through NumRanges/RangeAt) instead of materializing a combined slice, so the resolver walks both in submission order and ack processing is no longer O(n) in the in-flight window.

An A/B run against a real Delta table showed no throughput change for proto/JSON ingestion from the generalized ack model. That run predates the copy removal above, so it is a lower bound on the current state.

Test plan

• cd purego && go test -race ./... — 543 tests pass, with and without -race
• cd purego && go test -count=3 ./internal/stream — no flakiness in the timing-sensitive tests
• cd purego && go vet ./... and gofmt -l clean
• deadcode ./... reports nothing

durability_model_test.go exercises the new generic seam through a protocol-neutral row-based fake: unit-resolution table cases, partial acks, replay of only the unacknowledged remainder, receipt-preserved prefix acks, and hook validation.

Every behavior fix above has a regression guard that was confirmed to fail without its fix, by running the new test against the unmodified core in a scratch worktree:

• TestHookProtocolImpossibleReceiptIsTerminal — both receipt shapes fail with recovery exhausted after 4 attempt(s), showing the payload was replayed into the same defect.
• TestHookProtocolPromotedItemGetsFreshAckBudget — fails with opened 2 connections, want 1, showing the promoted item was torn down and replayed.
• TestHookProtocolPrefixAckPreservesTerminalRecvStatus — fails with recovery exhausted after 4 attempt(s) instead of surfacing the server's Unauthenticated status, and asserts the credential is invalidated exactly once.
• The recovery guard in core_test.go passes against the pre-change core but failed against this branch; the teardown guard hangs to the test timeout without its fix.

Notes

• No public API change and no Arrow dependency. This is the generic-core foundation for the PureGo Arrow Flight path, split out so it can be reviewed on its own.
• No NEXT_CHANGELOG.md entry: the change is confined to purego/internal/. Partial acknowledgments, submission receipts, and the record-count ack path are all unreachable for the shipping proto/JSON protocols, and the recovery fix restores behavior main already has, so there is no user-visible difference relative to main.
• Now based directly on main. It was previously stacked on purego-descriptor-cache-lifecycle, which was closed without merging.

[teodordelibasic-db]

A terminal receive status can be lost once ackProgress satisfies this wait. We
then close the transport and wait for receiverDone, but never consume the
receiver result published during that close, so runOnce returns the earlier
send error. A permanent server rejection can consequently be retried, and an
authentication rejection skips credential invalidation.

Could we preserve a non-cancellation receiver error after reaping the receiver,
using the same arbitration rule as the other sender-first paths?

<-receiverDone
if !receiverReported {
        select {
        case receiverErr := <-receiverExitCh:
                if receiverErr != nil && !errors.Is(receiverErr, context.Canceled) {
                        cause = receiverErr
                }
        default:
        }
}

It would be good to cover a partial receipt followed by its prefix ACK, with
Recv returning codes.Unauthenticated when teardown closes the wire, and
assert that Flush returns codes.Unauthenticated rather than the send error.

[zlata-stefanovic-db]

Fixed as suggested

[teodordelibasic-db]

Refreshing only the head item's pendingAt leaves every trailing item on its
original deadline. If items A and B enter flight together and partial ACKs keep A
alive past one timeout, completing A makes B the head with an already-expired
deadline. The timer read at
purego/internal/stream/core.go:1352
then tears down a healthy stream immediately, even though durable progress just
arrived.

Since partial progress is intentionally allowed to extend the head item here,
could we carry that extension forward when the next item becomes head, or base
the timeout on the last genuine durability progress instead?

A focused regression case can enqueue two row-count items, partially ACK A before
the timeout, fully ACK A after B's original deadline, and assert that B is not
immediately replayed on a second connection.

[zlata-stefanovic-db]

Yes, I'll allow the partial progress to extend to the next item as well. Also another case I've noticed - if a partial ack extends progress of the current head, then the full ack of a previous item should refresh the new head's deadline as well, which we do not currently do. Now all progress refreshes the deadline.

Adding the regression cases too.

[teodordelibasic-db]

An over-reported receipt or a short successful receipt is a deterministic adapter
contract violation, but these paths create ordinary fmt.Errorf values, which
the supervisor treats as retryable. The duplicate checks at
purego/internal/stream/core.go:1550
do the same. Recovery therefore reconnects and replays the same payload after an
impossible local result.

Could we classify both validation sites as non-retryable, for example:

err = wrapValidation(fmt.Errorf("stream: invalid submission receipt: ..."))

return rejectSubmission(wrapValidation(fmt.Errorf(
        "stream: invalid submission receipt: ...",
)))

The existing overReportingRowWire is a deterministic regression seam. With
recovery enabled and a second wire available, the test can assert that the error
is terminal and no second connection receives a replay.

[zlata-stefanovic-db]

fixed now, made these non-retryable with wrapValidation, also added the suggested test

[teodordelibasic-db]

The scratch buffer removes the per-ACK allocation, but this append still copies
all unacknowledged ranges whenever an ACK arrives while Send is active. Under
continuous ingestion sending is usually non-nil, so ACK processing remains
O(n) in the outstanding window and can consume substantial memory bandwidth at
the default in-flight limit.

Could we represent the active range separately in AckState, or otherwise let
the resolver consume the submitted slice and the one active range without
materializing a combined slice?

[zlata-stefanovic-db]

Yes, thank you! I went with representing the active range separately in AckState (now holds Active+HasActive boolean). It exposes NumRanges&RangeAt functions to be iterated over. The ResolveAcknowledgedUnits now loops over the submitted and active range in submission order without combining them into a merged slice.

[elenagaljak-db]

Nice refactor!

[elenagaljak-db]

On main the pause case was an immediate hard abort (stream.Close()). Switching it to gracefulTeardown, plus resolvePendingOnExit(!isPause) now waiting up to DrainTimeout for the in-flight Send instead of aborting it, is a real behavior change on the shipping proto/JSON path — server CloseStreamSignal hits this.

Effects for proto/JSON:

• the in-flight Send is allowed to complete (and can become durable if an ack was already buffered) rather than being aborted and replayed;
• the server sees END_STREAM instead of an abrupt cancel;
• teardown can take up to ~2x DrainTimeout when a Send is slow/stuck — the receiver's resolvePendingOnExit wait gates receiverDone, and gracefulTeardown then waits on receiverDone again with its own budget.

It's bounded (no hang) and probably a fine change, but it's not behavior-neutral vs main, so the "no user-visible difference" rationale for skipping the changelog doesn't quite hold. Can we confirm it's intended and covered by a pause test that exercises a slow/stuck Send (not just fakeRPC.CloseSend closing Recv)?

[zlata-stefanovic-db]

Added two pause tests with a controlled Send: TestCoreStreamPauseAwaitsSlowSendAndKeepsAck
(Send completes after the pause deadline, the buffered ack becomes durable, nothing replays — fails against main's abort) and TestCoreStreamPauseStuckSendStaysBounded (Send never returns, with and without a buffered ack: bounded abort, record replays).

[elenagaljak-db]

This slice-before-decode makes the GetUnackedBatches doc comment above (lines 691-696) inaccurate for the generic core: it still says the grouping is "the unit the server acks atomically" and reproduces "the original durability boundaries," but for a record-count protocol the returned group is now the unacked suffix, not the original ingest group. Still true for proto/JSON (ackedUnits is always 0), but since this PR introduces the divergence, worth updating the comment here — and the public Stream.GetUnackedBatches wording ("grouped by ingest call") too.

[zlata-stefanovic-db]

Good catch, fixed both. The internal comment now says an entry is safe to replay as one batch — exactly the ingest call for proto/JSON, only the unacked suffix when a protocol acks part of an item — and the public doc reads "grouped for replay: one group per ingest call, minus any prefix already made durable."