mirror of
https://github.com/milvus-io/milvus.git
synced 2026-10-12 02:19:10 +00:00
tinswzy
836aa60f28
issue: #50000 design doc: docs/design-docs/design_docs/20260604-idempotent_write.md ## Problem A client whose Insert times out or loses its ack cannot tell whether the write landed. Retrying may duplicate the rows; not retrying may lose them. There is no way for the server to recognize that the second request is the same write as the first. ## Solution Make the WAL append itself idempotent, and implement Insert on top of it. ``` client ──(gRPC metadata idempotency-key / REST Idempotency-Key)──> proxy │ derives a key when ▼ the caller gave none message property `_ik` │ streamingnode: per-vchannel dedup window, consulted BEFORE append │ WAL summary: the window's durable half ``` **Dedup happens before the append.** The interceptor consults a per-vchannel window: a first sighting owns the key and appends; a concurrent retry waits on the owner; a completed key is answered with the first attempt's message id, timetick and last-confirmed position, and nothing reaches the WAL. A multi-message insert (a partition-key collection, or a payload split by `pulsar.maxMessageSize`) is grouped into a transaction and the key is carried on the synthesized `CommitTxn`, so a partial write unit can never land. **The key is either supplied or derived.** With `collection.insert.idempotency.enabled` and no client key, the key is a hash of the destination and the payload — names only, no ids and no collection generation, hashed as the client sent it, so a retry reproduces it before any server-side resolution. Under `autoID` the candidate ids are pre-allocated so a retry reproduces the first attempt's primary keys and its row-to-shard routing. **The window survives a restart.** A new pchannel-scoped WAL summary store persists the window: each idempotent write contributes a record, records are sealed into chunk objects, and an object-storage manifest indexes them. On WAL open, recovery reads the chunks back and rebuilds each vchannel's window. **The chunk write is ordered against the WAL checkpoint.** It happens inside the recovery storage's dirty-snapshot persist, before the consume checkpoint covering those records is saved, and a failure fails that persist. The checkpoint is therefore itself the boundary between what the store holds and what the WAL still holds: no second position, no clamp on the checkpoint, and no rewind at recovery. The cost is deliberate — object-storage latency sits on the checkpoint path, and a stalled store stalls the checkpoint rather than silently dropping records. **A superseded owner cannot advance the consume checkpoint.** `WALCheckpoint` now carries the term of the publisher that last advanced it, and the advancement is a compare-and-swap. The new term claims the checkpoint — writing its term, leaving the position — *before* its forward probe, which is what leaves no gap: whatever the superseded owner wrote before the claim was written before the probe, so the probe adopts it; whatever it writes after cannot advance the checkpoint, so those records stay replayable from the WAL. Claiming after the probe is not enough, and neither is a compare-and-swap on its own: nothing records the new term until it claims. **DDL that empties a collection invalidates the window.** `DropCollection`, `TruncateCollection` and `DropPartition` each record a per-vchannel tombstone in the manifest, published before the checkpoint that covers the DDL. Without it, an auto-derived key would hash the same after the data underneath it is gone and the re-insert would be answered as a duplicate — a write reported as successful into an empty collection. ## Configuration All off by default; the feature needs both the global flag and the collection property. | Key | Default | | | --- | --- | --- | | `streaming.idempotency.enabled` | `false` | global kill switch | | `collection.insert.idempotency.enabled` | unset | per-collection property | | `streaming.idempotency.maxBytesPerWindow` | `16MiB` | per-vchannel in-memory window cap | | `streaming.idempotency.maxRetainedBytes` | `256MiB` | per-pchannel budget of retained chunk objects | | `streaming.idempotency.maxRetainedChunks` | `256` | per-pchannel cap on the NUMBER of chunks | | `streaming.idempotency.maxKeyLength` | `256` | shared with the broadcast path | Retention is byte- and count-bounded, never time-bounded: a horizon expressed in time is invalidated by time passing, which would empty the window after an outage — exactly when a resuming client needs it. Duplicate visibility is therefore measured in bytes of writes, not in minutes. ## Scope **Insert only.** Upsert rejects a configured key; Delete has no way to set one. Replicated writes bypass the window entirely — the replicate stream has its own exactly-once delivery, and a key inside a replicated message belongs to the source cluster's history. ## Note for reviewers This PR introduces `internal/streamingnode/server/wal/walsummary/` into master. #53114 organizes the same package for the transform log; whichever lands second rebases onto the other. The layout, object paths and key format here are the ones that PR settled on. The checkpoint fence overlaps #52542, which introduces the same `WALCheckpoint.term` field and the same conditional-commit primitive. **One difference is deliberate:** that PR claims the checkpoint after summary recovery has sealed the inherited coverage, and that ordering still leaves the window open — a chunk the superseded owner writes between the probe and the claim is in neither the probe result nor blocked by the CAS, so it can still advance the checkpoint and truncate past records nothing will then read. Claiming before the probe is what closes it, and `TestFenceConsumeCheckpointPrecedesSummaryRecovery` pins that order. Whichever PR lands second rebases onto the other; the ordering above is the part worth keeping. ## Known limitations - **Ambiguous append errors.** Releasing the key on append failure assumes an error means nothing was written; a WAL implementation that lands the write anyway (pulsar documents this) can let a same-key retry duplicate rows — the same outcome a retry without the feature produces. Crash recovery is unaffected. Closing the live-process gap needs the window to reconcile against the recovery-side observer. - **Trusted keys.** No payload fingerprint is stored. An explicit key reused for a different logical write returns the original write's result; key uniqueness per logical request is the client's contract. - **A DDL tombstone is coarser than the rows it buries.** `DropPartition` invalidates the whole vchannel's window. - **Per-shard dedup visibility.** A multi-shard insert fans out to independent per-vchannel windows with no cross-shard atomicity, so at a retention boundary a retry can be deduplicated on some shards and applied fresh on others. The proxy logs the per-shard breakdown when a fan-out mixes the two. - **autoID cost.** Idempotent autoID inserts consume roughly 2x auto-id space and usually one extra allocator round-trip, the price of keeping retry row-to-shard assignment consistent with Delete/Upsert hash routing. - **A DDL that empties a collection also forgets explicit keys.** The tombstone exists to stop a derived key from answering a re-insert of the same rows once the data under it is gone, and a persisted record does not say whether its key was client-supplied or derived, so the tombstone cannot spare one and bury the other. Separating them would mean carrying that distinction in every record and keeping explicit keys alive past a drop; the contract is instead scoped to a collection's life. - **The checkpoint fence does not extend to the rest of the snapshot.** When a snapshot exceeds the store's txn op limit, the segment/vchannel/salvage writes go out unguarded before the guarded checkpoint commit. Closing it needs a conditional multi-batch commit in the metastore txn layer, which is the same layer work the lost-claim item above needs. - **Orphan collection needs retention to roll.** Objects a superseded owner wrote after its successor's probe are collected when retention retires the last chunk of their term, so a pchannel whose retention never rolls never collects. That is self-limiting rather than unbounded — producing an orphan requires a superseded owner to write a chunk, and rolling retention requires chunks to be written. - **Object storage outliving etcd is not detected.** The store's keys are scoped by pchannel and term, and the term comes from etcd. Resetting etcd while keeping the bucket makes recovery adopt a higher-term manifest from the previous incarnation. Nothing else in Milvus supports that state either. --------- Signed-off-by: Zack <[email protected]>