[Rust] Receiver silently absorbs server over-ack and can wedge the stream #638

Summary

The gRPC receiver silently absorbs a server ack that exceeds what the client has actually sent, and unconditionally advances its ack watermark to the bogus value. This can leave later legitimate records permanently unacknowledged, wedging the stream until the lack-of-ack timeout fires.

Location

rust/sdk/src/stream/grpc/receiver.rs:113-130

for _offset_to_ack in (last_acked_offset + 1)..=durability_ack_up_to_offset {
    if let Ok(record) = landing_zone.remove_observed() {
        // ... fulfil oneshot / callback ...
    }
    // else: remove_observed() returned Err (nothing left observed) -> silently skipped
}
drop(map);
last_acked_offset = durability_ack_up_to_offset;   // advanced unconditionally

Behavior

If the server sends durability_ack_up_to_offset larger than the highest physical offset actually observed/sent, remove_observed() returns Err for the excess iterations and the if let Ok(...) swallows it with no error. Worse, last_acked_offset is then set to the inflated server value regardless. Because subsequent acks are only processed in the range (last_acked_offset + 1)..=, any later real records whose physical offsets fall at or below the inflated watermark are never acked — their oneshots/callbacks never fire, and the stream eventually trips the server-lack-of-ack timeout.

Absent/negative offsets are only partially guarded: the fully-absent durability_ack_up_to_offset case is rejected (receiver.rs:99-110), but an out-of-range positive value is not.

Why it matters

A buggy or misbehaving server (or a protocol desync after recovery) causes silent data-durability confusion rather than a clean, observable failure. Records reported to the user as pending are never resolved.

Suggested fix

Validate each ack against the highest in-flight offset and treat an over-ack as a fatal protocol violation that tears the stream down (letting the supervisor decide whether to reconnect), rather than advancing the watermark past what was sent.

Reference: how the pure-Go core handles this

The pure-Go reimplementation validates every ack against the in-flight set and fails fast on an over-ack:

purego/internal/stream/core.go:1044-1050

highest, ok := cs.buf.highestInFlight()
if !ok || offset > highest {
    return fmt.Errorf("stream: server ack offset %d exceeds highest in-flight offset %d", offset, highest)
}

It also treats malformed/negative-offset acks as fatal (purego/internal/stream/ackmodel.go:44-50).