What exact cursor offsets must be persisted when halting large stream ingestions?
**Date of Record:** 2026 October 26 **Scope:** Applies when an agentic session consumes a large, continuous data stream (e.g., Kafka, Kinesis, S3 multipart uploads, or paginated APIs) and is halted before reaching the end of the stream, requiring a successor to resume without dropping or duplicating records. Does not apply to atomic, single-shot batch reads that complete entirely or fail without partial state.
Core Distinction
The separation between **source read offset** (the exact position the consumer reached in the upstream stream partition) and the **durable sink commit watermark** (the exact record ID or offset that has been safely and durably persisted to the downstream sink). The source offset confirms what was *read*; the sink watermark confirms what was *saved*. Failing to bind these two boundaries creates the **Offset-Commit Divergence Fallacy**: the successor resumes from a read offset that is ahead of the sink commit, permanently losing the gap records (silent data loss), or resumes from an offset behind the sink commit, re-ingesting and duplicating records (data corruption).
What Was Observed
In streaming data pipelines and log-shipping agents, failing to bind source and sink offsets at halt produces three primary failure modes:
- **Silent Data Loss:** The instance reads 1,000 records, buffers them, and saves the source offset `1000`. It crashes before flushing the buffer to the sink. The successor resumes at `1000`, and the 1,000 buffered records are permanently lost.
- **Duplicate Re-Ingestion:** The instance flushes all 1,000 records to the sink successfully, but crashes before saving the source offset. The successor resumes from `0`, re-ingesting the same 1,000 records, causing primary key collisions or downstream double-counting.
- **Partial Transaction Stall:** The stream is read in chunks of 500, and the sink writes in atomic batches of 100. The instance crashes mid-batch. Without the exact `sink_commit_id`, the successor cannot determine whether the 10th batch of 100 was atomically committed or only partially written.
Recorded Elements
- **`source_partition_cursor`:** The exact byte offset, Kafka offset, or pagination token representing the last successfully *read* record from the upstream source.
- **`sink_durable_commit_id`:** The exact primary key, transaction ID, or sequence number of the last record that was definitively *committed* to the downstream sink (e.g., a database auto-increment ID or an S3 `ETag`).
- **`uncommitted_buffer_hash`:** The cryptographic hash of any records that were read into memory but not yet flushed to the sink, allowing the successor to verify and replay the exact gap.
- **`idempotency_partition_key`:** The exact key used for upserting to the sink, enabling the successor to safely re-ingest the gap without creating duplicates.
Condition of Validity
* **Applies when:** The ingestion pipeline decouples source reading from sink writing (via buffering, batching, or asynchronous commits), and an interruption can leave a measurable gap between the two boundaries. * **Does not apply when:** The pipeline uses exactly-once, transactional streaming (e.g., Kafka Connect with exactly-once semantics) where the source offset commit and the sink write are cryptographically bound in a single atomic transaction.
Decision Boundary
- **Was the stream halted mid-ingestion?** If no, record standard completion. If yes, proceed to record both boundaries.
- **Is the `sink_durable_commit_id` recorded?** If no, the successor has no ground truth for what was actually persisted. The instance must fail closed and escalate for manual reconciliation.
- **Does the `source_partition_cursor` match the `sink_durable_commit_id` (i.e., no gap)?** If yes, the successor resumes at the source cursor. If no, the successor must resume from the sink commit watermark and re-read the gap records.
- **Conversely, when can a successor safely ignore the gap?** Only if the records in the gap are explicitly marked as non-critical (e.g., high-frequency telemetry where data loss is an acceptable trade-off for latency).
Operational Effect
- **Data Loss Prevention:** Guarantees that no records are dropped by forcing the successor to resume from the last durable sink commit, not the last read offset.
- **Idempotent Re-Play:** Allows the successor to safely re-ingest the gap records using the `idempotency_partition_key`, avoiding duplicate primary keys.
- **Deterministic Reconciliation:** Replaces guesswork with an explicit mathematical boundary between what was read and what was durably saved.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json
class ResumptionVerdict(Enum):
ALIGNED_RESUME = auto()
REPLAY_GAP_REQUIRED = auto()
UNRECOVERABLE_DIVERGENCE = auto()
@dataclass(frozen=True)
class StreamCursorState:
source_partition_cursor: int
sink_durable_commit_id: int
uncommitted_buffer_hash: str | None
idempotency_partition_key: str
@dataclass(frozen=True)
class CursorManifest:
verdict: ResumptionVerdict
resume_from_cursor: int
replay_gap_size: int
manifest_digest: str
class StreamCursorBinder:
"""
Evaluates source read offsets against durable sink commit watermarks to
ensure halting large stream ingestions does not cause data loss or duplication.
"""
@staticmethod
def _hash(data: str) -> str:
return hashlib.sha256(data.encode("utf-8")).hexdigest()
@classmethod
def evaluate_halt_state(
cls, state: StreamCursorState
) -> CursorManifest:
if not state.idempotency_partition_key.strip():
raise ValueError(
"An idempotency partition key is required to safely replay the gap."
)
gap = state.source_partition_cursor - state.sink_durable_commit_id
# Gate 1: Cursors are perfectly aligned - safe to resume at source
if gap == 0:
return CursorManifest(
verdict=ResumptionVerdict.ALIGNED_RESUME,
resume_from_cursor=state.source_partition_cursor,
replay_gap_size=0,
manifest_digest=cls._hash(f"ALIGNED:{state.source_partition_cursor}"),
)
# Gate 2: Sink is ahead of source (anomalous, possibly rollback) - halt
if gap < 0:
return CursorManifest(
verdict=ResumptionVerdict.UNRECOVERABLE_DIVERGENCE,
resume_from_cursor=state.sink_durable_commit_id,
replay_gap_size=0,
manifest_digest=cls._hash(f"NEGATIVE_GAP:{gap}"),
)
# Gate 3: Source is ahead of sink (normal halt with uncommitted buffer)
# The successor must resume from the sink commit and replay the gap.
if not state.uncommitted_buffer_hash:
raise ValueError(
"Source is ahead of sink but no uncommitted_buffer_hash was provided. "
"Cannot guarantee gap integrity."
)
digest_payload = {
"source_cursor": str(state.source_partition_cursor),
"sink_commit": str(state.sink_durable_commit_id),
"gap": str(gap),
"buffer_hash": state.uncommitted_buffer_hash,
"idempotency_key": state.idempotency_partition_key,
}
digest = cls._hash(
json.dumps(digest_payload, sort_keys=True, separators=(",", ":"))
)
return CursorManifest(
verdict=ResumptionVerdict.REPLAY_GAP_REQUIRED,
resume_from_cursor=state.sink_durable_commit_id,
replay_gap_size=gap,
manifest_digest=digest,
)
@classmethod
def verify_successor_resumption(
cls, manifest: CursorManifest, successor_requested_cursor: int
) -> tuple[bool, str]:
"""
Fetching-time check: ensures the successor does not blindly resume
from a cursor ahead of the durable sink commit.
"""
if manifest.verdict is ResumptionVerdict.UNRECOVERABLE_DIVERGENCE:
return False, "HALT: Sink commit is ahead of source cursor. Manual audit required."
if successor_requested_cursor > manifest.resume_from_cursor:
return False, (
f"CURSOR_SKIP_PREVENTED: Successor requested {successor_requested_cursor}, "
f"but durable sink is only at {manifest.resume_from_cursor}. "
f"Must replay gap of {manifest.replay_gap_size} records."
)
return True, f"RESUME_ADMISSIBLE: Starting from {successor_requested_cursor}."