issue: #50425
Disable growing-source flush by default so TEXT collections use the
writer-buffer path unless the feature is explicitly enabled.
Also write TEXT V3 bloom filter and BM25 stats into the manifest instead
of returning legacy stats logs.
Signed-off-by: luzhang <luzhang@zilliz.com>
Co-authored-by: luzhang <luzhang@zilliz.com>
issue: #51250
Keep TEXT segments on StorageV3 while making growing-source flush an
explicit allowed mode instead of a mandatory TEXT path.
Propagate create-segment storage versions through WAL flusher and write
buffer, reject storage-version mismatches in DataCoord, and keep raw
segcore chunks sticky for segments that may be flushed from growing
source.
Signed-off-by: luzhang <luzhang@zilliz.com>
Co-authored-by: luzhang <luzhang@zilliz.com>
issue: #51436
### What this PR does
- Propagates a CreateCollection RecoveryStorage observation failure from
the flusher component to WAL dispatch.
- Stops before creating a DataSyncService when the recovery state was
not recorded.
- Adds coverage for the canceled observation path, including the absence
of schema lookup.
### Verification
- `go test -v -tags dynamic,test -gcflags="all=-N -l" -count=1
./internal/streamingnode/server/flusher/flusherimpl -run
"TestWALFlusher.*CreateCollection"`
- `make static-check`
- `gofumpt -l` on changed Go files
- `git diff --check`
Signed-off-by: sijie-ni-0214 <sijie.ni@zilliz.com>
issue: #51446
- message properties: add an explicit `Unreplicable` marker for concrete
WAL messages and expose builder opt-in through `WithUnreplicable`
- replication: skip unreplicable messages in the CDC sender and
secondary replicate interceptor before callback replay
- DDL producers: mark unsupported snapshot, manifest backfill, and
external collection refresh broadcasts as unreplicable
- streaming docs: document the message-level skip marker and current
unsupported DDL replication behavior
Signed-off-by: chyezh <chyezh@outlook.com>
Make ComposeTSByTime the logical-zero time-based timestamp API, add
ComposeTSByTimeWithLogical for explicit logical composition in tests,
and remove the duplicate GetCurrentTime wrapper.
See also: #51119
Signed-off-by: yangxuan <xuan.yang@zilliz.com>
Signed-off-by: yangxuan <xuan.yang@zilliz.com>
issue: #51183
`TestWALLifetime` is flaky (seen on unrelated PR #51180, the only
failure among 14937 Go tests). Root cause is a nondeterministic
contract, not test timing: `WaitCurrentStateReachExpected` only consults
`ctx` when it actually has to wait — if the background task has already
converged, the fast path returns without observing cancellation. Since
the background intentionally executes even canceled expected states (see
the deadlock note in `doLifetimeChanged`), `Open`/`Remove` with a
pre-canceled context race between `context.Canceled` and `nil`.
## Changes
- **`wal_state_pair.go`**: check `ctx.Err()` at the entry of
`WaitCurrentStateReachExpected`, making the canceled-context outcome
deterministic (standard Go fail-fast convention). Behavior is otherwise
unchanged: the expected state is registered before the wait, the
background still converges, WAL lifecycle ownership stays with the
background task, and the coord-side recovery (retry with new/higher
term) is untouched — only the caller's wait terminates
deterministically.
- **`wal_lifetime_test.go`**: gate the term-11 mock open so the
canceled-context waiters provably have to wait, released right after the
assertions — a regression pin that keeps the test deterministic even
independently of the fail-fast check.
## Verification
- `go test -race -tags dynamic,test -run TestWALLifetime -count=50` →
50/50 pass
- full `walmanager` package, `-race -count=3` → pass
- `gofmt` / `go vet` clean
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Signed-off-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
issue: #50931
## Summary
Initialize analyzer runtime options with Lindera download URLs and the
default dictionary path. Register runtime config callbacks so analyzer
yaml updates are propagated to the Rust analyzer layer.
Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
issue: #50846
- streaming replication: treat non-current txn messages covered by the
replicate checkpoint as stale duplicates
- replicate tests: cover stale txn body retry while another txn is in
progress
Signed-off-by: chyezh <chyezh@outlook.com>
issue: #50499
## Summary
- Return a retryable import system error when CommitImport arrives
before an import job reaches Uncommitted.
- Make HandleCommitVchannel wait for Uncommitted before mutating
committed vchannels or invoking the visibility callback.
- Retry HandleCommitVchannel in the WAL flusher with the default retry
backoff, and observe the CommitImport WAL message only after the commit
ack succeeds.
## Test Plan
- [x]
`PKG_CONFIG_PATH=/Users/yihao.dai/workspace/milvus/internal/core/output/lib/pkgconfig
CGO_LDFLAGS='-Wl,-rpath,/Users/yihao.dai/workspace/milvus/internal/core/output/lib'
GOTOOLCHAIN=auto go test -tags dynamic,test -gcflags="all=-N -l"
-count=1 ./internal/datacoord -run
'TestCommitImportCallback|TestHandleCommitVchannel'`
- [x]
`PKG_CONFIG_PATH=/Users/yihao.dai/workspace/milvus/internal/core/output/lib/pkgconfig
CGO_LDFLAGS='-Wl,-rpath,/Users/yihao.dai/workspace/milvus/internal/core/output/lib'
GOTOOLCHAIN=auto go test -tags dynamic,test -gcflags="all=-N -l"
-count=1 ./internal/streamingnode/server/flusher/flusherimpl`
- [x]
`PKG_CONFIG_PATH=/Users/yihao.dai/workspace/milvus/.worktrees/fix-import-early-commit-retry/internal/core/output/lib/pkgconfig
CGO_LDFLAGS='-Wl,-rpath,/Users/yihao.dai/workspace/milvus/.worktrees/fix-import-early-commit-retry/internal/core/output/lib'
GOTOOLCHAIN=auto make static-check`
---------
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
relate: #49716
pr: https://github.com/milvus-io/milvus/pull/49717
## Summary
Split FunctionRunnerManager lifecycle references between WAL and
delegator keys while still sharing runners by function signature. Add
latest-version lookup support for schemaVersion 0 or nil schema in the
manager; master call sites still pass the current schema version
explicitly.
Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
issue: #47420
## What this PR does
Project-wide migration of raw `fmt.Errorf` / `errors.New` in function
bodies onto
the `merr` framework, plus the Sys-vs-Input error classification and the
machinery it drives (retriability, fine-grained metrics, segcore
unification),
plus the convention docs and a linter that keeps it from regressing.
Scope: storage, proxy, coordinators (root/data/query), query node, data
node,
`pkg/util` & `internal/util`, expression parser, message queue,
streaming, and
misc packages. Bare raw-error usages went from ~3000 to a ~340 allowlist
(package-level sentinels / build-tag / test sites).
---
## How to review this PR
It is large but the vast majority is mechanical. Changes fall into three
tiers;
spend review budget on Part 2 and Part 3.
### Part 1 — Mechanical standardization (low risk, verify by rule)
Each converted call follows one of a small fixed set of rules. To
review, check
that each site obeys the matching rule rather than reading every line:
| Pattern | Rule |
|---|---|
| `fmt.Errorf("...")` originating a new error | →
`merr.WrapErrXxxMsg("...")` with a code matching the failure's meaning |
| Adding context to an existing typed error | → `merr.Wrap(err, "...")`
/ `merr.Wrapf(...)` — **preserves** the inner code (never `WrapErr*Err`,
which overwrites it) |
| Errors inside the streaming subsystem | → `status.New*` factories
(StreamingError), **not** merr — this is the component-internal dialect
(see `docs/dev/error_handling_guide.md`) |
| Low-level / control-flow signal caught by `errors.Is` | → kept as a
package-level `errors.New` sentinel (lowercase, same-package) |
Conventions are documented in `docs/dev/error_handling_guide.md`
(how-to) and
`docs/dev/error_sentinel_convention.md` (rules + audit). A
`gocritic`/`ruleguard`
rule (`rawmerrerror`, in `rules.go`) enforces "no raw `return
errors.New/fmt.Errorf`"
under `make verifiers`.
### Part 2 — Behavior changes (review these closely)
These are the sites where the wire contract or runtime behavior changes,
not just
the source text. Listed by category; representative locations given,
full set in
the diff.
**A. gRPC wire-code shifts: `UnexpectedError(1)/Code 65535` → typed
code.**
Where a handler previously returned a raw error (collapsed to
`Code=65535` on the
wire), it now returns a typed merr, so the client sees a real code. The
most
common shift is to `IllegalArgument(5)/Code 1100` (ParameterInvalid).
Touch
points include datanode task handlers (CreateTask/Query/Drop), proxy
Upsert,
querynode GetMetrics, datacoord CreateIndex, httpserver query-response
builder,
and typeutil schema validation. One code refinement: an index-param
validation
moved `1100` → `1101` (ParameterMissing). **Client/SDK assertions and
any code
that switched on `Code=65535` for these paths must be re-checked** (the
go_client
e2e assertions were already aligned in this PR).
**B. Prometheus `status` label contract change (externally visible).**
The proxy metric's coarse `fail` / `rejected` values are split into
`fail_input` / `fail_system` and `rejected_user` / `rejected_system` (in
`requestutil.ParseMetricLabel`; auth/privilege rejections count as
`rejected_user`), so dashboards can attribute a failure to caller vs
operator.
**Dashboards/alerts querying `status="fail"` must migrate to
`status=~"fail_.*"`, and `status="rejected"` to
`status=~"rejected_.*"`.** The
in-repo Grafana dashboard is already migrated; external dashboards built
on the
old values silently go empty after upgrade. This is the one change that
requires an ops-side migration.
**C. Retriability semantics.**
- C1: `merr.Status(err)` now forces `Retriable=false` when the error is
an
`InputError` — a malformed request can never succeed on blind retry, so
clients
never get the self-contradictory "your input is wrong but you may
retry".
- C2: `retry.Do` short-circuits an `InputError` (non-retriable) — **but
only when
the caller did not pass a `RetryErr` predicate**. The check is an
`if c.isRetryErr != nil { ... } else if InputError { ... }` *mutually
exclusive*
branch (`pkg/util/retry/retry.go`): an explicit `RetryErr` takes
precedence and
bypasses the InputError abort. `retry.Handle` deliberately does **not**
apply
the InputError abort (its callers signal abort via `shouldRetry=false`).
Four
flusher startup callsites that must retry through transient "not ready"
errors
were given explicit `RetryErr` escape hatches.
**D. segcore (C++→Go) error classification.**
A single shared Go-side table (`pkg/util/merr/segcore.go`) maps each
segcore code
to a merr sentinel + InputError/signal category, replacing scattered
hand-written
`if errorCode == ...` switches in the cgo wrappers. **Wire `Code` values
change
for every segcore pass-through error, not just the remapped ones.**
Named
sentinels remap (C++ `2003` → merr `2001`, `2033` → `2002`,
Folly/Knowhere codes
likewise); **all remaining pass-through codes (`2004`–`2043`, previously
surfaced to clients as raw C++ enum values) now serialize as `2000`**
(`ErrSegcore`), with the original C++ code preserved in the `Reason`
text
(`segcoreCode=...`); unknown/future codes collapse to `2000` as well
(pinned by
the `wire_code_projection` test). Transient segcore classes (object
storage /
file IO / OOM / mmap / FieldNotLoaded — 11 codes) now report
`Retriable=true`.
**Any client switching on raw segcore codes in the `2004`–`2043` range
must be
re-checked**; the in-Reason code remains available for diagnostics.
Signal
codes (PretendFinished / FollyCancel) are recognized centrally.
`errors.Is`-based
control flow on these (e.g. scheduler skip/retry) is preserved.
**E. InputError classification (25 sentinels + dynamic marks).**
25 sentinels in `errors.go` carry `WithErrorType(InputError)` (the
Collection /
ResourceGroup / Database families, `ErrIndexDuplicate`,
`ErrParameterInvalid`,
`ErrPrivilegeNotAuthenticated`, `ErrImportFailed`, `ErrQueryPlan`, ...),
plus dynamic
marks for the 8 segcore input codes (ExprInvalid, DimNotMatch,
MetricTypeInvalid, FieldIDInvalid, ...) and
`WrapErrAsInputError`. The widest blast radius is `ErrParameterInvalid`
(1100):
~2335 `WrapErrParameterInvalid*` callsites now classify as input /
non-retriable. Because of C1/C2 this changes retriability for
any path that returns these. **The audit to confirm no transient path
was
mis-marked is the single most important review item** (see Part 3). One
reverse
correction: storage field-stats parsing moved from `ErrParameterInvalid`
(input)
to `ErrDataIntegrity` — a corrupted stored stat is data corruption, not
user
input.
### Part 3 — Known risks & traps (called out proactively)
1. **`merr.Wrap` vs `WrapErr*Err` (code-masking).** `WrapErr*Err` builds
a
`wrappedMilvusError{sentinel: ErrServiceInternal}` whose `code()`
returns the
*outer* sentinel — it overwrites the inner typed code and hides the
`errors.Is`
chain. This is intentional (use it to *deliberately* downgrade), but it
was a
recurring conversion defect; the rule "add context with `merr.Wrap`,
downgrade
with `WrapErr*Err`" is enforced by convention and reviewed across the
diff.
2. **InputError × `retry.Do` blast radius.** Marking a sentinel
`InputError` makes
any `retry.Do(...)` without a `RetryErr` predicate stop retrying it.
Reviewers
should sanity-check that no transient use of the 19 newly-marked
sentinels
(especially `ErrParameterInvalid`) sits inside a retry loop that needed
to keep
spinning. The known flusher cases were handled (see C2).
3. **The ~340 raw-error allowlist.** What remains as bare `errors.New`
is, by
design: package-level sentinels (caught by `errors.Is`), `//go:build
test`
sites, and out-of-band trees (`cmd/`, `tests/`, codegen, walimpls). The
linter
only bans the *direct-return* form; assignment-then-return escapes and
the full
no-exceptions ban are deferred to an AST-based linter (Tier 2,
documented).
4. **segcore C++ second step deferred.** This PR unifies classification
on the Go
side; splitting the dual-semantic C++ codes at the source is a
follow-up.
---
## Validation
- `make verifiers`: Go side clean (gofmt + static-check across modules,
including
the new `rawmerrerror` rule with a 0-hit baseline repo-wide).
- `make test-go`: passing; the one real regression introduced (a
datanode
`invalid_task_type` assertion shifting `1` → `5` from a ParameterInvalid
conversion) was fixed in-tree.
- go_client e2e CreateIndex assertions aligned to the new merr messages.
---------
Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
relate: #49716
## Summary
Move BM25/MinHash function execution from DN/QN pipeline embedding nodes
to the write-before path before WAL append. Add collection-scoped
function runner caching and keep pipeline-side compatibility fill for
old insert messages without generated function output fields.
Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
issue: #50344
## Summary
After a `force_promote`, the persisted `SalvageCheckpoint` was
unreachable from any client. Two compounding defects:
### Defect 1 — `GetReplicateInfo` returns early, never reaching
`GetSalvageCheckpoint`
`internal/proxy/impl.go`: the handler called `GetReplicateCheckpoint`
first and returned on any error. On a standalone-primary (post
`force_promote`) the WAL is no longer a secondary, so
`GetReplicateCheckpoint` fails with `STREAMING_CODE_REPLICATE_VIOLATION`
("wal is not a secondary cluster in replicating topology") and the
follow-up `GetSalvageCheckpoint` — the whole point — was never reached.
**Fix:** treat **only** `REPLICATE_VIOLATION` as non-fatal — leave the
live checkpoint `nil` and continue to the salvage checkpoint. All other
errors stay fatal (no blanket swallow).
### Defect 2 — client retries `REPLICATE_VIOLATION` to the deadline
`internal/streamingnode/client/handler/handler_client_impl.go`:
`createHandlerAfterStreamingNodeReady` retried the (permanent) violation
in a backoff loop until the caller's context cancelled, so it surfaced
as `DEADLINE_EXCEEDED` rather than a typed error.
**Fix:** return immediately on
`status.AsStreamingError(err).IsUnrecoverable()` — a category that
**already** includes replicate violation (and is documented as "Stop
resuming retry and report to user"). The retry loop simply wasn't
honoring it.
Fixing Defect 2 is also what lets Defect 1 detect the violation:
otherwise the error would have been swallowed into a deadline before the
proxy could classify it.
## Changes
- `internal/proxy/impl.go`: `GetReplicateInfo` continues to
`GetSalvageCheckpoint` on `REPLICATE_VIOLATION`, returns
`checkpoint=nil` in that case.
- `internal/streamingnode/client/handler/handler_client_impl.go`: stop
retrying unrecoverable errors; return them within RTT.
- `internal/streamingnode/client/handler/handler_client_test.go`: add
`TestHandlerClient_GetReplicateCheckpointReplicateViolation` asserting
an immediate, typed return (no retry loop / `Watch` call).
## Test Plan
- [x] gofmt clean; `internal/util/streamingutil/status` builds
- [x] New unit test mirrors the existing handler-client mock harness
- [ ] CI: `TestHandlerClient_GetReplicateCheckpointReplicateViolation`
passes
- [ ] CI: `replication/data_salvage` integration suite —
`TestGetReplicateInfoOnPrimaryCluster` now returns cleanly on a primary
instead of erroring
> Note: the proxy / streamingnode packages can't be fully built locally
in this environment due to a pre-existing stale C++ artifact mismatch in
the unrelated `internal/storagev2/packed` cgo package (reproducible on a
clean master checkout). The changed packages are pure-Go and use
already-existing `StreamingError` APIs; CI compiles and runs them.
---------
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
## What was fixed
This fixes import two-phase commit timestamp handling so imported
segments keep the commit timestamp assigned by the import transaction
instead of being made visible at an unrelated later timestamp.
Changes included:
- Preserve import commit timestamps in DataCoord segment metadata.
- Broadcast import commit and rollback through DML channels so
downstream consumers observe the 2PC result consistently.
- Apply delete visibility in sealed segments based on the imported
segment commit timestamp.
- Carry the import transaction DML position needed for commit/rollback
propagation.
- Guard import in replicating clusters behind a disabled-by-default
config switch, and only allow `auto_commit=false` when that switch is
enabled.
- Downgrade only unsupported `GetSalvageCheckpoint` errors for
compatibility with older StreamingCoord implementations.
- Move import jobs to `Committing` before handling per-vchannel commit
acknowledgements, including the race where the vchannel commit is
handled before the DDL ack callback.
- Sync the generated `configs/milvus.yaml` entry for the new import
replication guard.
Fixes#48525
## Test plan
- `source ./scripts/setenv.sh && make SKIP_3RDPARTY=1
build-cpp-with-unittest`
- `make build-go`
- `make static-check`
-
`CLANG_FORMAT=/opt/homebrew/Caskroom/miniconda/base/envs/milvus2/bin/clang-format
make cppcheck`
- `make rustcheck`
- `make fmt`
- `go test ./cmd/tools/config -run TestYamlFile -count=1`
- `go test -tags dynamic,test -gcflags="all=-N -l" ./internal/datacoord
-run 'TestHandleCommitVchannel|TestCommitImportCallback' -count=1`
- Local import commit timestamp tests under
`~/workspace/snippets/tests/test_import_commit_ts`:
- `test_import_delete.py`
- `test_import_auto_commit.py`
- `test_import_basic.py`
- `test_import_mvcc.py`
- `test_import_insert_delete.py`
- `test_import_insert_delete_auto_id.py`
- `test_import_ttl.py`
- Local import + replication tests under
`~/workspace/snippets/tests/test_cdc/import`.
Local Python coverage used StorageV2. StorageV3 needs a separate run
with `common.storage.useLoonFFI=true`.
---------
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
## Summary
Implements Two-Phase Commit (2PC) for Import in primary/secondary
replication clusters. Data stays invisible (`is_importing=true`) until
an explicit commit signal is delivered via WAL, ensuring primary and
secondary clusters reach the same visible state at the same logical
position.
- **New proto**: `CommitImport=44`, `RollbackImport=45` WAL message
types; `Uncommitted=8`, `Committing=9` `ImportJobState` values;
`CommitImport`, `AbortImport`, `HandleCommitVchannel` RPCs on DataCoord;
`committed_vchannels` + `auto_commit` fields on `ImportJob`
- **WAL broadcast**: DataCoord broadcasts
`CommitImportMessage`/`RollbackImportMessage` to all vchannels via DDL
broadcast; CDC replicates to secondary clusters verbatim
- **DDL ack callbacks**: `commitImportV2AckCallback` CAS
`Uncommitted→Committing`; `rollbackImportV2AckCallback` CAS `*→Failed` +
segment drop
- **WAL flusher**: `wal_flusher.dispatch()` intercepts
`CommitImportMessage` per-vchannel, calls `wbMgr.FlushChannel` +
`DataCoord.HandleCommitVchannel`; no-op handler for
`RollbackImportMessage`
- **ImportChecker**: new `Uncommitted` case (auto-commits when
`auto_commit=true`); new `Committing` case (transitions to `Completed`
when all vchannels confirmed)
- **auto_commit option**: default `true` (backward compatible); `false`
lets replication platform control commit timing
- **RESTful API**: `POST /v2/vectordb/jobs/import/commit` and `POST
/v2/vectordb/jobs/import/abort`; `GetImportProgress` surfaces
`Uncommitted` and `Committing` states
**Out of scope**: `commit_timestamp` propagation (handled in companion
PR)
## Test Plan
- [ ] Unit tests for `IsAutoCommit` helper
- [ ] Unit tests for `HandleCommitVchannel` idempotency
- [ ] Unit tests for `CommitImport`/`AbortImport` RPC handlers (state
validation, mutex, broadcast)
- [ ] Unit tests for DDL ack callbacks (CAS races for commit/abort)
- [ ] Unit tests for `ImportChecker` `Uncommitted`/`Committing` cases
- [ ] Unit tests for WAL flusher dispatch (CommitImport/RollbackImport
cases)
- [ ] Unit tests for RESTful commit/abort handlers (valid jobId, invalid
jobId error path)
- [ ] `GetImportProgress` surfaces `Uncommitted` and `Committing` states
- [ ] E2E: non-replication cluster with `auto_commit=true` (default)
behaves identically to pre-2PC
issue: #48525
design doc: https://github.com/milvus-io/milvus-design-docs/pull/29🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
issue: #49959
- streaming flusher: only observe truncate collection before flusher
handling, keeping drop collection on the deferred observe path
- streaming flusher tests: cover drop collection cleanup ordering and
truncate collection early observe behavior
---------
Signed-off-by: chyezh <chyezh@outlook.com>
issue: #49634
This fixes a flaky WAL adaptor unit test introduced by #49696.
The test mocks RecoverWALFlusher and stores the parameter for assertions
after handleAlterWALFlushingStage returns. The real call passes a
non-escaping RecoverWALFlusherParam composite literal, so keeping the
pointer in the mock can read unstable stack data. Copy the parameter
value inside the mock callback before asserting on it.
Validation:
- go test -v -count=1 -tags dynamic,test -gcflags="all=-N -l"
-ldflags="-r <artifact>/cmake_build/lib -r
<artifact>/internal/core/output/lib"
./internal/streamingnode/server/wal/adaptor -run
^TestHandleAlterWALFlushingStagePassesRateLimitComponent$ -timeout 300s
- go test -v -count=1 -tags dynamic,test -gcflags="all=-N -l"
-ldflags="-r <artifact>/cmake_build/lib -r
<artifact>/internal/core/output/lib"
./internal/streamingnode/server/wal/adaptor -timeout 300s
- gofumpt -l internal/streamingnode/server/wal/adaptor/opener_test.go
- git diff --check
Signed-off-by: sijie-ni-0214 <sijie.ni@zilliz.com>
issue: #49634
Fix WAL recovery fast-fail follow-up issues and stabilize related WAL
tests.
Validation:
- go test -tags test -gcflags="all=-N -l"
./internal/streamingnode/server/wal/adaptor -count=1
---------
Signed-off-by: chyezh <chyezh@outlook.com>
## Summary
This PR adds a StreamingNode-local blocking-L0 seal policy so delete
pressure can seal growing L1 segments before `dataCoord.segment.maxLife`
expires.
The existing blocking-L0 policy only runs in the DataCoord
SegmentManager seal path. When StreamingNode owns the flush/seal path,
configured values such as
`dataCoord.sealPolicy.channel.blockingL0EntryNum` and
`dataCoord.sealPolicy.channel.blockingL0SizeInMB` could not trigger
sealing for local growing segments blocked by L0 deletes.
issue: #49647
## Solution
- Record local delete pressure in StreamingNode through
`ShardManager.ApplyDelete`.
- Keep delete pressure aggregated per eligible growing L1 segment
instead of storing one entry per delete message.
- Scope delete accounting by `pchannel/vchannel` and index growing L1
segments by channel, so `RecordDelete` only scans affected segments.
- Select the earliest growing L1 segment per `pchannel/vchannel` when
its delete pressure reaches either existing blocking-L0 threshold:
- `dataCoord.sealPolicy.channel.blockingL0EntryNum`
- `dataCoord.sealPolicy.channel.blockingL0SizeInMB`
- Preserve existing disable semantics: the policy is disabled only when
both thresholds are negative.
- Keep the stats local and best-effort; they are used only as
StreamingNode seal hints.
## Tests
- [x] `conda run -n milvus2 go test -tags test
./internal/streamingnode/server/wal/interceptors/shard/stats -count=1`
- [x] affected shard packages with shared C++ artifacts
- [x] `git diff --check`
- [ ] `make static-check` fails locally because this worktree lacks
ignored generated file
`cmd/tools/migration/legacy/legacypb/legacy.pb.go`; the main checkout
has it generated.
---------
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
issue: #47314
During rolling upgrades, old StreamingNodes may not carry resource group
labels and are reported as __default_resource_group. Keep uncovered
replicas on that legacy default SQN pool while preserving RG-isolated
assignment for replicas whose RGs are already covered by labeled
StreamingNodes.
Also update CollectAllStatus empty-resource-group semantics to prefer
__default_resource_group when it exists. This keeps newly prepared
labeled StreamingNodes out of WAL balance until
streaming.primaryResourceGroup is explicitly configured, while
preserving the previous collect-all behavior for deployments that have
no default RG nodes.
---------
Signed-off-by: chyezh <chyezh@outlook.com>
* Optimize metadata and request handling hot paths
* Make replication pending-message queue capacity and max size
configurable instead of hard-coded.
* Batch QueryCoord collection metadata saves with MultiSave to avoid
oversized single writes while still reducing per-key etcd operations.
* Reduce avoidable allocations in DataCoord catalog tests, access log
list formatting, HTTP array joining, and HTTP JSON field extraction.
* Use WAL-specific message ID decoding during flusher recovery.
* Use set-based RootCoord collection-name filtering and structured
collection-not-found errors.
issue: #44452
Signed-off-by: Buqian Zheng <zhengbuqian@gmail.com>
Related to #49398
Bump Go module references from pkg/v2 and milvus-proto/go-api/v2 to
pkg/v3 and milvus-proto/go-api/v3 so the client tracks the Milvus 3.x
release line.
This prepares the repository for the upcoming 3.x.y release by aligning
imports, module dependencies, and proto API references with the new
major-version module paths.
---------
Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
## Summary
- Add schema version check on insert path in streaming node shard
interceptor, rejecting inserts with mismatched schema version
- Add
`AlterCollection`/`AppendNewCollectionSchema`/`CheckIfCollectionSchemaVersionMatch`
methods to shard manager for schema lifecycle management
- Propagate schema version through segment allocation chain:
`shard_manager_segment` → `partition_manager` → `segment_alloc_worker` →
`CreateSegment` message header
- Add `SchemaVersionMismatch` streaming error type (classified as
unrecoverable to force proxy metadata refresh)
- Track collection schema in `CollectionInfo` with
`CollectionSchemaOfVChannel` for version gating
related: https://github.com/milvus-io/milvus/issues/48808
design doc:
https://github.com/milvus-io/milvus-design-docs/blob/main/design_docs/20260129-add-function-field-design.md
## Test plan
- [x] partition_manager_test.go updated for new `schemaVersion`
parameter
- [x] wal_test.go updated with CollectionSchema in CreateCollection
messages
- [ ] CI: code-check, ut-go, integration-test
🤖 Generated with [Claude Code](https://claude.ai/code)
Signed-off-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
## Summary
- Revert #48994 to explore a better approach for dropping replicate
configuration
- The original implementation treated empty `ReplicateConfiguration` as
a drop signal, which will be replaced with a cleaner design
issue: #48993
## Test plan
- [ ] CI passes with the revert applied
- [ ] Follow-up PR with improved implementation
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
## Summary
- Allow `UpdateReplicateConfiguration` to accept empty
`ReplicateConfiguration` as a "drop/clear" signal
- Atomically delete both config key and all replicate pchannel metadata
from etcd
- Handle nil config safely across the entire replication stack
(interceptor, channel manager, CDC util)
issue: #48993
## Test plan
- [ ] Unit tests for validator drop path (`config_validator_test.go`)
- [ ] Unit tests for catalog `DropReplicateConfiguration`
(`kv_catalog_test.go`)
- [ ] Integration test: set config → drop → re-set
(`drop_replicate_config_test.go`)
- [ ] Verify idempotency: drop when already nil is a no-op
- [ ] Verify guard: drop with multi-cluster or topology edges is
rejected
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
## Summary
- Strip replicate headers from incomplete broadcast messages during
force promote to re-append as primary messages
- Implement `GetSalvageCheckpoint` remote WAL path via gRPC, fixing
proxy infinite retry when handling `GetReplicateInfo` for non-failover
CDC sync
- Replace `FastLock`+panic with blocking `Lock` during force promote to
handle concurrent ack callbacks (e.g., replicated `DropLoadConfig`) that
still hold resource key locks
- Fix data race in `broadcastTask` between `MarkIgnore` and `Header`:
`MarkIgnore` (for force promote) mutates `b.msg` and the shared
properties map, breaking the immutability assumption. Split `Header()`
into locked/unlocked versions, deep copy properties in `MarkIgnore`, and
add locks to all unprotected `b.msg` readers
(`IsAlterReplicateConfigMessage`, `IsForcePromoteMessage`,
`isControlChannelAcked`)
## Test plan
- [x] Unit tests added for `ClearReplicateHeader`, `WithWALTerm`
idempotency, and force promote lock contention
- [x] Manual CDC replication test (`test_partition.py`) passed
- [x] Manual force promote test (`test_incomplete_broadcast_ddl.py`)
passed
- [x] Broadcaster unit tests pass with `-race` detector (3 sub-packages)
issue: https://github.com/milvus-io/milvus/issues/48821🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
issue: https://github.com/milvus-io/milvus/issues/47598
design doc:
https://github.com/milvus-io/milvus-design-docs/blob/main/design_docs/20260205-data_salvage_for_force_failover.md
## Summary
Implement data salvage feature that enables zero data loss after force
failover by capturing and exposing the salvage checkpoint, along with a
DumpMessages API to retrieve unsynchronized messages.
### SalvageCheckpoint Capture & Persistence
- Add `salvageCheckpoint` field to `ReplicateManager`
- Capture last synced position on force promote before clearing
secondary state
- Persist to separate etcd key via `StreamingNodeCataLog`
- Load salvage checkpoint on WAL recovery
### API Exposure
- Return `salvage_checkpoint` in `GetReplicateInfoResponse`
- Add `GetSalvageCheckpoint` to `HandlerClient` and `ReplicateService`
### DumpMessages Streaming API
- Implement `DumpMessages` RPC for streaming WAL messages
- Support `start_timetick` and `end_timetick` filters
- Filter out system messages (TimeTick, CreateSegment, Flush,
RollbackTxn)
- Use oneof response for status or message
### Tests
- Unit tests for salvage checkpoint capture
- Unit tests for `shouldDumpMessage` filter
- Integration tests for `GetReplicateInfo` and `DumpMessages`
## Test plan
- [ ] Unit tests pass
- [ ] Integration tests pass
- [ ] Manual test with force failover scenario
proto pr: https://github.com/milvus-io/milvus-proto/pull/568🤖 Generated with [Claude Code](https://claude.ai/code)
---------
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
Add force_promote flag to UpdateReplicateConfiguration API for disaster
recovery.
Changes:
- Add ForcePromote field to UpdateReplicateConfigurationRequest
- Refactor UpdateReplicateConfiguration to accept request object instead
of separate params
- Add WithForcePromote() method to ReplicateConfigurationBuilder
- Implement force promote validation and handling in assignment service
- Add integration tests for force promote scenarios
Force promote allows a secondary cluster to immediately become
standalone primary
when the original primary is unavailable, enabling active-passive
failover.
issue: https://github.com/milvus-io/milvus/issues/47351
design doc:
https://github.com/milvus-io/milvus-design-docs/blob/main/design_docs/20260202-force_promote_failover.md
---------
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
issue: #48389
Previously, all old version (v0) WAL messages shared the same
lastConfirmedMessageID pointing to the very first v0 message. When a
tailing scanner fell back to catchup mode (e.g., due to WAL ownership
change), it would restart from this extremely old position, causing
catchup times of 14+ minutes during which tsafe could not advance and
all search requests would timeout.
This change replaces the fixed first-message ID with a configurable
sliding window (default size 30). The lastConfirmedMessageID now points
to the message N positions back, bounding the WAL replay distance on
fallback to at most N messages.
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
issue: #47178
This commit introduces a rate limiting mechanism for Write-Ahead Logging
(WAL) operations to prevent overload during high traffic. Key changes
include:
- Added `RateLimitObserver` to monitor and control the rate of DML
operations.
- Add Adaptive RateLimitController to apply the strategy of rate limit.
- WAL will slow down if the recovery-storage works on catchup mode or
node memory is high.
- Updated `WAL` and related components to handle rate limit states,
including rejection and slowdown.
- Introduced new error codes for rate limit rejection in the streaming
error handling.
- Enhanced tests to cover the new rate limiting functionality.
These changes aim to improve the stability and performance of the
streaming service under load.
---------
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
issue: #47802
When WAL recovery replays messages for collections that have been
dropped (due to non-atomic etcd persistence or Kafka offset compaction),
orphaned GROWING segments can appear in the RecoverySnapshot, causing
newSegmentAllocManagersFromRecovery to panic.
Three fixes ensure orphaned segments never reach the snapshot:
1. handleDropCollection/handleDropPartition: always flush segments first
even when vchannel is already DROPPED or missing, making replay
idempotent.
2. handleCreateSegment: skip segment creation when vchannel is dropped
or non-existent.
3. getSnapshot(): filter out GROWING segments whose vchannel is not
active or whose partition has been dropped.
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Related to #46199
## Summary
Remove 5 unused or misused Go dependencies to reduce module bloat and
consolidate overlapping libraries:
- **`mgutz/ansi`** → replaced with inline ANSI escape codes (only used
for 3 color constants in migration console)
- **`valyala/fastjson`** → replaced with `tidwall/gjson` (only 1 file
used fastjson; gjson is already used in 22+ files)
- **`google.golang.org/grpc/examples`** → replaced with existing
`rootcoordpb` (test file pulled in entire grpc examples repo for a mock
server)
- **`remeh/sizedwaitgroup`** → replaced with `chan` semaphore +
`sync.WaitGroup` (only 2 files, trivial pattern)
- **`pkg/errors`** → replaced with `cockroachdb/errors` (the project
standard; `pkg/errors` was used in 1 file)
## Behavior change: DeleteLog.Parse() fail-fast on missing fields
The `fastjson` → `gjson` migration adds explicit `Exists()` validation
for `ts`, `pk`, and `pkType` fields in the JSON parsing branch.
Previously, both fastjson and gjson would silently return zero values
for missing fields, causing `dl.Pk` to remain nil and panicking
downstream. The new code fails fast with a descriptive error at parse
time. This is a defensive improvement (the original code had identical
silent-failure behavior).
## Performance impact
| Change | Path type | Perf delta | Matters? |
|--------|-----------|------------|----------|
| `pkg/errors` → `cockroachdb/errors` | Cold (offline CLI tool
`config-docs-generator`) | Negligible | No |
| `mgutz/ansi` → inline ANSI codes | Cold (offline CLI tool
`migration/console`) | Marginally faster (eliminates map lookup) | No |
| `fastjson` → `gjson` (`DeleteLog.Parse`) | Warm — old-format deltalog
deserialization only | **~2.5x slower** per JSON parse (143ns→361ns) |
**No** — see below |
| `grpc/examples` → `rootcoordpb` | Test only (`client_test.go`) | None
| No |
| `sizedwaitgroup` → chan+WaitGroup | Test only (`wal_test.go`,
`test_framework.go`) | None | No |
### fastjson → gjson regression detail
`DeleteLog.Parse()` is called per-row during deltalog deserialization,
but **only for the legacy single-field format**. The new multi-field
parquet format (`newDeltalogMultiFieldReader`) reads pk/ts as separate
Arrow columns and bypasses `Parse()` entirely. Legacy deltalogs are
rewritten to parquet format during compaction, so this is a dying code
path. Additionally, deltalog loading is I/O-bound — the JSON parse cost
(~361ns/row) is negligible compared to disk read and Arrow
deserialization overhead.
Benchmark (Go 1.24, arm64):
```
BenchmarkFastjsonSmall-4 8,315,624 143.1 ns/op 0 B/op 0 allocs/op
BenchmarkGjsonOptimized-4 3,321,613 361.4 ns/op 96 B/op 1 allocs/op
```
## Test plan
- [x] CI build passes
- [x] CI code-check passes
- [ ] CI ut-go passes
- [ ] CI e2e passes
- [x] Boundary test cases added (bare number, missing pkType/ts/pk)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Signed-off-by: Li Liu <li.liu@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
issue: #47647
Refactor the cluster-level broadcast mechanism to decouple the message
package from the channel registration lifecycle:
- Replace internal provider pattern with opaque ClusterChannels type
passed externally to WithClusterLevelBroadcast()
- Add channel package singleton (syncutil.Future) exposing
GetClusterChannels() and GetPChannelNames() blocking accessors
- Add PChannel() interface to MutableMessage/ImmutableMessage for
deriving physical channel from virtual channel
- Validate non-control-channel entries are physical channels using
funcutil.IsPhysicalChannel and use funcutil.IsOnPhysicalChannel for
control channel matching
- Move control channel substitution logic into WithClusterLevelBroadcast
to simplify callers (datacoord, coordinator, assignment service)
- Add lock interceptor unit tests and cluster broadcast test coverage
- Add integration test for FlushAll with streaming node restart to
verify data integrity across node lifecycle
---------
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
issue: #47647
Control channel messages (AlterReplicateConfig, FlushAll, AlterWAL) were
incorrectly falling through to the per-vchannel lock path, causing wrong
lock granularity and incomplete transaction cleanup. This fix ensures
these messages acquire the global write lock, consistent with recovery
storage and wal flusher handling.
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
## Summary
- Recovery storage's control channel filter silently dropped FlushAll
messages,
preventing growing segments from being marked as flushed
- Replace explicit type checks with `IsBroadcastToAll()` to allow all
broadcast
messages through, matching the pattern in `wal_flusher.go` and
`shard_interceptor.go`
## Test plan
- [x] Unit test: FlushAll on control channel flushes all growing
segments
- [x] Unit test: FlushAll on regular channel still works (regression)
- [x] Unit test: FlushAll is idempotent
- [x] Existing recovery storage tests pass
- [x] golangci-lint passes
issue: https://github.com/milvus-io/milvus/issues/47637🤖 Generated with [Claude Code](https://claude.com/claude-code)
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
issue: #47295
When WAL append fails, the message may have been contaminated with
internal state (e.g., WAL term already set). Reusing such a message on
retry would cause a panic. This fix ensures a fresh message is built on
each attempt.
For segmentAllocWorker, the segmentID, storageVersion, and limitation
are preserved across retries to ensure consistent message structure,
while the message itself is rebuilt each time.
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
