mirror of
https://github.com/milvus-io/milvus.git
synced 2026-10-28 17:05:53 +08:00
XuanYang-cn
085fedac74
QueryNode may replay historical WAL messages from a dropped database while consuming another collection on the same physical channel. If the dropped database's encryption zone is no longer available, decrypting those messages before checking their destination can panic the consumer. This 2.6 PR also fixes a separate deletion-ordering issue: DropCollection could complete before Flusher consumed preceding encrypted messages, allowing a subsequent DropDatabase to remove the encryption zone too early. issue: #54628 pr: #54639 **QueryNode: previous flow** The adaptor converted each V1 Insert/Delete message into a decoded message before passing it to the dispatcher. This conversion called `Payload()`, which decrypted the body. ```text WAL message → adaptor reads and decrypts the body → dispatcher checks the vchannel subscription → deliver to a subscriber, or discard ``` For example, collections A and B share a pchannel, but the consumer only subscribes to B. Replaying an old encrypted message from A could panic inside the adaptor, before the dispatcher could discard it. **QueryNode: new flow** The adaptor preserves the original V1 Insert/Delete message in a lazy wrapper. Routing information remains accessible without decoding the body. The dispatcher triggers decryption only after matching the message's vchannel to a current subscription. ```text WAL message → adaptor creates a lazy wrapper without decrypting → dispatcher checks the vchannel subscription ├─ no matching subscriber → discard without decrypting └─ matching subscriber → decrypt and decode → deliver ``` In the example above, A's message is discarded without requesting its encryption key, while B's message is decoded and delivered. **Additional 2.6 fix: wait for DropCollection consuming ACKs** A separate Flusher panic remained after the QueryNode change; see the [sanitized stack and validation evidence](https://github.com/milvus-io/milvus/issues/54628#issuecomment-6412440222). DropCollection in 2.6 did not enable `AckSyncUp`. The broadcaster could use FastAck after WAL append and execute the callback that removes collection metadata. DropDatabase could then see an empty database and remove its EZ while Flusher was still decoding earlier DML. ```text Previous flow: append DropCollection → FastAck → remove collection metadata → DropDatabase removes EZ Flusher may still be decoding earlier encrypted DML ``` Enable `AckSyncUp` for DropCollection, matching the requirement already present in master and 3.0 through [#49412](https://github.com/milvus-io/milvus/pull/49412). The existing Flusher path closes the collection's data service before sending its consuming ACK. The broadcaster now waits for all target channels before executing the deletion callback. ```text New flow: append DropCollection → Flusher consumes preceding DML and closes the data service → receive consuming ACKs from all target channels → remove collection metadata → DropDatabase can remove EZ ``` The wait belongs on DropCollection because its broadcast includes the collection's data channels. DropDatabase broadcasts only to the control channel; waiting for that ACK alone would not establish completion on the data channels. **Scope and behavior** - Deferred decoding applies to **V1 Insert/Delete messages in the `deferDML=true` adaptor path**, including those inside committed transactions. - Legacy-format and non-DML messages continue through the existing conversion logic. The `deferDML=false` branch also retains its existing eager conversion behavior. - The dispatcher controls when decoding starts; encryption/decryption remains implemented in the message layer. The shared `Payload()` and `Body()` APIs retain their behavior for callers that read message bodies directly. - The selection check uses current vchannel subscriptions. It does not check database existence, and per-target replay-position filtering still runs later. - Retryable decryption failures return control to the dispatcher. It retains the pending pack across pause/resume and does not advance its progress while that pack remains pending. Preserve 2.6's existing KMS-invalid retry classification and WAL trace propagation. - Selected messages within a transaction are decoded before any of them are dispatched. Terminal decryption errors on subscribed messages retain the existing panic behavior. - The DropCollection change waits for its consuming ACKs; it does not wait for every WAL reader or guarantee publication of the global recovery checkpoint. It does not rewrite previously persisted broadcast tasks or restore already deleted keys. **Validation** QueryNode regression tests use a mocked cipher to cover unsubscribed encrypted Insert/Delete messages, target changes after prefetch, transaction replay, decoding failures, and pause/resume during a retry. `TestDropCollectionWaitsForConsumingAck` uses the real broadcaster and RootCoord callbacks with mocked storage and controlled consuming ACKs. On the old code, it fails because DropCollection completes before any consuming ACK. With the fix, zero or partial ACKs keep the database non-droppable; all ACKs allow deletion. This verifies the ordering defect without instantiating the cipher plugin. Local validation on `b6049d7b7e1b4bc6006875a3b9af7ae5eb67417e` (2026-10-26): - All tests in the following seven packages passed: **306 top-level tests, no failures and no skipped tests**: - `pkg/mq/msgdispatcher` - `pkg/streaming/util/message/adaptor` - `pkg/streaming/util/message` - `internal/rootcoord` - `internal/streamingcoord/server/broadcaster` - `internal/streamingnode/server/flusher/flusherimpl` - `internal/streamingnode/server/wal/recovery` - Used local Pulsar 2.8.2 and etcd, with the shared resource guard and one effective job. Flags: `-tags dynamic,test -gcflags="all=-N -l" -count=1 -parallel=1`; all test names in these packages were selected. - Completed a cached C++ and Go rebuild with one job and `SKIP_3RDPARTY=1` at `2335c37e554a0c09d658673966e1edbc51100c8c`. The only subsequent change is the test-helper cleanup described below; production code is unchanged. - `gofumpt` and `git diff --check` passed for the test-helper change. Scoped lint for the three message packages passed in earlier validation; it was not rerun in this Go UT pass. The single-CPU local run exposed an existing test-helper leak: delayed ACK goroutines could retry forever after `TestBroadcaster` closed its broadcaster, starving the later `TestResourceKeyLocker/deadlock_prevention` test. The timeout reproduced after the rebuild, and a CPU profile identified the ACK retry loop. The helper now observes its owning test's context. The same broadcaster package then passed, followed by the seven-package run above. No assertions, timeout values, or production lock behavior were changed. An earlier full local `static-check` failed because installed native headers lacked `C.SegcoreSetSparseInterimIndexAlgorithms`. The rebuild now installs both the declaration and exported symbol, but **full `static-check` has not been rerun**. A full `make test-go` was not performed, and encrypted-cluster E2E has not been rerun with the additional DropCollection fix. --------- Signed-off-by: yangxuan <xuan.yang@zilliz.com>