mirror of
https://github.com/milvus-io/milvus.git
synced 2026-10-28 08:55:40 +08:00
XuanYang-cn
6c78b6403a
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. issue: #54628 pr: #54639 **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. **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. **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 3.0's existing KMS-invalid retry classification and WAL trace propagation. - Selected messages within a transaction are decoded before any of them are dispatched. - EZ deletion ordering is outside this change. DropCollection's consuming-ACK behavior is already present on the 3.0 base and is unchanged by this PR. Terminal decryption errors on subscribed messages retain the existing panic behavior. **Validation** 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. Local Go UT revalidation on `8edd12cce8f823cc0dbc24de042c3a36718138f7` (2026-10-26): - All tests in `pkg/mq/msgdispatcher`, `pkg/streaming/util/message/adaptor`, and `pkg/streaming/util/message` passed. Across these packages, 76 top-level tests passed, with no skipped tests. - Used local Pulsar 2.8.2 and the shared resource guard with one effective job, `-tags dynamic,test -gcflags="all=-N -l" -count=1 -parallel=1`; all test names were selected. - Scoped lint for these three packages passed with 0 issues in the earlier validation; earlier local statement coverage for the new `consume_msg.go` adaptor was 96.1%. Lint and coverage were not rerun during this Go UT pass. - `gofumpt` and `git diff --check` passed in the earlier validation. - An earlier full local `static-check` attempt failed because `internal/storagev2/filesystem_metrics.go` references `C.LoonFilesystemMetricsList` and `C.loon_filesystem_{get,list,free}_metrics`, while the installed native headers expose the newer metrics-sources API. The Go calls match upstream/3.0. That full check was not rerun in this pass; scoped lint does not replace it. - A full `make test-go` and an encrypted-cluster E2E run with the real CMEK plugin were not performed in this validation. --------- Signed-off-by: yangxuan <xuan.yang@zilliz.com>