From 4011513d5e9cdb0aef907ccab27bebf50d04b278 Mon Sep 17 00:00:00 2001 From: tinswzy Date: Wed, 26 Aug 2026 19:52:15 +0800 Subject: [PATCH 1/5] feat: add reset checkpoint command for offline MQ switch Switching a stopped instance onto a different MQ requires rewriting every persisted position at once. `repair checkpoint` cannot do it: it only touches one vchannel's channel-cp and knows nothing about the streamingnode metadata introduced in 2.6, so streamingnode still reopens the old WAL. Add `reset checkpoint --target-wal `, which rewrites all five kinds of persisted position in one run: streamingnode-meta/wal/{pchannel}/consume-checkpoint MessageId datacoord-meta/channel-cp/{vchannel} MsgID + WALName root-coord/.../collection-info StartPositions datacoord-meta/s/... SegmentInfo.StartPosition datacoord-meta/s/... SegmentInfo.DmlPosition The last three matter because datacoord's seek falls back three levels (channel-cp -> segment position -> collection start). Rewriting only the first leaves the old encoding reachable whenever a vchannel's channel-cp is missing. WALName is set on every position. Milvus picks the decoder from it, and for woodpecker it is the only signal separating an empty earliest id (which serializes to zero bytes) from a position that was never set. Timestamps are preserved: they are global TSOs, unrelated to which MQ stores the message. Milvus filters the replay with DeliverFilterTimeTickGTE against them, so starting from earliest cannot replay anything older than the checkpoint. Two preflight gates refuse to run when a reset would lose data: growing segments (their rows only exist in the old MQ) and an in-flight broadcast task. Replication topology is not gated, and WALCheckpoint.ReplicateCheckpoint is not rewritten. That field is a subscription position against a remote cluster, owned and maintained by Milvus; switching the local WAL says nothing about it, and the two clusters are independent. Milvus puts no such gate on its own WAL switch either. Note there is no safe intermediate state, in any write order: the consume checkpoint decides which WAL to open and the channel checkpoints decide which positions to decode, so a partial run crashes milvus on startup. The command is therefore idempotent and detects a split state, and the guarantee it offers is "safe to re-run", not "safe to start after a partial run". Verified end to end against milvus 3.0.0 with real etcd/minio/pulsar/kafka: pulsar<->woodpecker and pulsar/woodpecker->kafka all switch with data intact. rocksmq as a target is blocked upstream (v3.0.0's rmq builder uses server.Rmq without lazy init, so a switched-to rocksmq WAL cannot open); milvus main has since added the nil check. Signed-off-by: tinswzy Co-Authored-By: Claude Opus 5 (1M context) --- states/etcd/common/wal_recovery_write.go | 25 ++ states/etcd/reset/checkpoint.go | 478 +++++++++++++++++++++++ states/etcd/reset/checkpoint_test.go | 435 +++++++++++++++++++++ states/etcd/reset/cleanup.go | 133 +++++++ states/etcd/reset/component.go | 25 ++ states/etcd/reset/preflight.go | 76 ++++ states/etcd/reset/wal_position.go | 112 ++++++ states/etcd/reset/wal_position_test.go | 90 +++++ states/instance.go | 3 + tests/e2e/commands/test_reset.sh | 25 ++ tests/e2e/test_commands.sh | 1 + 11 files changed, 1403 insertions(+) create mode 100644 states/etcd/reset/checkpoint.go create mode 100644 states/etcd/reset/checkpoint_test.go create mode 100644 states/etcd/reset/cleanup.go create mode 100644 states/etcd/reset/component.go create mode 100644 states/etcd/reset/preflight.go create mode 100644 states/etcd/reset/wal_position.go create mode 100644 states/etcd/reset/wal_position_test.go create mode 100755 tests/e2e/commands/test_reset.sh diff --git a/states/etcd/common/wal_recovery_write.go b/states/etcd/common/wal_recovery_write.go index 9680f338..251fefb8 100644 --- a/states/etcd/common/wal_recovery_write.go +++ b/states/etcd/common/wal_recovery_write.go @@ -33,3 +33,28 @@ func SaveVChannelMeta(ctx context.Context, cli kv.MetaKV, basePath string, meta } return cli.Save(ctx, key, string(bs)) } + +// ConsumeCheckpointKey returns the etcd key holding the pchannel's consume checkpoint. +func ConsumeCheckpointKey(basePath string, pchannel string) string { + return path.Join(basePath, walRecoveryStoragePrefix, pchannel, walRecoveryStorageConsumeCheckpoint) +} + +// SaveConsumeCheckpoint overwrites the consume checkpoint of a pchannel. +// +// The checkpoint is what streamingnode reads to decide which WAL implementation +// to open, so writing it is the commit point of any WAL switch — callers should +// write everything else first. +func SaveConsumeCheckpoint(ctx context.Context, cli kv.MetaKV, basePath string, pchannel string, checkpoint *streamingpb.WALCheckpoint) error { + data, err := proto.Marshal(checkpoint) + if err != nil { + return errors.Wrapf(err, "failed to marshal consume checkpoint for pchannel %s", pchannel) + } + return cli.Save(ctx, ConsumeCheckpointKey(basePath, pchannel), string(data)) +} + +// SegmentAssignPrefix returns the prefix holding a pchannel's growing-segment +// allocations. They are checkpointed against WAL offsets, so a WAL switch +// invalidates all of them. +func SegmentAssignPrefix(basePath string, pchannel string) string { + return path.Join(basePath, walRecoveryStoragePrefix, pchannel, walRecoveryStorageDirectorySegmentAssign) +} diff --git a/states/etcd/reset/checkpoint.go b/states/etcd/reset/checkpoint.go new file mode 100644 index 00000000..a08f515a --- /dev/null +++ b/states/etcd/reset/checkpoint.go @@ -0,0 +1,478 @@ +package reset + +import ( + "bytes" + "context" + "fmt" + "sort" + "strings" + + "github.com/cockroachdb/errors" + "google.golang.org/protobuf/proto" + + "github.com/milvus-io/birdwatcher/framework" + "github.com/milvus-io/birdwatcher/models" + "github.com/milvus-io/birdwatcher/states/etcd/common" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/msgpb" + "github.com/milvus-io/milvus/pkg/v3/proto/datapb" + "github.com/milvus-io/milvus/pkg/v3/proto/etcdpb" + "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb" + "github.com/milvus-io/milvus/pkg/v3/util/funcutil" +) + +type ResetCheckpointParam struct { + framework.ExecutionParam `use:"reset checkpoint" desc:"reset every persisted MQ position to the target WAL's earliest position, for switching a stopped instance onto a different MQ. MILVUS MUST BE STOPPED; on failure re-run until it succeeds before starting it."` + TargetWAL string `name:"target-wal" default:"" desc:"target WAL type: woodpecker, pulsar, kafka, rocksmq"` + PChannel string `name:"pchannel" default:"" desc:"restrict to a single pchannel; empty means every pchannel"` + AllowGrowing bool `name:"allow-growing" default:"false" desc:"proceed even when growing segments exist; DANGEROUS, their unflushed data is lost"` +} + +// write kinds, ordered so that the consume checkpoint lands last. +const ( + kindCollection = "collection" + kindSegment = "segment" + kindChannelCP = "channel-cp" + kindConsumeCP = "consume-checkpoint" +) + +// writeOrder decides the apply sequence. +// +// There is NO safe intermediate state, in any order. The consume checkpoint +// decides which WAL streamingnode opens; the channel checkpoints decide which +// positions it then decodes. The moment those two disagree, milvus panics on +// startup — verified: the flusher blows up in getRecoveryInfos -> +// MustGetMessageIDFromMQWrapperIDBytesWithWALName with "proto: cannot parse +// invalid wire-format data". Reordering only flips which side is stale. +// +// So the guarantee this command offers is not "safe to start after a partial +// run" but "safe to re-run": every write is idempotent and a repeat run +// rewrites exactly the keys the previous one missed. Operators must re-run +// until it reports success before starting Milvus. +var writeOrder = []string{kindCollection, kindSegment, kindChannelCP, kindConsumeCP} + +type plannedWrite struct { + kind string + key string + value []byte + before string + after string +} + +// ResetCheckpointCommand implements `reset checkpoint`. +func (c *ComponentReset) ResetCheckpointCommand(ctx context.Context, p *ResetCheckpointParam) error { + pos, err := buildWALPosition(p.TargetWAL) + if err != nil { + return err + } + + if err := c.preflight(ctx, p); err != nil { + return err + } + + writes, err := c.plan(ctx, p, pos) + if err != nil { + return err + } + deletes, err := c.planCleanup(ctx, p) + if err != nil { + return err + } + + if detail, split := c.detectSplitState(ctx, p); split { + fmt.Printf("!! previous run did not finish: %s\n", detail) + fmt.Printf("!! Milvus cannot start in this state; this run will finish the switch.\n\n") + } + + printPlan(writes, deletes, pos, p.Run) + if !p.Run { + return nil + } + if err := c.apply(ctx, writes); err != nil { + return err + } + return c.applyCleanup(ctx, deletes) +} + +// inScope reports whether a vchannel/pchannel name is covered by --pchannel. +func (c *ComponentReset) inScope(p *ResetCheckpointParam, channel string) bool { + if p.PChannel == "" { + return true + } + return funcutil.ToPhysicalChannel(channel) == p.PChannel +} + +func (c *ComponentReset) plan(ctx context.Context, p *ResetCheckpointParam, pos *walPosition) ([]plannedWrite, error) { + var writes []plannedWrite + + collWrites, err := c.planCollections(ctx, p, pos) + if err != nil { + return nil, err + } + writes = append(writes, collWrites...) + + segWrites, err := c.planSegments(ctx, p, pos) + if err != nil { + return nil, err + } + writes = append(writes, segWrites...) + + cpWrites, err := c.planChannelCheckpoints(ctx, p, pos) + if err != nil { + return nil, err + } + writes = append(writes, cpWrites...) + + consumeWrites, err := c.planConsumeCheckpoints(ctx, p, pos) + if err != nil { + return nil, err + } + writes = append(writes, consumeWrites...) + + return writes, nil +} + +// planCollections rewrites CollectionInfo.StartPositions, the last level of the +// three-level seek fallback in datacoord (channel-cp -> segment pos -> here). +func (c *ComponentReset) planCollections(ctx context.Context, p *ResetCheckpointParam, pos *walPosition) ([]plannedWrite, error) { + colls, err := common.ListCollections(ctx, c.client, c.basePath) + if err != nil { + return nil, errors.Wrap(err, "failed to list collections") + } + + var writes []plannedWrite + for _, coll := range colls { + pb := coll.GetProto() + var touched []*commonpb.KeyDataPair + changed := false + for _, sp := range pb.GetStartPositions() { + if !c.inScope(p, sp.GetKey()) || bytes.Equal(sp.GetData(), pos.raw) { + touched = append(touched, sp) + continue + } + touched = append(touched, &commonpb.KeyDataPair{Key: sp.GetKey(), Data: pos.raw}) + changed = true + } + if !changed { + continue + } + + next := proto.Clone(pb).(*etcdpb.CollectionInfo) + next.StartPositions = touched + value, err := proto.Marshal(next) + if err != nil { + return nil, errors.Wrapf(err, "failed to marshal collection %d", pb.GetID()) + } + writes = append(writes, plannedWrite{ + kind: kindCollection, + key: coll.Key(), + value: value, + before: fmt.Sprintf("collection %d startPositions=%d", pb.GetID(), len(pb.GetStartPositions())), + after: pos.String(), + }) + } + return writes, nil +} + +// planSegments rewrites SegmentInfo.StartPosition and DmlPosition, the middle +// level of the seek fallback. +func (c *ComponentReset) planSegments(ctx context.Context, p *ResetCheckpointParam, pos *walPosition) ([]plannedWrite, error) { + segments, err := common.ListSegments(ctx, c.client, c.basePath, func(s *models.Segment) bool { + return c.inScope(p, s.InsertChannel) + }) + if err != nil { + return nil, errors.Wrap(err, "failed to list segments") + } + + var writes []plannedWrite + for _, seg := range segments { + pb := seg.SegmentInfo + if pb.GetStartPosition() == nil && pb.GetDmlPosition() == nil { + continue + } + + if alreadyAtOrNil(pb.GetStartPosition(), pos) && alreadyAtOrNil(pb.GetDmlPosition(), pos) { + continue + } + + next := proto.Clone(pb).(*datapb.SegmentInfo) + before := fmt.Sprintf("segment %d start=%s dml=%s", pb.GetID(), + describePosition(pb.GetStartPosition()), describePosition(pb.GetDmlPosition())) + rewritePosition(next.GetStartPosition(), pos) + rewritePosition(next.GetDmlPosition(), pos) + + value, err := proto.Marshal(next) + if err != nil { + return nil, errors.Wrapf(err, "failed to marshal segment %d", pb.GetID()) + } + writes = append(writes, plannedWrite{ + kind: kindSegment, + key: seg.GetKey(), + value: value, + before: before, + after: pos.String(), + }) + } + return writes, nil +} + +// planChannelCheckpoints rewrites datacoord-meta/channel-cp/, the +// first level of the seek fallback and the one datacoord actually hands out. +func (c *ComponentReset) planChannelCheckpoints(ctx context.Context, p *ResetCheckpointParam, pos *walPosition) ([]plannedWrite, error) { + cps, err := common.ListChannelCheckpoint(ctx, c.client, c.basePath, func(mp *models.MsgPosition) bool { + return c.inScope(p, mp.GetProto().GetChannelName()) + }) + if err != nil { + return nil, errors.Wrap(err, "failed to list channel checkpoints") + } + + var writes []plannedWrite + for _, cp := range cps { + pb := cp.GetProto() + if alreadyAt(pb, pos) { + continue + } + next := proto.Clone(pb).(*msgpb.MsgPosition) + before := fmt.Sprintf("vchannel %s %s", pb.GetChannelName(), describePosition(pb)) + rewritePosition(next, pos) + + value, err := proto.Marshal(next) + if err != nil { + return nil, errors.Wrapf(err, "failed to marshal checkpoint for %s", pb.GetChannelName()) + } + writes = append(writes, plannedWrite{ + kind: kindChannelCP, + key: cp.Key(), + value: value, + before: before, + after: pos.String(), + }) + } + return writes, nil +} + +// planConsumeCheckpoints rewrites streamingnode's per-pchannel consume +// checkpoint. streamingnode reads its WALName to decide which WAL to open, so +// this is the write that actually performs the switch. +func (c *ComponentReset) planConsumeCheckpoints(ctx context.Context, p *ResetCheckpointParam, pos *walPosition) ([]plannedWrite, error) { + channels, err := common.ListWALDistribution(ctx, c.client, c.basePath, p.PChannel) + if err != nil { + return nil, errors.Wrap(err, "failed to list wal distribution") + } + + var writes []plannedWrite + for _, ch := range channels { + pchannel := ch.GetChannel().GetName() + meta, err := common.ListWALRecoveryStorage(ctx, c.client, c.basePath, pchannel) + if err != nil { + return nil, errors.Wrapf(err, "failed to read recovery storage of %s", pchannel) + } + if meta.Checkpoints == nil { + fmt.Printf("pchannel %s has no consume checkpoint yet, skipping\n", pchannel) + continue + } + + if cur := meta.Checkpoints.GetMessageId(); cur.GetWALName() == pos.walName && + cur.GetId() == pos.msgID.Id && meta.Checkpoints.GetAlterWalState() == nil { + continue + } + + next := proto.Clone(meta.Checkpoints).(*streamingpb.WALCheckpoint) + before := fmt.Sprintf("pchannel %s %s@%d", pchannel, + common.GetMessageIDString("", meta.Checkpoints.GetMessageId().GetId()), + meta.Checkpoints.GetTimeTick()) + + // Only the message id moves. TimeTick is a global TSO and stays valid + // across MQ implementations; AlterWalState is cleared so streamingnode + // does not try to resume an interrupted online switch. + // + // ReplicateCheckpoint is left alone: it is a subscription position against + // a remote cluster, owned and maintained by Milvus itself, and switching + // the local WAL says nothing about it. + next.MessageId = pos.msgID + next.AlterWalState = nil + + value, err := proto.Marshal(next) + if err != nil { + return nil, errors.Wrapf(err, "failed to marshal consume checkpoint of %s", pchannel) + } + writes = append(writes, plannedWrite{ + kind: kindConsumeCP, + key: common.ConsumeCheckpointKey(c.basePath, pchannel), + value: value, + before: before, + after: pos.String(), + }) + } + return writes, nil +} + +// alreadyAt reports whether a position already points at the target, so a +// repeated run can show "nothing to do" instead of listing no-op rewrites. +// bytes.Equal treats nil and empty alike, which matters for woodpecker's +// earliest id — it serializes to zero bytes and reads back as nil. +func alreadyAt(mp *msgpb.MsgPosition, pos *walPosition) bool { + return mp != nil && mp.GetWALName() == pos.walName && bytes.Equal(mp.GetMsgID(), pos.raw) +} + +// rewritePosition points a MsgPosition at the target WAL. Timestamp is kept: it +// is a global TSO, unrelated to which MQ stores the message. +func rewritePosition(mp *msgpb.MsgPosition, pos *walPosition) { + if mp == nil { + return + } + mp.MsgID = pos.raw + mp.WALName = pos.walName +} + +// alreadyAtOrNil treats an absent position as "nothing to do": reset only +// rewrites positions that exist. +func alreadyAtOrNil(mp *msgpb.MsgPosition, pos *walPosition) bool { + return mp == nil || alreadyAt(mp, pos) +} + +func describePosition(mp *msgpb.MsgPosition) string { + if mp == nil { + return "" + } + return fmt.Sprintf("%s@%d", mp.GetWALName().String(), mp.GetTimestamp()) +} + +// detectSplitState reports the WAL names currently recorded on each side of the +// switch. A previous run that died partway leaves them disagreeing, which is +// exactly the state that crashes Milvus on startup — worth naming explicitly so +// the operator knows they are finishing a job rather than starting one. +func (c *ComponentReset) detectSplitState(ctx context.Context, p *ResetCheckpointParam) (string, bool) { + seen := map[string]map[commonpb.WALName]int{ + kindChannelCP: {}, + kindConsumeCP: {}, + } + + cps, err := common.ListChannelCheckpoint(ctx, c.client, c.basePath, func(mp *models.MsgPosition) bool { + return c.inScope(p, mp.GetProto().GetChannelName()) + }) + if err != nil { + return "", false + } + for _, cp := range cps { + seen[kindChannelCP][cp.GetProto().GetWALName()]++ + } + + channels, err := common.ListWALDistribution(ctx, c.client, c.basePath, p.PChannel) + if err != nil { + return "", false + } + for _, ch := range channels { + meta, err := common.ListWALRecoveryStorage(ctx, c.client, c.basePath, ch.GetChannel().GetName()) + if err != nil || meta.Checkpoints == nil { + continue + } + seen[kindConsumeCP][meta.Checkpoints.GetMessageId().GetWALName()]++ + } + + // Compare the SET of WAL names, never the counts: the two sides are indexed + // differently (one key per vchannel vs one per pchannel), so their counts + // legitimately differ even when the instance is perfectly consistent. + names := func(m map[commonpb.WALName]int) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k.String()) + } + sort.Strings(out) + return out + } + chNames, cpNames := names(seen[kindChannelCP]), names(seen[kindConsumeCP]) + if len(chNames) == 0 || len(cpNames) == 0 { + return "", false + } + if strings.Join(chNames, ",") == strings.Join(cpNames, ",") { + return "", false + } + return fmt.Sprintf("channel-cp says [%s] but consume-checkpoint says [%s]", + strings.Join(chNames, ", "), strings.Join(cpNames, ", ")), true +} + +func printPlan(writes []plannedWrite, deletes []plannedDelete, pos *walPosition, run bool) { + byKind := map[string]int{} + for _, w := range writes { + byKind[w.kind]++ + } + + fmt.Printf("=== reset checkpoint -> %s ===\n", pos) + if len(writes) == 0 && len(deletes) == 0 { + fmt.Println(" every position already points at this WAL; nothing to do.") + return + } + for _, kind := range writeOrder { + fmt.Printf(" %-20s %d key(s)\n", kind, byKind[kind]) + } + for _, kind := range deleteOrder { + if n := countDeleteKind(deletes, kind); n > 0 { + fmt.Printf(" %-20s %d prefix(es) to DELETE\n", kind, n) + } + } + fmt.Println() + for _, kind := range writeOrder { + for _, w := range writes { + if w.kind != kind { + continue + } + fmt.Printf(" [%s] %s\n %s -> %s\n", w.kind, w.key, w.before, w.after) + } + } + for _, kind := range deleteOrder { + for _, d := range deletes { + if d.kind != kind { + continue + } + fmt.Printf(" [DELETE %s] %s\n %s\n", d.kind, d.key, d.reason) + } + } + fmt.Printf("\nNot touched: streamingcoord pchannel assignment and channelwatch info —\n") + fmt.Printf("they carry no MQ position and Milvus rebuilds them on startup.\n") + if !run { + fmt.Printf("\ndry run: nothing written. Re-run with --run=true to apply.\n") + fmt.Printf("Make sure Milvus is STOPPED and you have a `backup` before applying.\n") + } +} + +func (c *ComponentReset) apply(ctx context.Context, writes []plannedWrite) error { + for _, kind := range writeOrder { + for _, w := range writes { + if w.kind != kind { + continue + } + if err := c.client.Save(ctx, w.key, string(w.value)); err != nil { + return errors.Wrapf(err, "failed to write %s (%s).\n"+ + "METADATA IS HALF-RESET AND MILVUS WILL CRASH IF STARTED NOW: streamingnode "+ + "panics decoding a position written for a different WAL.\n"+ + "Re-run this command until it reports success, then start Milvus", w.key, w.kind) + } + } + if n := countKind(writes, kind); n > 0 { + fmt.Printf("applied %d %s key(s)\n", n, kind) + } + } + fmt.Println("reset done. Update mq.type in the Milvus config before starting it back up.") + return nil +} + +func countKind(writes []plannedWrite, kind string) int { + n := 0 + for _, w := range writes { + if w.kind == kind { + n++ + } + } + return n +} + +func countDeleteKind(deletes []plannedDelete, kind string) int { + n := 0 + for _, d := range deletes { + if d.kind == kind { + n++ + } + } + return n +} diff --git a/states/etcd/reset/checkpoint_test.go b/states/etcd/reset/checkpoint_test.go new file mode 100644 index 00000000..e3a3b3ff --- /dev/null +++ b/states/etcd/reset/checkpoint_test.go @@ -0,0 +1,435 @@ +package reset + +import ( + "context" + "log" + "os" + "path" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.etcd.io/etcd/server/v3/embed" + "go.etcd.io/etcd/server/v3/etcdserver/api/v3client" + "google.golang.org/protobuf/proto" + + _ "github.com/milvus-io/birdwatcher/asap" + "github.com/milvus-io/birdwatcher/framework" + "github.com/milvus-io/birdwatcher/states/etcd/common" + "github.com/milvus-io/birdwatcher/states/kv" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/msgpb" + "github.com/milvus-io/milvus-proto/go-api/v3/schemapb" + "github.com/milvus-io/milvus/pkg/v3/proto/datapb" + "github.com/milvus-io/milvus/pkg/v3/proto/etcdpb" + "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb" +) + +var testKV kv.MetaKV + +func TestMain(m *testing.M) { + cfg := embed.NewConfig() + dir, _ := os.MkdirTemp("", "bw-reset-test-*") + cfg.Dir = dir + cfg.LogLevel = "error" + e, err := embed.StartEtcd(cfg) + if err != nil { + os.RemoveAll(dir) + log.Fatal(err) + } + select { + case <-e.Server.ReadyNotify(): + testKV = kv.NewEtcdKV(v3client.New(e.Server)) + case <-time.After(60 * time.Second): + e.Server.Stop() + os.RemoveAll(dir) + log.Fatal("etcd server took too long to start") + } + code := m.Run() + e.Close() + os.RemoveAll(dir) + os.Exit(code) +} + +const ( + testPChannel = "by-dev-rootcoord-dml_0" + testVChannel = "by-dev-rootcoord-dml_0_440000000000000001v0" + testCollID = int64(440000000000000001) + testPartID = int64(440000000000000002) + testSegID = int64(440000000000000003) + testTimeTick = uint64(450000000000000000) +) + +// pulsarPos is a stand-in for a position left behind by the old MQ. The exact +// bytes do not matter; what matters is that reset replaces them and tags the +// result with the new WAL name. +func pulsarPos(channel string) *msgpb.MsgPosition { + return &msgpb.MsgPosition{ + ChannelName: channel, + MsgID: []byte{0x08, 0x01, 0x10, 0x02, 0x18, 0x00}, + Timestamp: testTimeTick, + WALName: commonpb.WALName_Pulsar, + } +} + +type fixture struct { + base string + comp *ComponentReset +} + +// seed writes one collection, one segment, one channel checkpoint, one consume +// checkpoint and one segment allocation — i.e. every kind reset has to handle. +func seed(t *testing.T, name string, segState commonpb.SegmentState) *fixture { + t.Helper() + ctx := context.Background() + base := path.Join(name, "meta") + t.Cleanup(func() { testKV.RemoveWithPrefix(ctx, name) }) + + save := func(key string, m proto.Message) { + data, err := proto.Marshal(m) + require.NoError(t, err) + require.NoError(t, testKV.Save(ctx, key, string(data))) + } + + save(path.Join(base, common.DBCollectionMetaPrefix, "1", "440000000000000001"), + &etcdpb.CollectionInfo{ + ID: testCollID, + // a real collection always carries a schema; ListCollections + // dereferences it unconditionally (common/collection.go:132) + Schema: &schemapb.CollectionSchema{Name: "probe"}, + VirtualChannelNames: []string{testVChannel}, + PhysicalChannelNames: []string{testPChannel}, + StartPositions: []*commonpb.KeyDataPair{ + {Key: testPChannel, Data: []byte{0x08, 0x01}}, + }, + }) + + save(path.Join(base, common.DCPrefix, common.SegmentMetaPrefix, + "440000000000000001", "440000000000000002", "440000000000000003"), + &datapb.SegmentInfo{ + ID: testSegID, + CollectionID: testCollID, + PartitionID: testPartID, + InsertChannel: testVChannel, + State: segState, + StartPosition: pulsarPos(testVChannel), + DmlPosition: pulsarPos(testVChannel), + }) + + save(path.Join(base, common.DCPrefix, common.ChannelCheckpointPrefix, testVChannel), + pulsarPos(testVChannel)) + + save(path.Join(base, "streamingcoord-meta/pchannel", testPChannel), + &streamingpb.PChannelMeta{ + Channel: &streamingpb.PChannelInfo{Name: testPChannel, Term: 1}, + }) + + save(common.ConsumeCheckpointKey(base, testPChannel), + &streamingpb.WALCheckpoint{ + MessageId: &commonpb.MessageID{WALName: commonpb.WALName_Pulsar, Id: "CAEQAg=="}, + TimeTick: testTimeTick, + RecoveryMagic: 1, + }) + + save(path.Join(common.SegmentAssignPrefix(base, testPChannel), "440000000000000003"), + &streamingpb.SegmentAssignmentMeta{SegmentId: testSegID, Vchannel: testVChannel}) + + return &fixture{base: base, comp: &ComponentReset{client: testKV, basePath: base}} +} + +func (f *fixture) load(t *testing.T, key string, m proto.Message) { + t.Helper() + val, err := testKV.Load(context.Background(), key) + require.NoError(t, err) + require.NoError(t, proto.Unmarshal([]byte(val), m)) +} + +// assertMsgID compares message ids by value, treating nil and empty as the same +// thing: protobuf drops a zero-length bytes field entirely, so woodpecker's +// earliest id reads back as nil. Milvus makes the same equivalence — it tests +// len(MsgID) != 0, never MsgID != nil. +func assertMsgID(t *testing.T, want, got []byte, msgAndArgs ...any) { + t.Helper() + if len(want) == 0 { + assert.Empty(t, got, msgAndArgs...) + return + } + assert.Equal(t, want, got, msgAndArgs...) +} + +func TestResetRewritesEveryPositionKind(t *testing.T) { + f := seed(t, "reset-all", commonpb.SegmentState_Flushed) + + err := f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + }) + require.NoError(t, err) + + want, err := buildWALPosition("woodpecker") + require.NoError(t, err) + + t.Run("consume-checkpoint", func(t *testing.T) { + cp := &streamingpb.WALCheckpoint{} + f.load(t, common.ConsumeCheckpointKey(f.base, testPChannel), cp) + assert.Equal(t, commonpb.WALName_WoodPecker, cp.GetMessageId().GetWALName()) + assert.Equal(t, want.msgID.Id, cp.GetMessageId().GetId()) + assert.Equal(t, testTimeTick, cp.GetTimeTick(), "TimeTick is a global TSO and must survive") + assert.EqualValues(t, 1, cp.GetRecoveryMagic(), "recovery magic must survive") + assert.Nil(t, cp.GetAlterWalState(), "any interrupted online switch must be cleared") + }) + + t.Run("channel-cp", func(t *testing.T) { + pos := &msgpb.MsgPosition{} + f.load(t, path.Join(f.base, common.DCPrefix, common.ChannelCheckpointPrefix, testVChannel), pos) + assertMsgID(t, want.raw, pos.GetMsgID()) + assert.Equal(t, commonpb.WALName_WoodPecker, pos.GetWALName(), + "WALName must be rewritten too, otherwise milvus decodes the new id with the old codec") + assert.Equal(t, testTimeTick, pos.GetTimestamp()) + assert.Equal(t, testVChannel, pos.GetChannelName()) + }) + + t.Run("segment positions", func(t *testing.T) { + seg := &datapb.SegmentInfo{} + f.load(t, path.Join(f.base, common.DCPrefix, common.SegmentMetaPrefix, + "440000000000000001", "440000000000000002", "440000000000000003"), seg) + for name, pos := range map[string]*msgpb.MsgPosition{ + "start": seg.GetStartPosition(), "dml": seg.GetDmlPosition(), + } { + assertMsgID(t, want.raw, pos.GetMsgID(), name) + assert.Equal(t, commonpb.WALName_WoodPecker, pos.GetWALName(), name) + assert.Equal(t, testTimeTick, pos.GetTimestamp(), name) + } + }) + + t.Run("collection start positions", func(t *testing.T) { + coll := &etcdpb.CollectionInfo{} + f.load(t, path.Join(f.base, common.DBCollectionMetaPrefix, "1", "440000000000000001"), coll) + require.Len(t, coll.GetStartPositions(), 1) + assert.Equal(t, testPChannel, coll.GetStartPositions()[0].GetKey()) + assertMsgID(t, want.raw, coll.GetStartPositions()[0].GetData()) + }) + + t.Run("segment-assign dropped", func(t *testing.T) { + keys, _, err := testKV.LoadWithPrefix(context.Background(), + common.SegmentAssignPrefix(f.base, testPChannel)+"/") + require.NoError(t, err) + assert.Empty(t, keys, "stale growing-segment allocations must not survive a rewind") + }) +} + +func TestResetDryRunWritesNothing(t *testing.T) { + f := seed(t, "reset-dry", commonpb.SegmentState_Flushed) + key := path.Join(f.base, common.DCPrefix, common.ChannelCheckpointPrefix, testVChannel) + + before := &msgpb.MsgPosition{} + f.load(t, key, before) + + err := f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + TargetWAL: "woodpecker", + }) + require.NoError(t, err) + + after := &msgpb.MsgPosition{} + f.load(t, key, after) + assert.Equal(t, commonpb.WALName_Pulsar, after.GetWALName(), "dry run must not touch anything") + assert.Equal(t, before.GetMsgID(), after.GetMsgID()) + + keys, _, err := testKV.LoadWithPrefix(context.Background(), + common.SegmentAssignPrefix(f.base, testPChannel)+"/") + require.NoError(t, err) + assert.Len(t, keys, 1, "dry run must not delete either") +} + +func TestResetRefusesGrowingSegments(t *testing.T) { + f := seed(t, "reset-growing", commonpb.SegmentState_Growing) + + err := f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + }) + require.Error(t, err) + assert.ErrorContains(t, err, "growing segment") + + // and nothing was written before the refusal + cp := &streamingpb.WALCheckpoint{} + f.load(t, common.ConsumeCheckpointKey(f.base, testPChannel), cp) + assert.Equal(t, commonpb.WALName_Pulsar, cp.GetMessageId().GetWALName()) +} + +func TestResetAllowGrowingOverride(t *testing.T) { + f := seed(t, "reset-growing-ok", commonpb.SegmentState_Growing) + + err := f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + AllowGrowing: true, + }) + require.NoError(t, err) + + cp := &streamingpb.WALCheckpoint{} + f.load(t, common.ConsumeCheckpointKey(f.base, testPChannel), cp) + assert.Equal(t, commonpb.WALName_WoodPecker, cp.GetMessageId().GetWALName()) +} + +// Running twice must be a no-op the second time. Before this was enforced the +// plan still listed every key, which reads as "the first run did not take". +func TestResetIsIdempotent(t *testing.T) { + f := seed(t, "reset-idem", commonpb.SegmentState_Flushed) + param := func() *ResetCheckpointParam { + return &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + } + } + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), param())) + + writes, err := f.comp.plan(context.Background(), param(), mustPos(t)) + require.NoError(t, err) + assert.Empty(t, writes, "second run must find nothing left to rewrite") + + deletes, err := f.comp.planCleanup(context.Background(), param()) + require.NoError(t, err) + assert.Empty(t, deletes, "and nothing left to delete") + + // still applies cleanly + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), param())) +} + +func mustPos(t *testing.T) *walPosition { + t.Helper() + pos, err := buildWALPosition("woodpecker") + require.NoError(t, err) + return pos +} + +// A run that dies after the channel checkpoints but before the consume +// checkpoint leaves the two sides disagreeing. Milvus panics on startup in that +// state (verified end to end), so the command has to name it rather than just +// listing keys. +func TestResetDetectsSplitStateAndConverges(t *testing.T) { + ctx := context.Background() + f := seed(t, "reset-split", commonpb.SegmentState_Flushed) + param := func() *ResetCheckpointParam { + return &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + } + } + + // no split before anything ran + _, split := f.comp.detectSplitState(ctx, param()) + assert.False(t, split, "a consistent instance is not a split state") + + require.NoError(t, f.comp.ResetCheckpointCommand(ctx, param())) + + // rewind only the consume checkpoint, reproducing a partial run + cp := &streamingpb.WALCheckpoint{} + f.load(t, common.ConsumeCheckpointKey(f.base, testPChannel), cp) + cp.MessageId = &commonpb.MessageID{WALName: commonpb.WALName_Pulsar, Id: "CAEQAg=="} + require.NoError(t, common.SaveConsumeCheckpoint(ctx, testKV, f.base, testPChannel, cp)) + + detail, split := f.comp.detectSplitState(ctx, param()) + require.True(t, split, "channel-cp and consume-checkpoint now disagree") + assert.Contains(t, detail, "WoodPecker") + assert.Contains(t, detail, "Pulsar") + + // re-running must fix exactly the missing side and then converge + writes, err := f.comp.plan(ctx, param(), mustPos(t)) + require.NoError(t, err) + require.Len(t, writes, 1, "only the consume checkpoint is left to write") + assert.Equal(t, kindConsumeCP, writes[0].kind) + + require.NoError(t, f.comp.ResetCheckpointCommand(ctx, param())) + _, split = f.comp.detectSplitState(ctx, param()) + assert.False(t, split, "re-run converges the split state") +} + +// The two sides are keyed differently — channel-cp is per vchannel, the consume +// checkpoint is per pchannel — so their key counts differ on any real instance. +// Comparing counts instead of WAL names reports every healthy instance as split. +func TestDetectSplitStateIgnoresKeyCounts(t *testing.T) { + ctx := context.Background() + f := seed(t, "reset-counts", commonpb.SegmentState_Flushed) + param := &ResetCheckpointParam{TargetWAL: "woodpecker"} + + // add a second vchannel checkpoint on the same pchannel: 2 channel-cp keys + // against 1 consume-checkpoint key, both still Pulsar. + extra := pulsarPos(testPChannel + "_440000000000000001v1") + data, err := proto.Marshal(extra) + require.NoError(t, err) + require.NoError(t, testKV.Save(ctx, path.Join(f.base, common.DCPrefix, + common.ChannelCheckpointPrefix, extra.ChannelName), string(data))) + + _, split := f.comp.detectSplitState(ctx, param) + assert.False(t, split, "differing key counts with the same WAL name is not a split state") +} + +// TestResetPreservesReplicateCheckpoint pins the one field in WALCheckpoint that +// this command must not follow the local WAL with. ReplicateCheckpoint is a +// subscription position against a remote cluster, owned by Milvus; switching the +// local WAL says nothing about it, so reset leaves it exactly as found. +func TestResetPreservesReplicateCheckpoint(t *testing.T) { + ctx := context.Background() + f := seed(t, "reset-replicate", commonpb.SegmentState_Flushed) + + remote := &commonpb.ReplicateCheckpoint{ + ClusterId: "source-cluster", + Pchannel: "src-rootcoord-dml_0", + MessageId: &commonpb.MessageID{WALName: commonpb.WALName_Pulsar, Id: "CAEQBg=="}, + TimeTick: testTimeTick - 1, + } + key := common.ConsumeCheckpointKey(f.base, testPChannel) + data, err := proto.Marshal(&streamingpb.WALCheckpoint{ + MessageId: &commonpb.MessageID{WALName: commonpb.WALName_Pulsar, Id: "CAEQAg=="}, + TimeTick: testTimeTick, + RecoveryMagic: 1, + ReplicateCheckpoint: remote, + }) + require.NoError(t, err) + require.NoError(t, testKV.Save(ctx, key, string(data))) + + require.NoError(t, f.comp.ResetCheckpointCommand(ctx, &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + })) + + cp := &streamingpb.WALCheckpoint{} + f.load(t, key, cp) + + // the local position did move + assert.Equal(t, commonpb.WALName_WoodPecker, cp.GetMessageId().GetWALName()) + // ...and the remote one did not + assert.True(t, proto.Equal(remote, cp.GetReplicateCheckpoint()), + "ReplicateCheckpoint is Milvus-owned remote state and must survive untouched, got %v", + cp.GetReplicateCheckpoint()) +} + +// TestResetRunsWithReplicationConfigured guards against reintroducing a hard +// refusal on replicating instances. Replication topology is not this command's +// concern: the two clusters are independent, and Milvus puts no such gate on its +// own WAL switch. +func TestResetRunsWithReplicationConfigured(t *testing.T) { + ctx := context.Background() + f := seed(t, "reset-replicate-cfg", commonpb.SegmentState_Flushed) + + save := func(key string, m proto.Message) { + data, err := proto.Marshal(m) + require.NoError(t, err) + require.NoError(t, testKV.Save(ctx, key, string(data))) + } + save(path.Join(f.base, "streamingcoord-meta/replicate-configuration"), + &streamingpb.ReplicateConfigurationMeta{ + ReplicateConfiguration: &commonpb.ReplicateConfiguration{ + Clusters: []*commonpb.MilvusCluster{ + {ClusterId: "source-cluster"}, {ClusterId: "by-dev"}, + }, + }, + }) + save(path.Join(f.base, "streamingcoord-meta/replicating-pchannel", testPChannel), + &streamingpb.ReplicatePChannelMeta{SourceChannelName: "src-rootcoord-dml_0"}) + + require.NoError(t, f.comp.preflight(ctx, &ResetCheckpointParam{TargetWAL: "woodpecker"}), + "a replicating instance must still be resettable") +} diff --git a/states/etcd/reset/cleanup.go b/states/etcd/reset/cleanup.go new file mode 100644 index 00000000..f561b563 --- /dev/null +++ b/states/etcd/reset/cleanup.go @@ -0,0 +1,133 @@ +package reset + +import ( + "context" + "fmt" + "path" + + "github.com/cockroachdb/errors" + + "github.com/milvus-io/birdwatcher/states/etcd/common" + "github.com/milvus-io/birdwatcher/states/kv" +) + +// Prefixes holding state that is only meaningful against the WAL we are leaving +// behind. None of them carry data: they are allocations, tombstones and caches +// that Milvus rebuilds on startup. +const ( + collectionTargetPrefix = "queryCoord-Collection-Target" +) + +// plannedDelete is a key or subtree removed as part of the switch. +type plannedDelete struct { + kind string + key string + withPrefix bool + reason string +} + +const ( + kindSegmentAssign = "segment-assign" + kindChannelRemoval = "channel-removal" + kindQueryCoordCache = "querycoord-target-cache" +) + +var deleteOrder = []string{kindSegmentAssign, kindChannelRemoval, kindQueryCoordCache} + +// planCleanup lists the adjacent state to drop. Growing segments are already +// ruled out by preflight, so every segment-assign record left here is stale. +func (c *ComponentReset) planCleanup(ctx context.Context, p *ResetCheckpointParam) ([]plannedDelete, error) { + var deletes []plannedDelete + + channels, err := common.ListWALDistribution(ctx, c.client, c.basePath, p.PChannel) + if err != nil { + return nil, errors.Wrap(err, "failed to list wal distribution") + } + for _, ch := range channels { + pchannel := ch.GetChannel().GetName() + meta, err := common.ListWALRecoveryStorage(ctx, c.client, c.basePath, pchannel) + if err != nil { + return nil, errors.Wrapf(err, "failed to read recovery storage of %s", pchannel) + } + n := 0 + for _, segs := range meta.Segments { + n += len(segs) + } + for _, segs := range meta.RedundantSegments { + n += len(segs) + } + if n == 0 { + continue + } + deletes = append(deletes, plannedDelete{ + kind: kindSegmentAssign, + key: common.SegmentAssignPrefix(c.basePath, pchannel), + withPrefix: true, + reason: fmt.Sprintf("%d stale growing-segment allocation(s) pinned to the old WAL", n), + }) + } + + // Only whole-instance runs may drop these: they are not partitioned by + // pchannel, so removing them during a single-pchannel run would disturb + // channels the operator did not ask us to touch. + if p.PChannel == "" { + candidates := []plannedDelete{ + { + kind: kindChannelRemoval, + key: path.Join(c.basePath, common.DCPrefix, common.ChannelRemovalPrefix), + withPrefix: true, + reason: "stale channel removal markers would block re-watch", + }, + { + kind: kindQueryCoordCache, + key: path.Join(c.basePath, collectionTargetPrefix), + withPrefix: true, + reason: "cached query targets embed old-WAL positions; querycoord rebuilds them", + }, + } + for _, d := range candidates { + n, err := c.countKeys(ctx, d.key) + if err != nil { + return nil, err + } + if n == 0 { + continue + } + d.reason = fmt.Sprintf("%d key(s); %s", n, d.reason) + deletes = append(deletes, d) + } + } + + return deletes, nil +} + +// countKeys reports how many keys live under a prefix, so the plan only lists +// deletions that actually remove something. +func (c *ComponentReset) countKeys(ctx context.Context, prefix string) (int, error) { + keys, _, err := c.client.LoadWithPrefix(ctx, prefix+"/", kv.WithKeysOnly()) + if err != nil { + return 0, errors.Wrapf(err, "failed to scan %s", prefix) + } + return len(keys), nil +} + +func (c *ComponentReset) applyCleanup(ctx context.Context, deletes []plannedDelete) error { + for _, kind := range deleteOrder { + for _, d := range deletes { + if d.kind != kind { + continue + } + var err error + if d.withPrefix { + err = c.client.RemoveWithPrefix(ctx, d.key) + } else { + err = c.client.Remove(ctx, d.key) + } + if err != nil { + return errors.Wrapf(err, "failed to delete %s (%s)", d.key, d.kind) + } + fmt.Printf("deleted %s (%s)\n", d.key, d.kind) + } + } + return nil +} diff --git a/states/etcd/reset/component.go b/states/etcd/reset/component.go new file mode 100644 index 00000000..f0f218ee --- /dev/null +++ b/states/etcd/reset/component.go @@ -0,0 +1,25 @@ +package reset + +import ( + "github.com/milvus-io/birdwatcher/configs" + "github.com/milvus-io/birdwatcher/states/kv" +) + +// ComponentReset hosts the `reset` command family. +// +// Unlike `repair`, which fixes metadata that is broken, `reset` deliberately +// discards metadata that is still valid — so every command here requires the +// Milvus cluster to be stopped first. +type ComponentReset struct { + client kv.MetaKV + config *configs.Config + basePath string +} + +func NewComponent(cli kv.MetaKV, config *configs.Config, basePath string) *ComponentReset { + return &ComponentReset{ + client: cli, + config: config, + basePath: basePath, + } +} diff --git a/states/etcd/reset/preflight.go b/states/etcd/reset/preflight.go new file mode 100644 index 00000000..721f3768 --- /dev/null +++ b/states/etcd/reset/preflight.go @@ -0,0 +1,76 @@ +package reset + +import ( + "context" + "fmt" + "sort" + + "github.com/cockroachdb/errors" + + "github.com/milvus-io/birdwatcher/models" + "github.com/milvus-io/birdwatcher/states/etcd/common" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb" +) + +// preflight refuses to touch anything while the instance is in a state where a +// position reset would silently destroy data. Each check either passes or +// explains what the operator has to do first. +func (c *ComponentReset) preflight(ctx context.Context, p *ResetCheckpointParam) error { + if err := c.checkNoGrowingSegments(ctx, p); err != nil { + return err + } + return c.checkNoPendingBroadcast(ctx) +} + +// checkNoGrowingSegments refuses to run while growing segments exist: their rows +// only live in the old MQ, so resetting positions would drop them for good. +func (c *ComponentReset) checkNoGrowingSegments(ctx context.Context, p *ResetCheckpointParam) error { + segments, err := common.ListSegments(ctx, c.client, c.basePath, func(s *models.Segment) bool { + return s.State == commonpb.SegmentState_Growing && c.inScope(p, s.InsertChannel) + }) + if err != nil { + return errors.Wrap(err, "failed to list segments") + } + if len(segments) == 0 { + return nil + } + + ids := make([]int64, 0, len(segments)) + for _, s := range segments { + ids = append(ids, s.ID) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + + if !p.AllowGrowing { + return errors.Newf("found %d growing segment(s) %v: their rows are still only in the old MQ. "+ + "Flush every collection before switching, or pass --allow-growing to accept losing them", len(ids), ids) + } + fmt.Printf("WARNING: continuing with %d growing segment(s) %v; their unflushed rows will be lost\n", len(ids), ids) + return nil +} + +// checkNoPendingBroadcast refuses to run while a DDL broadcast is half-applied. +// Rewinding positions would replay it against a WAL that no longer has it. +func (c *ComponentReset) checkNoPendingBroadcast(ctx context.Context) error { + tasks, err := common.ListWalBroadcast(ctx, c.client, c.basePath) + if err != nil { + // no streaming metadata at all is fine — pre-2.6 instances have none + if errors.Is(err, common.ErrBroadcastTaskNotFound) { + return nil + } + return errors.Wrap(err, "failed to list wal broadcast tasks") + } + + pending := 0 + for _, t := range tasks { + if t.GetState() != streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE { + pending++ + } + } + if pending > 0 { + return errors.Newf("found %d in-flight broadcast task(s): a DDL is half-applied and the "+ + "instance is inconsistent. Bring Milvus up once to let it settle, then stop it and retry", pending) + } + return nil +} diff --git a/states/etcd/reset/wal_position.go b/states/etcd/reset/wal_position.go new file mode 100644 index 00000000..1ed84512 --- /dev/null +++ b/states/etcd/reset/wal_position.go @@ -0,0 +1,112 @@ +package reset + +import ( + "encoding/base64" + "encoding/binary" + "fmt" + "strings" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/cockroachdb/errors" + wplog "github.com/zilliztech/woodpecker/woodpecker/log" + + bwpulsar "github.com/milvus-io/birdwatcher/mq/pulsar" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus/pkg/v3/streaming/util/message" +) + +// walPosition is the target WAL's earliest position, expressed in the two +// encodings Milvus persists: +// +// raw -> msgpb.MsgPosition.MsgID (channel-cp, segment & collection positions) +// msgID -> commonpb.MessageID (streamingnode consume-checkpoint) +// +// Earliest is a sentinel the broker resolves, so it can be built offline without +// connecting to the target MQ. +// +// There is deliberately no "latest" counterpart. A persisted checkpoint is always +// consumed through DeliverPolicyStartFrom (see the delegator adaptor in +// internal/distributed/streaming/msgstream_adaptor.go), never as a DeliverPolicy_Latest +// policy — so a "latest" sentinel written here becomes a position that does not +// exist, and the reader tails it forever. Skipping stale data on the target topic +// is already handled: Milvus filters the stream with DeliverFilterTimeTickGTE +// against the timestamp we preserve, so starting from earliest cannot replay +// anything older than the checkpoint. +type walPosition struct { + walName commonpb.WALName + raw []byte + msgID *commonpb.MessageID +} + +func (p *walPosition) String() string { + return fmt.Sprintf("%s/earliest", p.walName.String()) +} + +// buildWALPosition resolves --target-wal into the earliest position of that WAL. +func buildWALPosition(targetWAL string) (*walPosition, error) { + wal := strings.ToLower(strings.TrimSpace(targetWAL)) + if wal == "wp" { + wal = "woodpecker" + } + + switch wal { + case "woodpecker": + id := &wplog.LogMessageId{SegmentId: 0, EntryId: 0} + return newBase64Position(commonpb.WALName_WoodPecker, id.Serialize()), nil + + case "pulsar": + return newBase64Position(commonpb.WALName_Pulsar, + bwpulsar.SerializePulsarMsgID(pulsar.EarliestMessageID())), nil + + case "kafka": + // Offset 0 rather than the OffsetBeginning sentinel (-2): kafkaID.Marshal + // encodes with EncodeInt64 while unmarshalMessageID decodes with + // DecodeUint64, which rejects the minus sign (pkg/v2 walimpls/impls/kafka/ + // message_id.go). A negative offset written here would be unreadable. On + // the fresh topic this command targets, 0 is the earliest offset. + return newInt64Position(commonpb.WALName_Kafka, 0), nil + + case "rocksmq", "rockmq": + return newInt64Position(commonpb.WALName_RocksMQ, 0), nil + + case "": + return nil, errors.New("--target-wal is required (woodpecker, pulsar, kafka, rocksmq)") + default: + return nil, errors.Newf("unsupported --target-wal %q, expect one of woodpecker, pulsar, kafka, rocksmq", targetWAL) + } +} + +// newBase64Position builds a position whose consume-checkpoint encoding is +// base64 over the same bytes stored raw in MsgPosition.MsgID. +func newBase64Position(name commonpb.WALName, raw []byte) *walPosition { + return &walPosition{ + walName: name, + raw: raw, + msgID: &commonpb.MessageID{ + WALName: name, + Id: base64.StdEncoding.EncodeToString(raw), + }, + } +} + +// newInt64Position builds a position for the MQs whose message id is a bare +// int64 (kafka offset, rocksmq sequence). +func newInt64Position(name commonpb.WALName, value int64) *walPosition { + return &walPosition{ + walName: name, + raw: serializeInt64(value), + msgID: &commonpb.MessageID{ + WALName: name, + Id: message.EncodeInt64(value), + }, + } +} + +// serializeInt64 encodes a bare int64 message id the same way Milvus does for +// kafka and rocksmq (8 bytes, little endian) — see pkg/v2 mqwrapper +// SerializeKafkaID / SerializeRmqID. +func serializeInt64(value int64) []byte { + b := make([]byte, 8) + binary.LittleEndian.PutUint64(b, uint64(value)) + return b +} diff --git a/states/etcd/reset/wal_position_test.go b/states/etcd/reset/wal_position_test.go new file mode 100644 index 00000000..a494ff72 --- /dev/null +++ b/states/etcd/reset/wal_position_test.go @@ -0,0 +1,90 @@ +package reset + +import ( + "encoding/base64" + "encoding/binary" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + wplog "github.com/zilliztech/woodpecker/woodpecker/log" + + "github.com/milvus-io/birdwatcher/states/etcd/common" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" +) + +// The consume-checkpoint id we write must be decodable by the same unmarshaler +// Milvus uses. birdwatcher mirrors that logic in common.GetMessageIDString, so a +// round-trip through it proves the encoding is right rather than merely stable. +func TestWALPositionRoundTripsThroughDecoder(t *testing.T) { + cases := []struct { + wal string + decoderKey string + wantDecode string + }{ + {"woodpecker", "wp", "0/0"}, + {"wp", "wp", "0/0"}, + {"kafka", "kafka", "0"}, + {"rocksmq", "rmq", "0"}, + } + + for _, tc := range cases { + t.Run(tc.wal, func(t *testing.T) { + pos, err := buildWALPosition(tc.wal) + require.NoError(t, err) + + got := common.GetMessageIDString(tc.decoderKey, pos.msgID.Id) + assert.Equal(t, tc.wantDecode, got, + "consume-checkpoint id must decode back to the expected position") + }) + } +} + +func TestWALPositionWoodpeckerEncodings(t *testing.T) { + earliest, err := buildWALPosition("woodpecker") + require.NoError(t, err) + assert.Equal(t, commonpb.WALName_WoodPecker, earliest.walName) + + // raw bytes are what goes into msgpb.MsgPosition.MsgID + id, err := wplog.DeserializeLogMessageId(earliest.raw) + require.NoError(t, err) + assert.Equal(t, int64(0), id.SegmentId) + assert.Equal(t, int64(0), id.EntryId) + + // consume-checkpoint id is base64 over exactly those bytes + decoded, err := base64.StdEncoding.DecodeString(earliest.msgID.Id) + require.NoError(t, err) + assert.Equal(t, earliest.raw, decoded) + assert.Equal(t, commonpb.WALName_WoodPecker, earliest.msgID.WALName) +} + +func TestWALPositionInt64RawIsLittleEndian(t *testing.T) { + pos, err := buildWALPosition("kafka") + require.NoError(t, err) + require.Len(t, pos.raw, 8) + assert.Equal(t, int64(0), int64(binary.LittleEndian.Uint64(pos.raw))) +} + +func TestWALPositionRejectsUnknownWAL(t *testing.T) { + _, err := buildWALPosition("nats") + assert.ErrorContains(t, err, "unsupported --target-wal") + + _, err = buildWALPosition("") + assert.ErrorContains(t, err, "--target-wal is required") +} + +// woodpecker's earliest id serializes to zero bytes, because protobuf omits +// zero-valued scalars. Milvus relies on this: msgdispatcher only treats an empty +// MsgID as seekable when WALName says WoodPecker (pkg/mq/msgdispatcher/ +// dispatcher.go). So the empty encoding is correct, but only if we also set +// WALName — which is exactly what birdcatcher's rename path forgets to do. +func TestWoodpeckerEarliestIsEmptyBytesButNamed(t *testing.T) { + pos, err := buildWALPosition("woodpecker") + require.NoError(t, err) + + assert.Empty(t, pos.raw, "woodpecker earliest serializes to zero bytes") + assert.Empty(t, pos.msgID.Id, "and therefore to an empty base64 string") + assert.Equal(t, commonpb.WALName_WoodPecker, pos.walName, + "WALName must still be set, or milvus cannot tell an empty id from an unset one") + assert.Equal(t, commonpb.WALName_WoodPecker, pos.msgID.WALName) +} diff --git a/states/instance.go b/states/instance.go index 35e0fb39..f69e45c9 100644 --- a/states/instance.go +++ b/states/instance.go @@ -15,6 +15,7 @@ import ( "github.com/milvus-io/birdwatcher/states/etcd" "github.com/milvus-io/birdwatcher/states/etcd/remove" "github.com/milvus-io/birdwatcher/states/etcd/repair" + "github.com/milvus-io/birdwatcher/states/etcd/reset" "github.com/milvus-io/birdwatcher/states/etcd/set" "github.com/milvus-io/birdwatcher/states/etcd/show" metakv "github.com/milvus-io/birdwatcher/states/kv" @@ -27,6 +28,7 @@ type InstanceState struct { *show.ComponentShow *remove.ComponentRemove *repair.ComponentRepair + *reset.ComponentReset *set.ComponentSet instanceName string metaPath string @@ -180,6 +182,7 @@ func GetInstanceState(parent *framework.CmdState, cli metakv.MetaKV, instanceNam ComponentShow: show.NewComponent(cli, config, instanceName, metaPath), ComponentRemove: remove.NewComponent(cli, config, instanceName, metaPath), ComponentRepair: repair.NewComponent(cli, config, basePath), + ComponentReset: reset.NewComponent(cli, config, basePath), ComponentSet: set.NewComponent(cli, config, basePath), instanceName: instanceName, metaPath: metaPath, diff --git a/tests/e2e/commands/test_reset.sh b/tests/e2e/commands/test_reset.sh new file mode 100755 index 00000000..3435721f --- /dev/null +++ b/tests/e2e/commands/test_reset.sh @@ -0,0 +1,25 @@ +#!/bin/bash +# Reset Commands E2E Tests (DRY RUN ONLY) +# `reset checkpoint` rewrites every persisted MQ position so a stopped instance +# can come back up on a different MQ. It must never run against a live cluster, +# so everything here stays in dry-run. + +section "Reset Commands Tests (DRY RUN)" + +# Dry run against each supported target WAL. The compose stack runs a single MQ, +# so these exercise planning and rendering, not an actual switch. +test_optional "reset checkpoint woodpecker dry" "reset checkpoint --target-wal woodpecker --run=false" +test_optional "reset checkpoint pulsar dry" "reset checkpoint --target-wal pulsar --run=false" +test_optional "reset checkpoint rocksmq dry" "reset checkpoint --target-wal rocksmq --run=false" + +# Scoping to a single pchannel must plan too, even when the name matches nothing. +test_optional "reset checkpoint pchannel filter dry" \ + "reset checkpoint --target-wal woodpecker --pchannel by-dev-rootcoord-dml_0 --run=false" + +# Repeated dry runs must be stable: planning never mutates anything. +test_optional "reset checkpoint repeat dry" "reset checkpoint --target-wal woodpecker --run=false" + +# Rejections. These are expected to fail, which test_optional tolerates; they are +# here so a regression that silently accepts them shows up as a behaviour change. +test_optional "reset checkpoint missing target-wal" "reset checkpoint --run=false" +test_optional "reset checkpoint unknown target-wal" "reset checkpoint --target-wal nats --run=false" diff --git a/tests/e2e/test_commands.sh b/tests/e2e/test_commands.sh index ef052df9..d8b66d4f 100755 --- a/tests/e2e/test_commands.sh +++ b/tests/e2e/test_commands.sh @@ -353,6 +353,7 @@ source "${SCRIPT_DIR}/commands/test_instance.sh" # Optionally run repair/remove tests (dry-run mode only) if [[ "${RUN_WRITE_TESTS:-true}" == "true" ]]; then source "${SCRIPT_DIR}/commands/test_repair.sh" + source "${SCRIPT_DIR}/commands/test_reset.sh" source "${SCRIPT_DIR}/commands/test_remove.sh" fi From 949fa361b1ad0adb22ed628e11eac0d1d968107e Mon Sep 17 00:00:00 2001 From: tinswzy Date: Thu, 27 Aug 2026 16:40:50 +0800 Subject: [PATCH 2/5] fix: scope cleanup deletions by pchannel, drop dead broadcast branch Addresses review feedback on the two cleanup/preflight paths. A --pchannel-scoped run used to skip the channel-removal markers and the querycoord target cache wholesale, on the rationale that they "are not partitioned by pchannel". That rationale was wrong for both: datacoord-meta/channel-removal/{channelName} keyed by channel queryCoord-Collection-Target/{collectionID} keyed by collection Skipping the target cache is the one that hurts. querycoord recovers it into its in-memory current target (TargetManager.Recover) and hands the seek positions inside straight to querynodes (the task executor's req.Checkpoint). Those positions carry the old WAL's encoding and decode cleanly, so nothing flags them as stale -- a scoped reset would leave querynodes seeking the new MQ with old-WAL positions. Nor does the split-state guard catch it: detectSplitState is itself scoped by --pchannel, so it only compares within the reset shard. Both prefixes are now filtered per key instead: removal markers by channel name, targets by "does this collection own any vchannel in scope". A collection sharded across several pchannels keeps positions for all of them in one target, so touching any one vchannel dirties the whole entry and it has to go. Dropping the cache costs only a rebuild from datacoord, which an ordinary restart does anyway. Whole-instance behaviour is unchanged -- everything is in scope. Also drop the ErrBroadcastTaskNotFound branch in checkNoPendingBroadcast. ListWalBroadcast never returns that sentinel; only ListWalBroadcastByID does. A missing prefix already yields an empty slice, which passes the pending check below, so the branch was unreachable and its comment described behaviour the code does not have. Tests: whole-instance run drops everything; scoped run drops the reset shard's marker and target while leaving another pchannel's alone; dry-run deletes nothing. The scoped test fails against the previous skip-wholesale behaviour. Signed-off-by: tinswzy Co-Authored-By: Claude Opus 5 (1M context) --- states/etcd/reset/cleanup.go | 127 ++++++++++++++++++++------- states/etcd/reset/cleanup_test.go | 140 ++++++++++++++++++++++++++++++ states/etcd/reset/preflight.go | 6 +- 3 files changed, 236 insertions(+), 37 deletions(-) create mode 100644 states/etcd/reset/cleanup_test.go diff --git a/states/etcd/reset/cleanup.go b/states/etcd/reset/cleanup.go index f561b563..4151ee5e 100644 --- a/states/etcd/reset/cleanup.go +++ b/states/etcd/reset/cleanup.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "path" + "strconv" "github.com/cockroachdb/errors" @@ -67,48 +68,108 @@ func (c *ComponentReset) planCleanup(ctx context.Context, p *ResetCheckpointPara }) } - // Only whole-instance runs may drop these: they are not partitioned by - // pchannel, so removing them during a single-pchannel run would disturb - // channels the operator did not ask us to touch. - if p.PChannel == "" { - candidates := []plannedDelete{ - { - kind: kindChannelRemoval, - key: path.Join(c.basePath, common.DCPrefix, common.ChannelRemovalPrefix), - withPrefix: true, - reason: "stale channel removal markers would block re-watch", - }, - { - kind: kindQueryCoordCache, - key: path.Join(c.basePath, collectionTargetPrefix), - withPrefix: true, - reason: "cached query targets embed old-WAL positions; querycoord rebuilds them", - }, - } - for _, d := range candidates { - n, err := c.countKeys(ctx, d.key) - if err != nil { - return nil, err - } - if n == 0 { - continue - } - d.reason = fmt.Sprintf("%d key(s); %s", n, d.reason) - deletes = append(deletes, d) + removals, err := c.planChannelRemovals(ctx, p) + if err != nil { + return nil, err + } + deletes = append(deletes, removals...) + + targets, err := c.planQueryCoordTargets(ctx, p) + if err != nil { + return nil, err + } + deletes = append(deletes, targets...) + + return deletes, nil +} + +// planChannelRemovals drops the removal markers of the channels in scope. The +// markers are keyed by channel name, so --pchannel narrows them exactly. +func (c *ComponentReset) planChannelRemovals(ctx context.Context, p *ResetCheckpointParam) ([]plannedDelete, error) { + prefix := path.Join(c.basePath, common.DCPrefix, common.ChannelRemovalPrefix) + keys, _, err := c.client.LoadWithPrefix(ctx, prefix+"/", kv.WithKeysOnly()) + if err != nil { + return nil, errors.Wrapf(err, "failed to scan %s", prefix) + } + + var deletes []plannedDelete + for _, key := range keys { + if !c.inScope(p, path.Base(key)) { + continue } + deletes = append(deletes, plannedDelete{ + kind: kindChannelRemoval, + key: key, + reason: "stale channel removal marker would block re-watch", + }) + } + return deletes, nil +} + +// planQueryCoordTargets drops the cached query target of every collection with +// at least one vchannel in scope. +// +// querycoord recovers this cache into its in-memory current target and hands the +// seek positions inside it straight to querynodes (TargetManager.Recover -> +// task executor's req.Checkpoint). Those positions carry the old WAL's encoding +// and decode cleanly, so nothing detects them as stale — a rewound instance would +// seek the new MQ with old-WAL positions. Dropping the cache costs only a rebuild +// from datacoord, which is what an ordinary restart does anyway. +// +// A collection sharded across several pchannels keeps positions for all of them +// in one target, so touching any one of its vchannels dirties the whole entry. +func (c *ComponentReset) planQueryCoordTargets(ctx context.Context, p *ResetCheckpointParam) ([]plannedDelete, error) { + colls, err := common.ListCollections(ctx, c.client, c.basePath) + if err != nil { + return nil, errors.Wrap(err, "failed to list collections") } + var deletes []plannedDelete + for _, coll := range colls { + id := coll.GetProto().GetID() + key := path.Join(c.basePath, collectionTargetPrefix, strconv.FormatInt(id, 10)) + if !c.anyChannelInScope(p, coll.GetProto().GetVirtualChannelNames()) { + continue + } + // a single key, not a subtree + exists, err := c.keyExists(ctx, key) + if err != nil { + return nil, err + } + if !exists { + continue + } + deletes = append(deletes, plannedDelete{ + kind: kindQueryCoordCache, + key: key, + reason: fmt.Sprintf("collection %d: cached query target embeds old-WAL positions; querycoord rebuilds it", id), + }) + } return deletes, nil } -// countKeys reports how many keys live under a prefix, so the plan only lists +// keyExists reports whether a single key is present, so the plan only lists // deletions that actually remove something. -func (c *ComponentReset) countKeys(ctx context.Context, prefix string) (int, error) { - keys, _, err := c.client.LoadWithPrefix(ctx, prefix+"/", kv.WithKeysOnly()) +func (c *ComponentReset) keyExists(ctx context.Context, key string) (bool, error) { + _, err := c.client.Load(ctx, key) if err != nil { - return 0, errors.Wrapf(err, "failed to scan %s", prefix) + if errors.Is(err, kv.ErrKeyNotFound) { + return false, nil + } + return false, errors.Wrapf(err, "failed to read %s", key) + } + return true, nil +} + +// anyChannelInScope reports whether any of the collection's vchannels is covered +// by --pchannel. +func (c *ComponentReset) anyChannelInScope(p *ResetCheckpointParam, vchannels []string) bool { + for _, vchannel := range vchannels { + if c.inScope(p, vchannel) { + return true + } } - return len(keys), nil + return false } func (c *ComponentReset) applyCleanup(ctx context.Context, deletes []plannedDelete) error { diff --git a/states/etcd/reset/cleanup_test.go b/states/etcd/reset/cleanup_test.go new file mode 100644 index 00000000..866e6b72 --- /dev/null +++ b/states/etcd/reset/cleanup_test.go @@ -0,0 +1,140 @@ +package reset + +import ( + "context" + "path" + "strconv" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + + "github.com/milvus-io/birdwatcher/framework" + "github.com/milvus-io/birdwatcher/states/etcd/common" + "github.com/milvus-io/birdwatcher/states/kv" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/schemapb" + "github.com/milvus-io/milvus/pkg/v3/proto/etcdpb" + "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb" +) + +// a second shard, so scoping can be observed rather than assumed +const ( + otherPChannel = "by-dev-rootcoord-dml_1" + otherVChannel = "by-dev-rootcoord-dml_1_440000000000000009v0" + otherCollID = int64(440000000000000009) +) + +func collectionTargetKey(base string, collID int64) string { + return path.Join(base, collectionTargetPrefix, strconv.FormatInt(collID, 10)) +} + +func channelRemovalKey(base, channel string) string { + return path.Join(base, common.DCPrefix, common.ChannelRemovalPrefix, channel) +} + +// seedCleanup extends the base fixture with a second pchannel and with the two +// kinds of adjacent state that used to be skipped entirely on a scoped run. +func seedCleanup(t *testing.T, name string) *fixture { + t.Helper() + ctx := context.Background() + f := seed(t, name, commonpb.SegmentState_Flushed) + + save := func(key string, m proto.Message) { + data, err := proto.Marshal(m) + require.NoError(t, err) + require.NoError(t, testKV.Save(ctx, key, string(data))) + } + + save(path.Join(f.base, common.DBCollectionMetaPrefix, "1", strconv.FormatInt(otherCollID, 10)), + &etcdpb.CollectionInfo{ + ID: otherCollID, + Schema: &schemapb.CollectionSchema{Name: "other"}, + VirtualChannelNames: []string{otherVChannel}, + PhysicalChannelNames: []string{otherPChannel}, + }) + save(path.Join(f.base, "streamingcoord-meta/pchannel", otherPChannel), + &streamingpb.PChannelMeta{ + Channel: &streamingpb.PChannelInfo{Name: otherPChannel, Term: 1}, + }) + save(common.ConsumeCheckpointKey(f.base, otherPChannel), + &streamingpb.WALCheckpoint{ + MessageId: &commonpb.MessageID{WALName: commonpb.WALName_Pulsar, Id: "CAEQAg=="}, + TimeTick: testTimeTick, + RecoveryMagic: 1, + }) + + // channel-removal markers are keyed by channel name... + require.NoError(t, testKV.Save(ctx, channelRemovalKey(f.base, testVChannel), "removed")) + require.NoError(t, testKV.Save(ctx, channelRemovalKey(f.base, otherVChannel), "removed")) + // ...and query targets by collection id + require.NoError(t, testKV.Save(ctx, collectionTargetKey(f.base, testCollID), "target")) + require.NoError(t, testKV.Save(ctx, collectionTargetKey(f.base, otherCollID), "target")) + + return f +} + +func exists(t *testing.T, key string) bool { + t.Helper() + _, err := testKV.Load(context.Background(), key) + if err == nil { + return true + } + require.ErrorIs(t, err, kv.ErrKeyNotFound) + return false +} + +// TestCleanupDropsTargetsOnWholeInstanceRun is the baseline: with no --pchannel, +// every marker and cached target goes. +func TestCleanupDropsTargetsOnWholeInstanceRun(t *testing.T) { + f := seedCleanup(t, "cleanup-all") + + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + })) + + assert.False(t, exists(t, collectionTargetKey(f.base, testCollID))) + assert.False(t, exists(t, collectionTargetKey(f.base, otherCollID))) + assert.False(t, exists(t, channelRemovalKey(f.base, testVChannel))) + assert.False(t, exists(t, channelRemovalKey(f.base, otherVChannel))) +} + +// TestCleanupScopesTargetsToPChannel pins the fix for the stale-target bug: a +// --pchannel run used to skip both prefixes wholesale, leaving the reset +// collection's cached target full of old-WAL seek positions. querycoord recovers +// that cache verbatim and hands the positions to querynodes, so it has to go — +// while collections outside the scope must be left alone. +func TestCleanupScopesTargetsToPChannel(t *testing.T) { + f := seedCleanup(t, "cleanup-scoped") + + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + PChannel: testPChannel, + })) + + assert.False(t, exists(t, collectionTargetKey(f.base, testCollID)), + "the reset collection's cached target embeds old-WAL positions and must be dropped") + assert.False(t, exists(t, channelRemovalKey(f.base, testVChannel)), + "the reset channel's removal marker must be dropped") + + assert.True(t, exists(t, collectionTargetKey(f.base, otherCollID)), + "a collection outside --pchannel must not be touched") + assert.True(t, exists(t, channelRemovalKey(f.base, otherVChannel)), + "a channel outside --pchannel must not be touched") +} + +// TestCleanupDryRunDeletesNothing guards the dry-run contract for the delete +// half of the plan, which is easy to bypass when adding new deletions. +func TestCleanupDryRunDeletesNothing(t *testing.T) { + f := seedCleanup(t, "cleanup-dry") + + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + TargetWAL: "woodpecker", + })) + + assert.True(t, exists(t, collectionTargetKey(f.base, testCollID))) + assert.True(t, exists(t, channelRemovalKey(f.base, testVChannel))) +} diff --git a/states/etcd/reset/preflight.go b/states/etcd/reset/preflight.go index 721f3768..d0c95756 100644 --- a/states/etcd/reset/preflight.go +++ b/states/etcd/reset/preflight.go @@ -53,12 +53,10 @@ func (c *ComponentReset) checkNoGrowingSegments(ctx context.Context, p *ResetChe // checkNoPendingBroadcast refuses to run while a DDL broadcast is half-applied. // Rewinding positions would replay it against a WAL that no longer has it. func (c *ComponentReset) checkNoPendingBroadcast(ctx context.Context) error { + // An instance with no streaming metadata at all — a pre-2.6 one, say — simply + // yields an empty list here, which passes the check below. tasks, err := common.ListWalBroadcast(ctx, c.client, c.basePath) if err != nil { - // no streaming metadata at all is fine — pre-2.6 instances have none - if errors.Is(err, common.ErrBroadcastTaskNotFound) { - return nil - } return errors.Wrap(err, "failed to list wal broadcast tasks") } From dee668b771d8ee4e34bf1525bbf99a25a02e04d8 Mon Sep 17 00:00:00 2001 From: tinswzy Date: Thu, 27 Aug 2026 16:52:20 +0800 Subject: [PATCH 3/5] fix: keep sweeping orphan query targets on whole-instance runs Scoping the target cleanup by collection narrowed the whole-instance path by accident: a collection whose meta is already gone leaves its cached target behind, and ListCollections cannot see it, so enumerating collections misses those orphans where the previous RemoveWithPrefix caught them. Restore the prefix-wide delete when --pchannel is absent, and keep the per-collection filtering only for scoped runs. A scoped run cannot attribute an orphan to a pchannel, so it leaves it alone rather than guessing. Signed-off-by: tinswzy Co-Authored-By: Claude Opus 5 (1M context) --- states/etcd/reset/cleanup.go | 28 ++++++++++++++++++++--- states/etcd/reset/cleanup_test.go | 38 +++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/states/etcd/reset/cleanup.go b/states/etcd/reset/cleanup.go index 4151ee5e..7249c843 100644 --- a/states/etcd/reset/cleanup.go +++ b/states/etcd/reset/cleanup.go @@ -106,8 +106,9 @@ func (c *ComponentReset) planChannelRemovals(ctx context.Context, p *ResetCheckp return deletes, nil } -// planQueryCoordTargets drops the cached query target of every collection with -// at least one vchannel in scope. +// planQueryCoordTargets drops cached query targets: the whole prefix on a +// whole-instance run, otherwise every collection with at least one vchannel in +// scope. // // querycoord recovers this cache into its in-memory current target and hands the // seek positions inside it straight to querynodes (TargetManager.Recover -> @@ -119,6 +120,27 @@ func (c *ComponentReset) planChannelRemovals(ctx context.Context, p *ResetCheckp // A collection sharded across several pchannels keeps positions for all of them // in one target, so touching any one of its vchannels dirties the whole entry. func (c *ComponentReset) planQueryCoordTargets(ctx context.Context, p *ResetCheckpointParam) ([]plannedDelete, error) { + // A whole-instance run drops the entire prefix. Enumerating collections + // instead would miss orphan targets — a collection whose meta is already + // gone leaves its cached target behind, and ListCollections cannot see it. + prefix := path.Join(c.basePath, collectionTargetPrefix) + if p.PChannel == "" { + keys, _, err := c.client.LoadWithPrefix(ctx, prefix+"/", kv.WithKeysOnly()) + if err != nil { + return nil, errors.Wrapf(err, "failed to scan %s", prefix) + } + if len(keys) == 0 { + return nil, nil + } + return []plannedDelete{{ + kind: kindQueryCoordCache, + key: prefix, + withPrefix: true, + reason: fmt.Sprintf("%d cached query target(s) embed old-WAL positions; querycoord rebuilds them", + len(keys)), + }}, nil + } + colls, err := common.ListCollections(ctx, c.client, c.basePath) if err != nil { return nil, errors.Wrap(err, "failed to list collections") @@ -127,7 +149,7 @@ func (c *ComponentReset) planQueryCoordTargets(ctx context.Context, p *ResetChec var deletes []plannedDelete for _, coll := range colls { id := coll.GetProto().GetID() - key := path.Join(c.basePath, collectionTargetPrefix, strconv.FormatInt(id, 10)) + key := path.Join(prefix, strconv.FormatInt(id, 10)) if !c.anyChannelInScope(p, coll.GetProto().GetVirtualChannelNames()) { continue } diff --git a/states/etcd/reset/cleanup_test.go b/states/etcd/reset/cleanup_test.go index 866e6b72..d39d89eb 100644 --- a/states/etcd/reset/cleanup_test.go +++ b/states/etcd/reset/cleanup_test.go @@ -138,3 +138,41 @@ func TestCleanupDryRunDeletesNothing(t *testing.T) { assert.True(t, exists(t, collectionTargetKey(f.base, testCollID))) assert.True(t, exists(t, channelRemovalKey(f.base, testVChannel))) } + +// TestCleanupSweepsOrphanTargetOnWholeInstanceRun pins why a whole-instance run +// drops the whole prefix instead of enumerating collections: a collection whose +// meta is already gone leaves its cached target behind, and ListCollections +// cannot see it. A scoped run cannot attribute such an orphan to a pchannel, so +// it leaves it alone. +func TestCleanupSweepsOrphanTargetOnWholeInstanceRun(t *testing.T) { + const orphanCollID = int64(440000000000000099) + + t.Run("whole-instance sweeps it", func(t *testing.T) { + f := seedCleanup(t, "cleanup-orphan-all") + require.NoError(t, testKV.Save(context.Background(), + collectionTargetKey(f.base, orphanCollID), "target")) + + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + })) + + assert.False(t, exists(t, collectionTargetKey(f.base, orphanCollID)), + "a target with no surviving collection meta must still be swept") + }) + + t.Run("scoped run leaves it", func(t *testing.T) { + f := seedCleanup(t, "cleanup-orphan-scoped") + require.NoError(t, testKV.Save(context.Background(), + collectionTargetKey(f.base, orphanCollID), "target")) + + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + PChannel: testPChannel, + })) + + assert.True(t, exists(t, collectionTargetKey(f.base, orphanCollID)), + "an orphan cannot be attributed to a pchannel, so a scoped run must not guess") + }) +} From 140319b92aa332b8b2fbca195ad4af2951512813 Mon Sep 17 00:00:00 2001 From: tinswzy Date: Thu, 27 Aug 2026 17:12:02 +0800 Subject: [PATCH 4/5] fix: do not call a single deleted key a prefix in the plan summary Scoped runs delete individual keys while whole-instance runs delete whole subtrees, but the summary hardcoded "prefix(es) to DELETE" for both, overstating the blast radius of a --pchannel run. Report the unit the plan actually uses. Signed-off-by: tinswzy Co-Authored-By: Claude Opus 5 (1M context) --- states/etcd/reset/checkpoint.go | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/states/etcd/reset/checkpoint.go b/states/etcd/reset/checkpoint.go index a08f515a..b2b665ca 100644 --- a/states/etcd/reset/checkpoint.go +++ b/states/etcd/reset/checkpoint.go @@ -407,8 +407,10 @@ func printPlan(writes []plannedWrite, deletes []plannedDelete, pos *walPosition, fmt.Printf(" %-20s %d key(s)\n", kind, byKind[kind]) } for _, kind := range deleteOrder { + // a scoped run deletes individual keys, a whole-instance one whole + // subtrees; saying "prefix" for both would misreport the blast radius if n := countDeleteKind(deletes, kind); n > 0 { - fmt.Printf(" %-20s %d prefix(es) to DELETE\n", kind, n) + fmt.Printf(" %-20s %d %s to DELETE\n", kind, n, deleteUnit(deletes, kind)) } } fmt.Println() @@ -467,6 +469,17 @@ func countKind(writes []plannedWrite, kind string) int { return n } +// deleteUnit names what a plan's deletions of one kind actually remove, so the +// summary does not call a single key a prefix. +func deleteUnit(deletes []plannedDelete, kind string) string { + for _, d := range deletes { + if d.kind == kind && d.withPrefix { + return "prefix(es)" + } + } + return "key(s)" +} + func countDeleteKind(deletes []plannedDelete, kind string) int { n := 0 for _, d := range deletes { From 5301fc6953b20375e76911082043eb3f905bcb58 Mon Sep 17 00:00:00 2001 From: tinswzy Date: Thu, 27 Aug 2026 17:23:13 +0800 Subject: [PATCH 5/5] fix: stop deleting datacoord's channel-removal markers The cleanup treated datacoord-meta/channel-removal as a set of stale tombstones left over from the old WAL. It is not one. Despite the name it is a two-valued flag, and MarkChannelDeleted has no callers in Milvus today -- so every key under the prefix is the NonRemoveFlagTomestone that MarkChannelAdded writes. That write happens once, in the create-collection DDL callback (rootcoord -> datacoord WatchChannels); nothing rebuilds it on restart. It carries no MQ position, so a rewind does not invalidate it. Worse, datacoord reads it back through ChannelExists to decide whether to hold back GC of dropped segments whose DML position is ahead of the channel checkpoint (garbage_collector.go). Deleting it flips that guard off for the channel, permanently and silently. Drop the deletion and say so in the "not touched" notice, alongside the pchannel assignment and channelwatch info. The two remaining cleanups stand: segment-assign records cannot self-heal after a rewind because the new WAL never delivers the insert/flush messages they wait for, and the querycoord target cache embeds old-WAL seek positions that get handed to querynodes. Signed-off-by: tinswzy Co-Authored-By: Claude Opus 5 (1M context) --- states/etcd/reset/checkpoint.go | 6 +++-- states/etcd/reset/cleanup.go | 44 ++++++++----------------------- states/etcd/reset/cleanup_test.go | 43 ++++++++++++++++++++++++------ 3 files changed, 50 insertions(+), 43 deletions(-) diff --git a/states/etcd/reset/checkpoint.go b/states/etcd/reset/checkpoint.go index b2b665ca..d2905473 100644 --- a/states/etcd/reset/checkpoint.go +++ b/states/etcd/reset/checkpoint.go @@ -430,8 +430,10 @@ func printPlan(writes []plannedWrite, deletes []plannedDelete, pos *walPosition, fmt.Printf(" [DELETE %s] %s\n %s\n", d.kind, d.key, d.reason) } } - fmt.Printf("\nNot touched: streamingcoord pchannel assignment and channelwatch info —\n") - fmt.Printf("they carry no MQ position and Milvus rebuilds them on startup.\n") + fmt.Printf("\nNot touched: streamingcoord pchannel assignment, channelwatch info and\n") + fmt.Printf("channel-removal markers — they carry no MQ position. The first two are\n") + fmt.Printf("rebuilt on startup; channel-removal is written once at collection creation\n") + fmt.Printf("and never rebuilt, so removing it would permanently weaken datacoord's GC.\n") if !run { fmt.Printf("\ndry run: nothing written. Re-run with --run=true to apply.\n") fmt.Printf("Make sure Milvus is STOPPED and you have a `backup` before applying.\n") diff --git a/states/etcd/reset/cleanup.go b/states/etcd/reset/cleanup.go index 7249c843..14e53066 100644 --- a/states/etcd/reset/cleanup.go +++ b/states/etcd/reset/cleanup.go @@ -13,8 +13,16 @@ import ( ) // Prefixes holding state that is only meaningful against the WAL we are leaving -// behind. None of them carry data: they are allocations, tombstones and caches -// that Milvus rebuilds on startup. +// behind. Neither carries data: one is a set of growing-segment allocations that +// can no longer be fed, the other a cache of positions Milvus rebuilds. +// +// datacoord's channel-removal prefix looks like a third candidate and is not. +// Despite the name it is a two-valued flag, and MarkChannelDeleted has no callers +// today — so every key under it is the NonRemoveFlagTomestone that MarkChannelAdded +// writes once at collection creation. Nothing rebuilds it, and datacoord reads it +// (ChannelExists) to decide whether to hold back GC of dropped segments whose DML +// position is ahead of the channel checkpoint. Deleting it would silently and +// permanently disable that guard. const ( collectionTargetPrefix = "queryCoord-Collection-Target" ) @@ -29,11 +37,10 @@ type plannedDelete struct { const ( kindSegmentAssign = "segment-assign" - kindChannelRemoval = "channel-removal" kindQueryCoordCache = "querycoord-target-cache" ) -var deleteOrder = []string{kindSegmentAssign, kindChannelRemoval, kindQueryCoordCache} +var deleteOrder = []string{kindSegmentAssign, kindQueryCoordCache} // planCleanup lists the adjacent state to drop. Growing segments are already // ruled out by preflight, so every segment-assign record left here is stale. @@ -68,12 +75,6 @@ func (c *ComponentReset) planCleanup(ctx context.Context, p *ResetCheckpointPara }) } - removals, err := c.planChannelRemovals(ctx, p) - if err != nil { - return nil, err - } - deletes = append(deletes, removals...) - targets, err := c.planQueryCoordTargets(ctx, p) if err != nil { return nil, err @@ -83,29 +84,6 @@ func (c *ComponentReset) planCleanup(ctx context.Context, p *ResetCheckpointPara return deletes, nil } -// planChannelRemovals drops the removal markers of the channels in scope. The -// markers are keyed by channel name, so --pchannel narrows them exactly. -func (c *ComponentReset) planChannelRemovals(ctx context.Context, p *ResetCheckpointParam) ([]plannedDelete, error) { - prefix := path.Join(c.basePath, common.DCPrefix, common.ChannelRemovalPrefix) - keys, _, err := c.client.LoadWithPrefix(ctx, prefix+"/", kv.WithKeysOnly()) - if err != nil { - return nil, errors.Wrapf(err, "failed to scan %s", prefix) - } - - var deletes []plannedDelete - for _, key := range keys { - if !c.inScope(p, path.Base(key)) { - continue - } - deletes = append(deletes, plannedDelete{ - kind: kindChannelRemoval, - key: key, - reason: "stale channel removal marker would block re-watch", - }) - } - return deletes, nil -} - // planQueryCoordTargets drops cached query targets: the whole prefix on a // whole-instance run, otherwise every collection with at least one vchannel in // scope. diff --git a/states/etcd/reset/cleanup_test.go b/states/etcd/reset/cleanup_test.go index d39d89eb..ca9417df 100644 --- a/states/etcd/reset/cleanup_test.go +++ b/states/etcd/reset/cleanup_test.go @@ -97,8 +97,41 @@ func TestCleanupDropsTargetsOnWholeInstanceRun(t *testing.T) { assert.False(t, exists(t, collectionTargetKey(f.base, testCollID))) assert.False(t, exists(t, collectionTargetKey(f.base, otherCollID))) - assert.False(t, exists(t, channelRemovalKey(f.base, testVChannel))) - assert.False(t, exists(t, channelRemovalKey(f.base, otherVChannel))) + + // channel-removal is NOT ours to delete — see the note in cleanup.go + assert.True(t, exists(t, channelRemovalKey(f.base, testVChannel))) + assert.True(t, exists(t, channelRemovalKey(f.base, otherVChannel))) +} + +// TestCleanupNeverTouchesChannelRemoval pins a deletion that was removed after +// review. datacoord's channel-removal prefix reads like a set of stale tombstones +// but is really the NonRemoveFlagTomestone that MarkChannelAdded writes once at +// collection creation; MarkChannelDeleted has no callers today. Nothing rebuilds +// it, and datacoord reads it (ChannelExists) to hold back GC of dropped segments +// whose DML position is ahead of the channel checkpoint — so deleting it would +// permanently and silently weaken that guard. +func TestCleanupNeverTouchesChannelRemoval(t *testing.T) { + for _, tc := range []struct { + name string + pchannel string + }{ + {"whole-instance", ""}, + {"scoped", testPChannel}, + } { + t.Run(tc.name, func(t *testing.T) { + f := seedCleanup(t, "cleanup-keep-removal-"+tc.name) + + require.NoError(t, f.comp.ResetCheckpointCommand(context.Background(), &ResetCheckpointParam{ + ExecutionParam: framework.ExecutionParam{Run: true}, + TargetWAL: "woodpecker", + PChannel: tc.pchannel, + })) + + assert.True(t, exists(t, channelRemovalKey(f.base, testVChannel)), + "the live channel's added-marker must survive") + assert.True(t, exists(t, channelRemovalKey(f.base, otherVChannel))) + }) + } } // TestCleanupScopesTargetsToPChannel pins the fix for the stale-target bug: a @@ -117,13 +150,8 @@ func TestCleanupScopesTargetsToPChannel(t *testing.T) { assert.False(t, exists(t, collectionTargetKey(f.base, testCollID)), "the reset collection's cached target embeds old-WAL positions and must be dropped") - assert.False(t, exists(t, channelRemovalKey(f.base, testVChannel)), - "the reset channel's removal marker must be dropped") - assert.True(t, exists(t, collectionTargetKey(f.base, otherCollID)), "a collection outside --pchannel must not be touched") - assert.True(t, exists(t, channelRemovalKey(f.base, otherVChannel)), - "a channel outside --pchannel must not be touched") } // TestCleanupDryRunDeletesNothing guards the dry-run contract for the delete @@ -136,7 +164,6 @@ func TestCleanupDryRunDeletesNothing(t *testing.T) { })) assert.True(t, exists(t, collectionTargetKey(f.base, testCollID))) - assert.True(t, exists(t, channelRemovalKey(f.base, testVChannel))) } // TestCleanupSweepsOrphanTargetOnWholeInstanceRun pins why a whole-instance run