feat: add reset checkpoint command for offline MQ switch (#523)

Closes #522

Adds `reset checkpoint`, which rewrites every persisted MQ position so a
stopped instance can be brought back up on a different MQ.

```
reset checkpoint --target-wal <woodpecker|pulsar|kafka|rocksmq>
                 [--pchannel <name>] [--allow-growing] [--run=true]
```

## What it writes

| etcd key | field |
| --- | --- |
| `streamingnode-meta/wal/{p}/consume-checkpoint` | `MessageId` |
| `datacoord-meta/channel-cp/{v}` | `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.
Rewriting only channel-cp leaves the old encoding reachable through
segment and collection positions whenever a vchannel's channel-cp is
missing.

It also drops adjacent state that a switch invalidates: `segment-assign`
(stale allocations pinned to the old WAL — the growing-segment gate
guarantees these are garbage), `channel-removal`, and the querycoord
target cache. It deliberately leaves `streamingcoord-meta/pchannel` and
the vchannel schema timeline alone: `PChannelMeta` carries no position,
and the schema timeline must survive.

## Design notes

**New `reset` namespace rather than another `repair` subcommand.**
`repair` fixes broken metadata and its commands run against a live
cluster; this one deliberately discards valid state and requires milvus
to be stopped. Burying a "running this online destroys the cluster" mode
inside a family people run online seemed like the wrong trade.

**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. This is also why there
is no `latest` option — a persisted checkpoint is always consumed
through `DeliverPolicyStartFrom`, 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. I tried it; the collection
never loads.

**`WALName` is set on every position.** Milvus picks the decoder from
it, and for woodpecker it is the only thing separating an empty earliest
id from an unset one.

**kafka uses offset 0, not `OffsetBeginning`.** `kafkaID.Marshal`
encodes with `EncodeInt64` while `unmarshalMessageID` decodes with
`DecodeUint64`, which rejects the minus sign — a negative offset written
into a checkpoint cannot be read back. On the fresh topic this command
targets, 0 is the earliest offset.

**Two preflight gates**, each refusing rather than warning: growing
segments (their rows only exist in the old MQ) and an in-flight
broadcast task (a half-applied DDL).

Replication topology is deliberately *not* gated, and
`WALCheckpoint.ReplicateCheckpoint` is deliberately 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.

**No safe intermediate state.** The consume checkpoint decides which WAL
opens; the channel checkpoints decide which positions get decoded. When
they disagree milvus panics in `getRecoveryInfos`. Reordering only flips
which side is stale, so the command is idempotent, detects a split state
and says so, and the guarantee is "safe to re-run" rather than "safe to
start after a partial run". This is spelled out in the command
description and in the error returned on a failed write.

## Verification

12 test functions using an embedded etcd (`embed.StartEtcd`) — real
protobuf seeded and read back, no mocks.
`tests/e2e/commands/test_reset.sh` follows the existing convention and
covers the dry-run and rejection paths.

Separately, end to end against milvus 3.0.0 with real
etcd/minio/pulsar/kafka, checking row content per row (vectors derived
from the id) plus an ANN search, not just counts:

| source → target | result |
| --- | --- |
| pulsar → woodpecker | pass |
| woodpecker → pulsar | pass |
| pulsar → kafka | pass |
| woodpecker → kafka | pass |
| any → rocksmq | blocked upstream, see below |

Also covered: 4-shard collections, the growing-segment gate (and that
`--allow-growing` really does lose the unflushed rows while flushed data
survives), idempotent re-runs, and recovery from a deliberately
interrupted run. That harness needs pulsar and kafka containers, so it
does not fit the single-MQ `tests/e2e` compose stack; happy to
contribute it separately if that would be useful.

**rocksmq as a target does not work on v3.0.0**, and the cause is
upstream, not here: `pkg/streaming/walimpls/impls/rmq/builder.go` uses
`server.Rmq` without initialising it, and that singleton is only set up
when rocksmq is the selected WAL from process start. A freshly created
rocksmq instance works fine; only a switched-to one fails, with `Rmq
server is nil` and a stalled channel tsafe. milvus main has since added
the `if server.Rmq == nil { InitRocksMQ(...) }` guard.

## Notes for review

- Rebased onto current main, so this is on `pkg/v3` / `go-api/v3` after
#518.
- kafka needs `make birdwatcher_wkafka` (CGO + librdkafka), as with the
existing kafka paths.
- The three new helpers live in
`states/etcd/common/wal_recovery_write.go` alongside the ones #531
added; `wal_recovery_storage.go` is untouched.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: tinswzy <zhenyuan.wei@zilliz.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: tinswzy

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)
+}