Device failover support

internal/config/case.go

@@ -1248,7 +1248,7 @@ func (tc *TestCase) validateCluster() error {
 		return fmt.Errorf("case %q: cluster.nodes must be >= 3, got %d", tc.Name, tc.Cluster.Nodes)
 	}
 	switch tc.Cluster.Action {
-	case "", "restart_follower", "restart_leader", "stop_two_recover", "agentless_failover", "cluster_ip_failover":
+	case "", "restart_follower", "restart_leader", "stop_two_recover", "agentless_failover", "device_failover", "cluster_ip_failover":
 	default:
 		return fmt.Errorf("case %q: unknown cluster.action %q", tc.Name, tc.Cluster.Action)
 	}

internal/runner/runner.go

@@ -2993,6 +2993,52 @@ func waitAgentlessCollectingExcluding(ctx context.Context, containers []string,
 	return "", "", false
 }
 
+// waitDeviceCollectingExcluding waits until some node other than `exclude` logs
+// `marker` in its recent window, returning that node's name and readiness. It is
+// the device_failover analogue of waitAgentlessCollectingExcluding, but keys on a
+// caller-supplied marker ("Starting event collection of") instead of "entries
+// from device": DB and file collectors log the collect-start line every cycle
+// even when a correct checkpoint resume forwards ZERO new records, so the
+// rows-forwarded marker would never fire on a clean resume.
+// waitDeviceCollecting waits until the placement device is assigned to a node AND
+// that owner logs `marker` in its recent window — the device_failover baseline
+// analogue of waitAgentlessCollecting. Returns the owner node name, its
+// container, and readiness. Delivery to the receiver is a separate (soft) signal:
+// a cluster collector device's route/sender may not be co-located with the
+// collecting owner, so cross-node payload transport is a known data-plane gap.
+func waitDeviceCollecting(ctx context.Context, subjectName string, containers []string, marker string, deadline time.Time) (string, string, bool) {
+	for time.Now().Before(deadline) {
+		if owner := agentlessDeviceOwner(containers); owner != "" {
+			ownerContainer := fmt.Sprintf("bench-subject-%s-%s", subjectName, owner)
+			if strings.Contains(dockerLogsSince(ownerContainer, "30s"), marker) {
+				return owner, ownerContainer, true
+			}
+		}
+		if sleepCtx(ctx, 5*time.Second) != nil {
+			break
+		}
+	}
+	return agentlessDeviceOwner(containers), "", false
+}
+
+func waitDeviceCollectingExcluding(ctx context.Context, containers []string, exclude, marker string, deadline time.Time) (string, bool) {
+	for time.Now().Before(deadline) {
+		for i, c := range containers {
+			name := strconv.Itoa(i + 1)
+			if name == exclude {
+				continue
+			}
+			if strings.Contains(dockerLogsSince(c, "30s"), marker) {
+				return name, true
+			}
+		}
+		if sleepCtx(ctx, 5*time.Second) != nil {
+			break
+		}
+	}
+	return "", false
+}
+
 // clusterReassignedFrom returns the "Reassigned device ... from <owner> to Y"
 // failover line found across all nodes, and whether one exists. It matches the
 // specific source node (the owner we stopped) so it can NOT be satisfied by the
@@ -3253,6 +3299,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 
 		isAgentless := tc.Cluster.Action == "agentless_failover"
 		isClusterIP := tc.Cluster.Action == "cluster_ip_failover"
+		isDeviceFailover := tc.Cluster.Action == "device_failover"
 		var baselineOK bool
 		var owner, ownerContainer string
 
@@ -3265,6 +3312,25 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				errs = append(errs, "agentless device did not deploy/collect on any node during baseline")
 			}
 			r.sampleDelivery(metricsPort, 0) // soft: downstream E2E delivery (see note above)
+		} else if isDeviceFailover {
+			// A placement collector device (DB / file). The baseline is the
+			// director-level proof — the device is assigned to ONE node and that
+			// owner runs a collect cycle. Receiver delivery is a SOFT signal: a
+			// cluster collector device's route/sender may not be co-located with
+			// the collecting owner, so cross-node payload transport is a known
+			// data-plane gap (the same one agentless_failover soft-logs). The hard
+			// verdict is single-owner placement + automatic failover + resume.
+			fmt.Printf("  baseline: waiting for the device to be placed + run a collect cycle (up to %s)…\n", time.Until(drainDeadline).Round(time.Second))
+			owner, ownerContainer, baselineOK = waitDeviceCollecting(r.ctx, subject.Name, nodes, "Starting event collection of", drainDeadline)
+			if baselineOK {
+				fmt.Printf("  device placed; owner = node %s (%s), collecting ✓\n", owner, ownerContainer)
+			} else {
+				errs = append(errs, "device was not placed / did not run a collect cycle on any node during baseline")
+			}
+			if rm, qerr := r.queryReceiverMetrics(metricsPort, 10*time.Second); qerr == nil {
+				finalCount = rm.LinesReceived
+				fmt.Printf("  (soft) baseline delivery: %s\n", formatCount(finalCount))
+			}
 		} else if isClusterIP {
 			// The VIP test's baseline is simply a formed cluster with a leader (already
 			// asserted by waitClusterReady above). Downstream delivery is a SOFT signal
@@ -3450,6 +3516,76 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				}
 				r.sampleDelivery(metricsPort, preFailover)
 			}
+		case "device_failover":
+			// Generalized placement-device failover (DB / file). Unlike agentless,
+			// the owner is resolved here (the hard-delivery baseline branch above
+			// did not set it), and downstream delivery is asserted HARD (below,
+			// via the receiver's own correctness verdict): a DB/file device
+			// forwards on the reliable direct path, not the cross-node payload
+			// store that makes agentless delivery a soft signal.
+			owner = agentlessDeviceOwner(nodes)
+			if owner == "" {
+				actionOK = false
+				errs = append(errs, "could not determine the placement device owner (no 'Assigned/Reassigned device' log) — failover untestable")
+			} else {
+				ownerContainer = fmt.Sprintf("bench-subject-%s-%s", subject.Name, owner)
+				preFailover := finalCount
+				fmt.Printf("  placement device owner = node %s (%s); stopping it to force a failover…\n", owner, ownerContainer)
+
+				// STOP (not restart): the owner must stay down past the ~15s
+				// heartbeat timeout so the leader reassigns; start it again after.
+				if serr := exec.Command("docker", "stop", "-t", "10", ownerContainer).Run(); serr != nil {
+					actionOK = false
+					errs = append(errs, fmt.Sprintf("docker stop %s failed: %v (disruption did not happen)", ownerContainer, serr))
+				}
+				fmt.Printf("  waiting %s for the leader to detect the down node and reassign the device…\n", settle)
+				if err := sleepCtx(r.ctx, settle); err != nil {
+					return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+				}
+
+				// Reassignment log (informational). The leader logs "Reassigned
+				// device ... from <owner> to Y" when the OWNER was a follower. When
+				// the owner was ALSO the leader (placement can land there), the
+				// newly elected leader may log a fresh "Assigned new device ... to Y"
+				// instead — so the absence of a "from <owner>" line is NOT a
+				// failure. HARD 1 below (a survivor actively collecting the device on
+				// a node != owner) is the robust, placement-form-independent proof
+				// that the device failed over.
+				if line, ok := clusterReassignedFrom(nodes, owner); ok {
+					fmt.Printf("  failover observed: %s\n", line)
+				} else {
+					fmt.Printf("  (note) no 'Reassigned ... from %s to Y' line (owner was likely the leader; relying on the survivor-collecting proof below)\n", owner)
+				}
+				// HARD 1: a survivor (node != stopped owner) actually RAN a poll cycle
+				//   on the reassigned device — proving the device failed over AND the
+				//   new owner resumed from the persisted checkpoint (a clean DB/file
+				//   resume forwards 0 new records, so key on the collect-start line,
+				//   not rows-forwarded). Checked while the old owner is still down so
+				//   placement is firmly on the survivor.
+				collectDeadline := time.Now().Add(settle + 90*time.Second)
+				if collectDeadline.After(runDeadline) {
+					collectDeadline = runDeadline
+				}
+				if newOwner, ok := waitDeviceCollectingExcluding(r.ctx, nodes, owner, "Starting event collection of", collectDeadline); ok {
+					fmt.Printf("  survivor node %s re-homed the device and ran a poll cycle ✓\n", newOwner)
+				} else {
+					actionOK = false
+					errs = append(errs, "no survivor ran a collection cycle after failover — device did not fail over")
+				}
+				// HARD 2: a leader still exists after the loss (cluster stayed healthy).
+				if _, ok := leaderExistsNow(nodes); !ok {
+					actionOK = false
+					errs = append(errs, "no leader after the owning node went down")
+				}
+
+				// Restore full strength.
+				fmt.Printf("  starting node %s again to restore the cluster…\n", ownerContainer)
+				if serr := exec.Command("docker", "start", ownerContainer).Run(); serr != nil {
+					actionOK = false
+					errs = append(errs, fmt.Sprintf("docker start %s failed: %v", ownerContainer, serr))
+				}
+				r.sampleDelivery(metricsPort, preFailover)
+			}
 		case "cluster_ip_failover":
 			// The elected leader must hold the virtual IP; followers must not. Then
 			// restart the leader and assert the IP migrates to the newly elected leader
@@ -3538,6 +3674,23 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 
 	if rm, qerr := r.queryReceiverMetrics(metricsPort, 30*time.Second); qerr == nil {
 		finalCount = rm.LinesReceived
+		// device_failover: receiver delivery is a SOFT signal. A cluster collector
+		// device's route/sender may run on a different node than the collecting
+		// owner, so cross-node payload transport is a known data-plane gap
+		// (mem:// → not-found, natsobj → evicted) — the same gap agentless_failover
+		// soft-logs. When delivery DOES happen (owner co-located with the router),
+		// the receiver's dedup verdict shows the resumed cursor did not re-deliver;
+		// log it either way, but do not gate the cluster verdict on it.
+		if tc.Cluster != nil && tc.Cluster.Action == "device_failover" {
+			switch {
+			case rm.Passed != nil && !*rm.Passed:
+				fmt.Printf("  (soft) post-failover delivery correctness: FAILED — known cluster cross-node data-plane gap (%s)\n", strings.Join(rm.Errors, "; "))
+			case finalCount > 0:
+				fmt.Printf("  (soft) post-failover delivery: %s delivered, receiver dedup clean (resumed cursor did not re-deliver)\n", formatCount(finalCount))
+			default:
+				fmt.Println("  (soft) post-failover delivery: 0 at the receiver (owner not co-located with router — known cluster data-plane gap)")
+			}
+		}
 	}
 
 	elapsed := time.Since(startTime).Seconds()