fix(wpcli): improve segment inspection and scan diagnostics

cmd/wpcli/README.md

@@ -92,10 +92,11 @@ See the [external Kubernetes configuration example](../../docs/admin-guides/wpcl
 
 ### Log end to end
 - `wp log scan <logName> [--mode quick|raw] [--from-segment N] [--to-segment N]` — sweep every
-  segment of a log and reconcile what metadata claims against what the replicas hold. `quick`
-  (default) reads structure only — one trailer read per sealed segment — and answers "does the
-  shape add up". `raw` reads through the normal path with codec and CRC and answers "how far does
-  a reader actually get". Reports one line per segment plus the log-level findings.
+  segment of a log and reconcile metadata with the replicas' local coverage. `quick`
+  (default) reads structure only; `raw` verifies local blocks with codec and CRC checks.
+  Reports metadata state/lastEntryId, observed coverage, and per-replica ranges, survey
+  stop reasons and request errors. Neither mode verifies pending writes, compacted objects
+  in shared storage, or end-to-end application reads; use `segment probe` for actual reads.
 
   Per-segment verdicts: `ok`, `short` (no replica has some of what metadata claims), `open` (still
   being written), `compacted` (served from one shared object-storage copy, which this sweep does
@@ -166,6 +167,29 @@ on that ask and only starts the refresh, so a lone reader sees a new range on it
 - `wp segment probe <logName> <segmentId>` — ask every replica how far it can read the segment; names a damaged replica failover is covering for
 - `wp segment inspect <logName> <segmentId>` — walk the blocks on every replica: which entries are damaged where, whether the damage is bounded, and what a skip would cost
 
+`log scan`, `segment inspect` and `segment probe` accept an optional paired
+`--bucket` / `--root-path` to select the data storage instance. These are distinct
+from `--meta-prefix`, which selects coordinator metadata. Explicit location is
+useful for empty or unloaded segments whose instance cannot be discovered:
+
+```bash
+wp log scan my-channel --mode raw \
+  --bucket milvus-bucket --root-path milvus/wp \
+  --etcd 127.0.0.1:32379 --meta-prefix milvus-prod/woodpecker
+wp segment probe my-channel 7 --from-entry 50 --max-entries 100 \
+  --bucket milvus-bucket --root-path milvus/wp \
+  --etcd 127.0.0.1:32379 --meta-prefix milvus-prod/woodpecker
+```
+
+Omitting both location flags retains node-side discovery. Providing just one is
+a usage error. A 404 or unavailable node establishes no data coverage; the scan
+reports the per-replica reason rather than treating it as empty. `local_empty`
+confirms no local durable entries, not absence of pending writes. An Active
+segment has no final boundary; confirmed local emptiness alone is not a permanent
+read stall. Failed raw block checksums must not be labelled as local emptiness.
+Sealed/compacted local copies are not authoritative; scan does not verify their
+shared object contents, while probe can actually read them.
+
 ### Logstore runtime
 - `wp logstore segments <node>` — list active segments
 - `wp logstore segment-show <node> --log X --seg Y` — detailed segment view

cmd/wpcli/cmd/address_mapping_test.go

@@ -38,7 +38,7 @@ func TestMappedQuorumRequests(t *testing.T) {
 			case "/admin/logstore/segment/probe":
 				fmt.Fprint(w, `{"first_entry":0,"last_entry":9,"entries_read":10,"outcome":"ok"}`)
 			case "/admin/logstore/segment/inspect":
-				fmt.Fprint(w, `{"survey":{"blocks":[],"lac":9}}`)
+				fmt.Fprint(w, `{"survey":{"blocks":[{"first_entry_id":0,"last_entry_id":9,"status":"ok"}],"lac":9,"stop_reason":"end_of_segment"}}`)
 			case "/admin/logstore/fence":
 				require.Equal(t, http.MethodPost, r.Method)
 				fmt.Fprint(w, `{}`)

cmd/wpcli/cmd/log_scan.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/url"
 	"sort"
 	"strconv"
 	"strings"
@@ -35,12 +36,16 @@ const (
 func newLogScanCommand() *cobra.Command {
 	var flags metaEtcdFlags
 	var mode string
+	var bucket, rootPath string
 	var from, to int64
 	cmd := &cobra.Command{
 		Use:   "scan <logName>",
-		Short: "Sweep a log to check that reading runs through and the data matches its metadata",
+		Short: "Compare metadata with replica local coverage; raw verifies local block checksums",
 		Args:  cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
+			if (bucket == "") != (rootPath == "") {
+				return wperrors.NewUsageError("--bucket and --root-path must be provided together")
+			}
 			if mode != scanModeQuick && mode != scanModeRaw {
 				return wperrors.NewUsageError(fmt.Sprintf("unknown --mode %q: use quick or raw", mode))
 			}
@@ -57,10 +62,12 @@ func newLogScanCommand() *cobra.Command {
 				return err
 			}
 			defer cli.Close()
-			return runLogScan(cmd, cli, conn.kb, res.Client, res.Members, args[0], mode, from, to)
+			return runLogScan(cmd, cli, conn.kb, res.Client, res.Members, args[0], mode, from, to, scanLocation{bucket, rootPath})
 		},
 	}
 	flags.register(cmd)
+	cmd.Flags().StringVar(&bucket, "bucket", "", "Storage bucket; pair with --root-path to locate unloaded or empty segments")
+	cmd.Flags().StringVar(&rootPath, "root-path", "", "Storage instance root path; pair with --bucket")
 	cmd.Flags().StringVar(&mode, "mode", scanModeQuick,
 		"quick reads only structure and metadata; raw reads the data through the real checksums")
 	cmd.Flags().Int64Var(&from, "from-segment", 0, "First segment to scan")
@@ -69,7 +76,7 @@ func newLogScanCommand() *cobra.Command {
 }
 
 func runLogScan(cmd *cobra.Command, cli *clientv3.Client, kb *meta.KeyBuilder,
-	ac *client.Client, members *client.Memberlist, logName, mode string, from, to int64,
+	ac *client.Client, members *client.Memberlist, logName, mode string, from, to int64, location ...scanLocation,
 ) error {
 	ctx, cancel := metaCtx()
 	defer cancel()
@@ -95,7 +102,7 @@ func runLogScan(cmd *cobra.Command, cli *clientv3.Client, kb *meta.KeyBuilder,
 		}
 		segments = append(segments, scanSegment{
 			id: sm.id, state: sm.meta.GetState(), metaLast: sm.meta.GetLastEntryId(),
-			nodes: scanSegmentNodes(ac, members, sm, logMeta.LogId, mode),
+			nodes: scanSegmentNodes(ac, members, sm, logMeta.LogId, mode, location...),
 		})
 	}
 
@@ -130,7 +137,10 @@ func runLogScan(cmd *cobra.Command, cli *clientv3.Client, kb *meta.KeyBuilder,
 	for _, row := range rows {
 		metaLast := strconv.FormatInt(row.MetaLast, 10)
 		if row.MetaLast < 0 {
-			metaLast = "-1 (open)"
+			metaLast = "-1 (empty metadata)"
+			if row.State == "Active" {
+				metaLast = "-1 (open)"
+			}
 		}
 		table = append(table, []string{
 			strconv.FormatInt(row.SegmentID, 10), row.State, metaLast,
@@ -142,6 +152,16 @@ func runLogScan(cmd *cobra.Command, cli *clientv3.Client, kb *meta.KeyBuilder,
 		return renderErr
 	}
 
+	fmt.Fprintln(w, "\nPer-replica inspection (local durable blocks; pending writes and object-storage contents are not verified):")
+	replicaRows := [][]string{}
+	for _, row := range rows {
+		for _, replica := range row.ReplicaDetails {
+			replicaRows = append(replicaRows, []string{fmt.Sprint(row.SegmentID), replica.Node, replica.Result, replica.Coverage, replica.StopReason, replica.Detail})
+		}
+	}
+	if err := output.RenderRowTable(w, []string{"SEGMENT", "NODE", "RESULT", "LOCAL RANGE", "STOP", "DETAIL"}, replicaRows); err != nil {
+		return err
+	}
 	fmt.Fprintln(w)
 	for _, line := range findings {
 		fmt.Fprintf(w, "%s\n", line)
@@ -224,7 +244,9 @@ func readSegmentMetas(ctx context.Context, cli *clientv3.Client, kb *meta.KeyBui
 // reconciliation reads as "not known" rather than as missing data.
 const scanMaxBlocks = 4096
 
-func scanSegmentNodes(ac *client.Client, members *client.Memberlist, sm segmentMeta, logID int64, mode string) []scanNode {
+type scanLocation struct{ bucket, rootPath string }
+
+func scanSegmentNodes(ac *client.Client, members *client.Memberlist, sm segmentMeta, logID int64, mode string, location ...scanLocation) []scanNode {
 	quorum := sm.meta.GetQuorum()
 	if quorum == nil || len(quorum.Nodes) == 0 {
 		return nil
@@ -244,17 +266,22 @@ func scanSegmentNodes(ac *client.Client, members *client.Memberlist, sm segmentM
 
 		path := fmt.Sprintf("/admin/logstore/segment/inspect?log_id=%d&segment_id=%d&max_blocks=%d",
 			logID, sm.id, scanMaxBlocks)
+		if len(location) > 0 && location[0].bucket != "" {
+			path += "&bucket_name=" + url.QueryEscape(location[0].bucket) + "&root_path=" + url.QueryEscape(location[0].rootPath)
+		}
 		if mode == scanModeQuick {
 			path += "&verify=false"
 		}
 		body, status, err := fetchAdminJSONWithStatus(ac.PeerAdminURL(member), path)
 		switch {
 		case err != nil:
 			node.state = posUnreachable
+			node.detail = err.Error()
 			nodes = append(nodes, node)
 			continue
 		case status != 200:
 			node.state = probeStateCannotAnswer
+			node.detail = fmt.Sprintf("HTTP %d: %s", status, adminErrorText(body, status))
 			nodes = append(nodes, node)
 			continue
 		}
@@ -275,10 +302,17 @@ func scanSegmentNodes(ac *client.Client, members *client.Memberlist, sm segmentM
 		}
 		if jsonErr := json.Unmarshal(body, &resp); jsonErr != nil {
 			node.state = posBadResponse
+			node.detail = jsonErr.Error()
+			nodes = append(nodes, node)
+			continue
+		}
+		if resp.Survey.StopReason == "" {
+			node.state, node.detail = posBadResponse, "inspection response is missing survey stop_reason"
 			nodes = append(nodes, node)
 			continue
 		}
 		node.answered, node.stopped = true, resp.Survey.StopReason
+		node.blocksFound = len(resp.Survey.Blocks) > 0
 		node.compactedMark = resp.Local.CompactedMark
 		for _, block := range resp.Survey.Blocks {
 			if block.FirstEntryID < 0 || block.LastEntryID < block.FirstEntryID {
@@ -287,6 +321,7 @@ func scanSegmentNodes(ac *client.Client, members *client.Memberlist, sm segmentM
 			// In raw mode only a block that verified counts as reached; in quick mode nothing was
 			// verified, so the coverage is what the structure claims.
 			if mode == scanModeRaw && block.Status != blockStatusOK {
+				node.detail += fmt.Sprintf("block %d..%d: %s; ", block.FirstEntryID, block.LastEntryID, block.Status)
 				continue
 			}
 			node.coverage = node.coverage.add(entryRange{block.FirstEntryID, block.LastEntryID})

cmd/wpcli/cmd/log_scan_reconcile.go

@@ -40,10 +40,12 @@ const gapNameBudget = 32
 // scanNode is what one replica reported for one segment: the entries its structure claims, or the
 // entries a read actually reached, depending on the mode.
 type scanNode struct {
-	label    string
-	answered bool
-	state    string // why it did not answer, when it did not
-	coverage rangeSet
+	detail      string
+	blocksFound bool
+	label       string
+	answered    bool
+	state       string // why it did not answer, when it did not
+	coverage    rangeSet
 	// stopped is where the node's survey ended, and it decides what the coverage amounts to: a
 	// survey that reached the end of the segment makes the coverage a claim about what this replica
 	// holds, and any other ending makes it a lower bound.
@@ -63,6 +65,11 @@ func (n scanNode) completeClaim() bool {
 	return n.stopped == surveyStopEnd || (n.stopped == surveyStopNoBlocks && !n.compactedMark)
 }
 
+// An empty verified range is not an empty local copy if unverified/damaged blocks were found.
+func (n scanNode) confirmedLocalEmpty() bool {
+	return n.completeClaim() && len(n.coverage) == 0 && !n.blocksFound
+}
+
 // servedElsewhere reports whether this replica's local copy is gone by design, its data now being
 // the object every replica shares.
 func (n scanNode) servedElsewhere() bool {
@@ -78,15 +85,24 @@ type scanSegment struct {
 }
 
 // scanRow is one line of the report.
+type scanReplicaDetail struct {
+	Node       string `json:"node"`
+	Result     string `json:"result"`
+	Coverage   string `json:"local_range"`
+	StopReason string `json:"stop_reason,omitempty"`
+	Detail     string `json:"detail,omitempty"`
+}
+
 type scanRow struct {
-	SegmentID int64  `json:"segment_id"`
-	State     string `json:"state"`
-	MetaLast  int64  `json:"meta_last_entry_id"`
-	Reaches   string `json:"reaches"`
-	Replicas  string `json:"replicas"`
-	Verdict   string `json:"verdict"`
-	Detail    string `json:"detail,omitempty"`
-	verdict   string
+	ReplicaDetails []scanReplicaDetail `json:"replica_details"`
+	SegmentID      int64               `json:"segment_id"`
+	State          string              `json:"state"`
+	MetaLast       int64               `json:"meta_last_entry_id"`
+	Reaches        string              `json:"reaches"`
+	Replicas       string              `json:"replicas"`
+	Verdict        string              `json:"verdict"`
+	Detail         string              `json:"detail,omitempty"`
+	verdict        string
 }
 
 // reconcileScan compares what metadata says against what the replicas reported, and returns one row
@@ -143,7 +159,7 @@ func reconcileScanWithGaps(segments []scanSegment, truncatedThrough, fromSegment
 		return rows, findings, wperrors.NewYellowFindingError(summary)
 	}
 	findings = append(findings, fmt.Sprintf(
-		"%d segments scanned; the log reads through with no problem found.", len(segments),
+		"%d segments scanned; no problem found in inspected local coverage. This does not establish end-to-end readability or verify object-storage contents.", len(segments),
 	))
 	return rows, findings, nil
 }
@@ -159,17 +175,35 @@ func reconcileSegment(segment scanSegment) (scanRow, []string, segmentOutcome) {
 	findings := make([]string, 0, 2)
 
 	held := rangeSet{}
-	answered, settled, elsewhere := 0, false, false
+	answered, settled, elsewhere := 0, true, false
 	silent := make([]string, 0, len(segment.nodes))
 	for _, node := range segment.nodes {
+		result, coverage := node.state, "unknown"
+		if node.answered {
+			coverage = node.coverage.String()
+			if len(node.coverage) == 0 {
+				coverage = "none"
+			}
+			result = "inspected"
+			if node.confirmedLocalEmpty() {
+				result = "local_empty"
+			}
+			if node.servedElsewhere() {
+				result = "compacted"
+			}
+		}
+		row.ReplicaDetails = append(row.ReplicaDetails, scanReplicaDetail{node.label, result, coverage, node.stopped, node.detail})
 		if !node.answered {
-			silent = append(silent, fmt.Sprintf("%s (%s)", node.label, node.state))
+			settled = false
+			reason := node.state
+			if node.detail != "" {
+				reason += "; " + node.detail
+			}
+			silent = append(silent, fmt.Sprintf("%s (%s)", node.label, reason))
 			continue
 		}
 		answered++
-		if node.completeClaim() {
-			settled = true
-		}
+		settled = settled && node.completeClaim()
 		if node.servedElsewhere() {
 			elsewhere = true
 		}
@@ -178,7 +212,7 @@ func reconcileSegment(segment scanSegment) (scanRow, []string, segmentOutcome) {
 	row.Replicas = fmt.Sprintf("%d/%d", answered, len(segment.nodes))
 	if answered == 0 {
 		findings = append(findings, fmt.Sprintf(
-			"segment %d: no replica answered (%s), so nothing is known about it.",
+			"segment %d: no replica answered with a usable inspection result (%s), so nothing is known about it.",
 			segment.id, strings.Join(silent, ", "),
 		))
 		return row, findings, outcomeInconclusive
@@ -195,7 +229,7 @@ func reconcileSegment(segment scanSegment) (scanRow, []string, segmentOutcome) {
 	// read. This sweep does not read the object, so it says that rather than guessing.
 	if elsewhere || segment.state == proto.SegmentState_Sealed {
 		row.Verdict, row.verdict = scanVerdictCompacted, scanVerdictCompacted
-		row.Detail = "served from object storage; local replica copies are not the authority"
+		row.Detail = "served from object storage; object-storage contents not verified; local replica copies are not the authority"
 		return row, findings, outcomeSound
 	}
 	// Retention is deleting this segment's data on purpose. Between the state change and the
@@ -229,8 +263,28 @@ func reconcileSegment(segment scanSegment) (scanRow, []string, segmentOutcome) {
 		// Still open: metadata records no last entry yet, so there is nothing to measure against.
 		// A hole inside what was written is still a hole, and the operator needs its position.
 		row.Verdict, row.verdict = scanVerdictOpen, scanVerdictOpen
+		if len(held) == 0 {
+			allEmpty := len(segment.nodes) > 0 && answered == len(segment.nodes)
+			for _, n := range segment.nodes {
+				allEmpty = allEmpty && n.confirmedLocalEmpty()
+			}
+			if allEmpty {
+				row.Detail = "all replicas confirm no local durable entries; pending writes are not checked"
+			} else {
+				row.Verdict, row.verdict = scanVerdictUnknown, scanVerdictUnknown
+				row.Detail = "no entries observed; some replicas or survey results are inconclusive"
+				findings = append(findings, fmt.Sprintf("segment %d: %s", segment.id, row.Detail))
+				return row, findings, outcomeInconclusive
+			}
+		}
 		if len(held) > 1 {
 			span := rangeSet{}.add(entryRange{held[0].From, held[len(held)-1].To})
+			if !settled {
+				row.Verdict, row.verdict = scanVerdictUnknown, scanVerdictUnknown
+				row.Detail = fmt.Sprintf("observed gap %s; replica coverage is incomplete", span.subtract(held))
+				findings = append(findings, fmt.Sprintf("segment %d: %s; an uninspected range or replica may cover the gap", segment.id, row.Detail))
+				return row, findings, outcomeInconclusive
+			}
 			findings = append(findings, fmt.Sprintf(
 				"segment %d is open and its coverage has a gap at %s — entries are missing in the middle of what was written.",
 				segment.id, span.subtract(held),
@@ -247,13 +301,12 @@ func reconcileSegment(segment scanSegment) (scanRow, []string, segmentOutcome) {
 		return row, findings, outcomeSound
 	}
 
-	// Every replica stopped before the end of its own copy, so the coverage is a lower bound and the
-	// entries past it were never looked at. Calling that a shortfall would report intact data as
-	// missing, which is what a node's own survey bound does to a segment larger than it.
+	// Missing ranges cannot be declared absent while any replica is unavailable or its survey
+	// is incomplete: that replica may still hold the entries absent from observed coverage.
 	if !settled {
-		row.Detail = "survey bounded; coverage is a lower bound"
+		row.Detail = "replica unavailable or survey bounded; coverage is a lower bound"
 		findings = append(findings, fmt.Sprintf(
-			"segment %d: no replica surveyed to the end of the segment (%s), so nothing could be established about %s. Run `wp segment inspect <logName> %d` to survey it without a whole-log sweep's bound.",
+			"segment %d: not every replica surveyed to the end of the segment (%s), so absence could not be established for %s. Run `wp segment inspect <logName> %d` to survey it without a whole-log sweep's bound.",
 			segment.id, boundedReasons(segment.nodes), missing, segment.id,
 		))
 		return row, findings, outcomeInconclusive
@@ -278,6 +331,7 @@ func boundedReasons(nodes []scanNode) string {
 	parts := make([]string, 0, len(nodes))
 	for _, node := range nodes {
 		if !node.answered {
+			parts = append(parts, fmt.Sprintf("%s: %s", node.label, node.state))
 			continue
 		}
 		parts = append(parts, fmt.Sprintf("%s stopped at %s", node.label, node.stopped))