test: systematic EC interruption verification — exhaustive model check + deterministic kill matrix (#10764)

* ec: bounded-exhaustive model check of the volume lifecycle

The randomized chaos harness samples the state space; this enumerates
it. The lifecycle is a state machine whose steps mirror the pipelines in
this package, and the checker explores every schedule within the bound:
a crash at every step boundary, an error return running the rollback
(itself crashable at every step), a volume-server restart applying the
startup reconciliation rules in every quiescent state, and the
prescribed restart-based recovery from every crashed state.

Checked in every reachable state: durability (a readable copy always
exists), at most one generation mounted, and — a property the sweep
discipline turns out to guarantee — at most one generation's files on
disk. From every quiescent state the recovery must converge to a clean
volume. Runs in well under a second.

* test: deterministic EC interruption matrix

Enumerate every phase of every interruptible EC operation and kill a
real weed shell exactly when the phase announces itself on the command
output, instead of at a random moment: four encode phases, four decode
phases, and the balance's move phase (set up with -rebalance=false so a
move is guaranteed). Each scenario prepares its precondition, kills at
the marker, runs the prescribed recovery, and verifies every stored byte
still reads back identical.

The interruption recoveries move out of the randomized ops into shared
chaosRun helpers both drivers use.

* test: make the randomized EC chaos walk opt-in

The systematic layers — the interruption matrix and the lifecycle model
check — carry the CI coverage deterministically; the randomized walk
stays for exploratory runs, behind EC_CHAOS_SEED.

* ci: bound the EC integration suite by the job budget, not go test's default

The suite with the interruption matrix runs close to the default 10m
binary timeout on slower runners.

* test: require every interruption-matrix marker to appear

A marker that never prints means a pipeline refactor renamed or dropped
the progress line; silently degenerating into a no-interruption run
would let CI pass without exercising the boundary the scenario names.
Also recheck the marker channel after the wait: a shell that prints and
exits at once makes both channels ready, and select picking the exit
case must not report a printed marker as missed.

.github/workflows/ec-integration-tests.yml

@@ -43,7 +43,10 @@ jobs:
     - name: Run EC Integration Tests
       working-directory: test/erasure_coding
       run: |
-        go test -v
+        # The suite now includes the interruption matrix and runs close to Go's
+        # default 10m binary timeout on slower runners; bound it by the job's
+        # 30m budget instead.
+        go test -v -timeout 25m
 
     - name: Collect server logs on failure
       if: failure()

test/erasure_coding/chaos_lifecycle_test.go

@@ -60,12 +60,13 @@ func TestECChaosLifecycle(t *testing.T) {
 		t.Skip("Skipping EC chaos lifecycle test in short mode")
 	}
 
-	seed := int64(1)
-	if s := os.Getenv("EC_CHAOS_SEED"); s != "" {
-		v, err := strconv.ParseInt(s, 10, 64)
-		require.NoError(t, err, "EC_CHAOS_SEED must be an integer")
-		seed = v
+	seedStr := os.Getenv("EC_CHAOS_SEED")
+	if seedStr == "" {
+		t.Skip("randomized exploration is opt-in: set EC_CHAOS_SEED to run it; " +
+			"systematic coverage lives in TestECInterruptionMatrix and weed/ec's TestECLifecycleModelExhaustive")
 	}
+	seed, err := strconv.ParseInt(seedStr, 10, 64)
+	require.NoError(t, err, "EC_CHAOS_SEED must be an integer")
 	steps := 8
 	if s := os.Getenv("EC_CHAOS_STEPS"); s != "" {
 		v, err := strconv.Atoi(s)
@@ -78,8 +79,8 @@ func TestECChaosLifecycle(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
 	defer cancel()
 
-	cluster, err := startChaosCluster(ctx, testDir)
-	require.NoError(t, err)
+	cluster, clusterErr := startChaosCluster(ctx, testDir)
+	require.NoError(t, clusterErr)
 	defer cluster.Stop()
 
 	require.NoError(t, waitForServer(chaosMasterAddr, 30*time.Second))
@@ -95,49 +96,10 @@ func TestECChaosLifecycle(t *testing.T) {
 	})
 	connectToMasterAndSync(ctx, t, commandEnv)
 
-	r := &chaosRun{
-		t:        t,
-		ctx:      ctx,
-		cluster:  cluster,
-		env:      commandEnv,
-		rng:      mrand.New(mrand.NewSource(seed)),
-		testDir:  testDir,
-		payloads: map[string][]byte{},
-		deleted:  map[string]bool{},
-		fidVol:   map[string]uint32{},
-		volumes:  map[uint32]*chaosVolumeState{},
-	}
+	r := newChaosRun(t, ctx, cluster, commandEnv, testDir, seed)
 	r.relock()
 	defer r.unlockIfHeld()
-
-	// Seed the cluster with payloads across several volumes. The first upload
-	// also creates the collection, which the volume.grow spread below needs.
-	for i := 0; i < 24; i++ {
-		r.uploadOne()
-	}
-	require.GreaterOrEqual(t, len(r.volumes), 2, "seeding should produce at least two volumes")
-	time.Sleep(3 * time.Second)
-
-	// Spread volumes onto every disk of every node: the master only enumerates
-	// disks that already hold data, and shards only spread across enumerated
-	// disks (see TestMultiDiskECBalanceNoShardLoss). Without this a balance can
-	// find no eligible targets and the encode's clump guard aborts.
-	require.Eventually(t, func() bool {
-		spread := nodeVolumeDiskCounts(t, commandEnv)
-		if len(spread) == chaosServerCount && allAtLeast(spread, 2) {
-			return true
-		}
-		for i := 0; i < chaosServerCount; i++ {
-			server := "127.0.0.1:" + chaosVolumePort(i)
-			if spread[server] < 2 {
-				out, gerr := captureCommandOutput(t, shell.Commands[findCommandIndex("volume.grow")],
-					[]string{"-collection", chaosCollection, "-dataNode", server, "-count", "4"}, commandEnv)
-				t.Logf("volume.grow on %s: err=%v output:\n%s", server, gerr, out)
-			}
-		}
-		return false
-	}, 90*time.Second, 2*time.Second, "volumes never spread across >=2 disks on all %d nodes", chaosServerCount)
-
+	r.seedAndSpread()
 	r.verify("seeding")
 
 	// Random schedule. Every op re-verifies the full payload set.
@@ -572,15 +534,22 @@ func (r *chaosRun) opInterruptedEncode() bool {
 	r.t.Logf("killed ec.encode v%d after %v; output so far:\n%s", vid, killAfter, out)
 	r.relock()
 
-	// Recovery. If the kill came after the originals were deleted, the encode
-	// had effectively completed and the volume is EC now; any earlier kill
-	// leaves the regular volume in place (possibly readonly, possibly beside
-	// partial shards), and a re-run must sweep the leftovers and finish.
+	r.recoverInterruptedEncode(vid)
+	return true
+}
+
+// recoverInterruptedEncode is the prescribed recovery after an encode was
+// killed mid-flight: if the kill came after the originals were deleted, the
+// encode had effectively completed and the volume is EC now; any earlier kill
+// leaves the regular volume in place (possibly readonly, possibly beside
+// partial shards), and a re-run must sweep the leftovers and finish.
+func (r *chaosRun) recoverInterruptedEncode(vid uint32) {
+	r.t.Helper()
 	if !r.volumeHasRegularReplica(vid) {
 		r.t.Logf("interrupted encode of volume %d had already completed", vid)
 		r.volumes[vid].encoded = true
 		r.requireSingleGeneration(vid, "interrupted ec.encode (completed)")
-		return true
+		return
 	}
 	out2, err := r.shellCommand("ec.encode",
 		"-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-force")
@@ -598,7 +567,6 @@ func (r *chaosRun) opInterruptedEncode() bool {
 	require.NoError(r.t, err, "recovery ec.encode volume %d after interruption", vid)
 	r.volumes[vid].encoded = true
 	r.requireSingleGeneration(vid, "recovery ec.encode")
-	return true
 }
 
 func (r *chaosRun) opInterruptedDecode() bool {
@@ -613,11 +581,18 @@ func (r *chaosRun) opInterruptedDecode() bool {
 	r.t.Logf("killed ec.decode v%d after %v; output so far:\n%s", vid, killAfter, out)
 	r.relock()
 
-	// Recovery: while any shards remain the decode is unfinished (the kill may
-	// have left a hybrid: regenerated volume plus undeleted shards); a re-run
-	// must complete it. No shards left means the decode had finished — and the
-	// master's view can lag the killed run's final deletions, so a re-run that
-	// finds no shards is also completion, not a failure.
+	r.recoverInterruptedDecode(vid)
+	return true
+}
+
+// recoverInterruptedDecode is the prescribed recovery after a decode was
+// killed mid-flight: while any shards remain the decode is unfinished (the
+// kill may have left a hybrid: regenerated volume plus undeleted shards) and
+// a re-run must complete it. No shards left means the decode had finished —
+// and the master's view can lag the killed run's final deletions, so a re-run
+// that finds no shards is also completion, not a failure.
+func (r *chaosRun) recoverInterruptedDecode(vid uint32) {
+	r.t.Helper()
 	if len(masterEcShardIds(r.env, vid)) > 0 {
 		out2, err := r.shellCommand("ec.decode",
 			"-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-checkMinFreeSpace=false")
@@ -631,7 +606,6 @@ func (r *chaosRun) opInterruptedDecode() bool {
 		r.t.Logf("interrupted decode of volume %d had already completed", vid)
 	}
 	r.volumes[vid].encoded = false
-	return true
 }
 
 func (r *chaosRun) opInterruptedBalance() bool {
@@ -651,12 +625,19 @@ func (r *chaosRun) opInterruptedBalance() bool {
 	r.t.Logf("killed ec.balance after %v; output so far:\n%s", killAfter, out)
 	r.relock()
 
-	// Recovery: an interrupted move leaves a shard copied but not yet deleted
-	// at the source. Re-running the balance must converge — its dedup phase
-	// removes the extra copies — until the replication check is clean. The
-	// master's cleanup of the killed shell's lock can invalidate the lock this
-	// harness re-acquired right after the kill, so a lock error inside the
-	// loop is answered the way an operator would: run lock again and retry.
+	r.recoverInterruptedBalance()
+	return true
+}
+
+// recoverInterruptedBalance is the prescribed recovery after a balance was
+// killed mid-flight: an interrupted move leaves a shard copied but not yet
+// deleted at the source. Re-running the balance must converge — its dedup
+// phase removes the extra copies — until the replication check is clean. The
+// master's cleanup of the killed shell's lock can invalidate the lock this
+// harness re-acquired right after the kill, so a lock error inside the loop
+// is answered the way an operator would: run lock again and retry.
+func (r *chaosRun) recoverInterruptedBalance() {
+	r.t.Helper()
 	require.Eventually(r.t, func() bool {
 		relockOn := func(err error) bool {
 			if err != nil && strings.Contains(err.Error(), "lock") {
@@ -691,7 +672,6 @@ func (r *chaosRun) opInterruptedBalance() bool {
 		}
 		return true
 	}, 120*time.Second, 3*time.Second, "cluster never converged to clean replication after interrupted balance")
-	return true
 }
 
 // classifyOverReplication parses an ec.check.replication -details report and
@@ -722,6 +702,89 @@ func classifyOverReplication(report string) (crossNode, sameNode bool) {
 
 // ── helpers ─────────────────────────────────────────────────────────────────
 
+// newChaosRun wires a run driver over a started cluster. The rng only shapes
+// randomized schedules and payload sizes; deterministic drivers (the
+// interruption matrix) never draw from it beyond seeding uploads.
+func newChaosRun(t *testing.T, ctx context.Context, cluster *chaosCluster, env *shell.CommandEnv, testDir string, seed int64) *chaosRun {
+	return &chaosRun{
+		t:        t,
+		ctx:      ctx,
+		cluster:  cluster,
+		env:      env,
+		rng:      mrand.New(mrand.NewSource(seed)),
+		testDir:  testDir,
+		payloads: map[string][]byte{},
+		deleted:  map[string]bool{},
+		fidVol:   map[string]uint32{},
+		volumes:  map[uint32]*chaosVolumeState{},
+	}
+}
+
+// seedAndSpread uploads the payload set and then spreads volumes onto every
+// disk of every node: the master only enumerates disks that already hold
+// data, and shards only spread across enumerated disks (see
+// TestMultiDiskECBalanceNoShardLoss). Without this a balance can find no
+// eligible targets and the encode's clump guard aborts.
+func (r *chaosRun) seedAndSpread() {
+	r.t.Helper()
+	for i := 0; i < 24; i++ {
+		r.uploadOne()
+	}
+	require.GreaterOrEqual(r.t, len(r.volumes), 2, "seeding should produce at least two volumes")
+	time.Sleep(3 * time.Second)
+
+	require.Eventually(r.t, func() bool {
+		spread := nodeVolumeDiskCounts(r.t, r.env)
+		if len(spread) == chaosServerCount && allAtLeast(spread, 2) {
+			return true
+		}
+		for i := 0; i < chaosServerCount; i++ {
+			server := "127.0.0.1:" + chaosVolumePort(i)
+			if spread[server] < 2 {
+				out, gerr := captureCommandOutput(r.t, shell.Commands[findCommandIndex("volume.grow")],
+					[]string{"-collection", chaosCollection, "-dataNode", server, "-count", "4"}, r.env)
+				r.t.Logf("volume.grow on %s: err=%v output:\n%s", server, gerr, out)
+			}
+		}
+		return false
+	}, 90*time.Second, 2*time.Second, "volumes never spread across >=2 disks on all %d nodes", chaosServerCount)
+}
+
+// ensureRegularVolume returns a tracked volume in the regular (not encoded)
+// state, decoding one if every tracked volume is EC.
+func (r *chaosRun) ensureRegularVolume() uint32 {
+	r.t.Helper()
+	if vid, ok := r.pickVolume(false); ok {
+		return vid
+	}
+	vid, ok := r.pickVolume(true)
+	require.True(r.t, ok, "no volumes tracked at all")
+	out, err := r.shellCommand("ec.decode",
+		"-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-checkMinFreeSpace=false")
+	r.t.Logf("ensureRegular ec.decode v%d output:\n%s", vid, out)
+	require.NoError(r.t, err, "decode volume %d to restore a regular volume", vid)
+	r.volumes[vid].encoded = false
+	return vid
+}
+
+// ensureEncodedVolume returns a tracked volume in the encoded state, encoding
+// one (hdd target, deterministic) if none is.
+func (r *chaosRun) ensureEncodedVolume() uint32 {
+	r.t.Helper()
+	if vid, ok := r.pickVolume(true); ok {
+		return vid
+	}
+	vid, ok := r.pickVolume(false)
+	require.True(r.t, ok, "no volumes tracked at all")
+	out, err := r.shellCommand("ec.encode",
+		"-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-force")
+	r.t.Logf("ensureEncoded ec.encode v%d output:\n%s", vid, out)
+	require.NoError(r.t, err, "encode volume %d", vid)
+	r.volumes[vid].encoded = true
+	r.requireSingleGeneration(vid, "ensureEncodedVolume")
+	return vid
+}
+
 func (r *chaosRun) pickVolume(encoded bool) (uint32, bool) {
 	var candidates []uint32
 	for vid, st := range r.volumes {

test/erasure_coding/systematic_interruption_test.go

@@ -0,0 +1,232 @@
+package erasure_coding
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"github.com/seaweedfs/seaweedfs/weed/shell"
+)
+
+// TestECInterruptionMatrix is the systematic counterpart of the randomized
+// chaos harness: instead of killing operations at random times, it enumerates
+// every phase of every interruptible EC operation and kills a real weed shell
+// exactly when that phase announces itself on the command's output. Each
+// (operation, phase) pair is one deterministic scenario: prepare the
+// precondition, kill at the phase marker, run the prescribed recovery, and
+// verify that every stored byte still reads back identical.
+//
+// The phase markers are the progress lines the pipelines print; each marker
+// below names the code that prints it. A marker is the START of its phase, so
+// killing on it lands the interruption inside that phase — deterministically
+// per phase, byte-exact timing within the phase left to the scheduler. Every
+// scenario REQUIRES its marker to appear: a marker that never prints means a
+// pipeline refactor renamed or dropped the progress line, and silently
+// degenerating into a no-interruption run would let CI pass without
+// exercising the boundary the scenario names. Update the marker with the
+// pipeline.
+//
+// Complementing this, weed/ec's TestECLifecycleModelExhaustive explores every
+// schedule of the same pipelines — crashes between every step, rollback
+// paths, restarts — exhaustively on a model whose steps mirror the code.
+func TestECInterruptionMatrix(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping EC interruption matrix in short mode")
+	}
+
+	testDir := t.TempDir()
+	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Minute)
+	defer cancel()
+
+	cluster, err := startChaosCluster(ctx, testDir)
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	require.NoError(t, waitForServer(chaosMasterAddr, 30*time.Second))
+	for i := 0; i < chaosServerCount; i++ {
+		require.NoError(t, waitForServer("127.0.0.1:"+chaosVolumePort(i), 30*time.Second))
+	}
+	time.Sleep(8 * time.Second)
+
+	commandEnv := shell.NewCommandEnv(&shell.ShellOptions{
+		Masters:        stringPtr(chaosMasterAddr),
+		GrpcDialOption: grpc.WithInsecure(),
+		FilerGroup:     stringPtr("default"),
+	})
+	connectToMasterAndSync(ctx, t, commandEnv)
+
+	r := newChaosRun(t, ctx, cluster, commandEnv, testDir, 1)
+	r.relock()
+	defer r.unlockIfHeld()
+	r.seedAndSpread()
+	r.verify("seeding")
+
+	// One scenario per (operation, phase marker). The preparation puts a
+	// dedicated volume into the operation's precondition state, so scenarios
+	// stay independent even though they share the cluster.
+	type scenario struct {
+		op      string
+		marker  string // printed by ↓
+		printed string // the code that prints it, for the reader
+	}
+	encodeScenarios := []scenario{
+		{"ec.encode", "markVolumeReadonly ", "weed/ec markVolumeReplicaWritable"},
+		{"ec.encode", "generateEcShards ", "weed/ec generateEcShards"},
+		{"ec.encode", "mount ", "weed/ec MountEcShards"},
+		{"ec.encode", "Deleting original volumes", "weed/ec ProcessEcEncodeBatch, the commit boundary"},
+	}
+	decodeScenarios := []scenario{
+		{"ec.decode", " shard locations:", "weed/ec DoEcDecode, before collect"},
+		{"ec.decode", "generateNormalVolume", "weed/ec generateNormalVolume"},
+		{"ec.decode", "unmount ec volume", "weed/ec unmountAndDeleteEcShards, teardown begin"},
+		{"ec.decode", "delete ec volume", "weed/ec unmountAndDeleteEcShards, shard deletion"},
+	}
+
+	for _, sc := range encodeScenarios {
+		sc := sc
+		t.Run(fmt.Sprintf("encode@%s", trimMarker(sc.marker)), func(t *testing.T) {
+			vid := r.ensureRegularVolume()
+			r.unlockIfHeld()
+			out, hit := runShellKillAtMarker(t, r.ctx,
+				fmt.Sprintf("ec.encode -volumeId %d -collection %s -force", vid, chaosCollection), sc.marker)
+			r.t.Logf("killed %s at %q (marker hit: %v); output:\n%s", sc.op, sc.marker, hit, out)
+			r.relock()
+			require.True(t, hit, "marker %q never appeared (printed by %s); update the marker with the pipeline", sc.marker, sc.printed)
+			r.recoverInterruptedEncode(vid)
+			r.verify(fmt.Sprintf("encode killed at %q", sc.marker))
+		})
+	}
+	for _, sc := range decodeScenarios {
+		sc := sc
+		t.Run(fmt.Sprintf("decode@%s", trimMarker(sc.marker)), func(t *testing.T) {
+			vid := r.ensureEncodedVolume()
+			r.unlockIfHeld()
+			out, hit := runShellKillAtMarker(t, r.ctx,
+				fmt.Sprintf("ec.decode -volumeId %d -collection %s -checkMinFreeSpace=false", vid, chaosCollection), sc.marker)
+			r.t.Logf("killed %s at %q (marker hit: %v); output:\n%s", sc.op, sc.marker, hit, out)
+			r.relock()
+			require.True(t, hit, "marker %q never appeared (printed by %s); update the marker with the pipeline", sc.marker, sc.printed)
+			r.recoverInterruptedDecode(vid)
+			r.verify(fmt.Sprintf("decode killed at %q", sc.marker))
+		})
+	}
+
+	// Balance: encode without rebalancing first, so the standalone balance is
+	// guaranteed to plan moves and the marker is guaranteed to print.
+	t.Run("balance@moves", func(t *testing.T) {
+		vid := r.ensureRegularVolume()
+		out, err := r.shellCommand("ec.encode",
+			"-volumeId", fmt.Sprintf("%d", vid), "-collection", chaosCollection, "-force", "-rebalance=false")
+		r.t.Logf("clumped encode v%d output:\n%s", vid, out)
+		require.NoError(t, err, "encode without rebalance")
+		r.volumes[vid].encoded = true
+
+		r.unlockIfHeld()
+		killOut, hit := runShellKillAtMarker(t, r.ctx,
+			fmt.Sprintf("ec.balance -collection %s -apply", chaosCollection), "moves ec shard")
+		r.t.Logf("killed ec.balance at move (marker hit: %v); output:\n%s", hit, killOut)
+		r.relock()
+		require.True(t, hit, "the balance never printed a move; the -rebalance=false setup should have guaranteed one")
+		r.recoverInterruptedBalance()
+		r.verify("balance killed at move")
+	})
+
+	r.verify("final")
+	t.Logf("interruption matrix done: %d payloads live, %d deleted, %d volumes tracked",
+		len(r.payloads), len(r.deleted), len(r.volumes))
+}
+
+func trimMarker(m string) string {
+	out := make([]rune, 0, len(m))
+	for _, c := range m {
+		if c == ' ' || c == ':' {
+			continue
+		}
+		out = append(out, c)
+	}
+	return string(out)
+}
+
+// runShellKillAtMarker feeds "lock" plus the command to a weed shell
+// subprocess, scans its combined output live, and kills the process the
+// moment a line containing marker appears. Returns the captured output and
+// whether the marker was seen before the shell exited on its own.
+func runShellKillAtMarker(t *testing.T, ctx context.Context, command, marker string) (string, bool) {
+	t.Helper()
+	weedBinary := findWeedBinary()
+	require.NotEmpty(t, weedBinary, "weed binary not found")
+	cmd := exec.CommandContext(ctx, weedBinary, "shell", "-master="+chaosMasterAddr)
+
+	pr, pw, err := os.Pipe()
+	require.NoError(t, err)
+	cmd.Stdout, cmd.Stderr = pw, pw
+	stdin, err := cmd.StdinPipe()
+	require.NoError(t, err)
+	require.NoError(t, cmd.Start())
+	pw.Close() // the child keeps its dup; ours would hold the reader open
+
+	var buf bytes.Buffer
+	var mu sync.Mutex
+	markerSeen := make(chan struct{})
+	scanDone := make(chan struct{})
+	go func() {
+		defer close(scanDone)
+		signaled := false
+		scanner := bufio.NewScanner(pr)
+		scanner.Buffer(make([]byte, 0, 256*1024), 1024*1024)
+		for scanner.Scan() {
+			line := scanner.Text()
+			mu.Lock()
+			buf.WriteString(line)
+			buf.WriteByte('\n')
+			mu.Unlock()
+			if !signaled && bytesContains(line, marker) {
+				signaled = true
+				close(markerSeen)
+			}
+		}
+	}()
+
+	fmt.Fprintf(stdin, "lock\n%s\n", command)
+	// Stdin stays open: the shell must die by kill, never by graceful EOF —
+	// a graceful exit would release the cluster lock cleanly and dodge the
+	// dead-holder cleanup path this test also exercises.
+	hit := false
+	select {
+	case <-markerSeen:
+		hit = true
+	case <-scanDone: // shell exited before printing the marker
+	case <-time.After(90 * time.Second):
+	}
+	if !hit {
+		// A shell that printed the marker and exited immediately can make both
+		// channels ready at once, and select picks between ready cases at
+		// random; a printed marker must never be reported as missed.
+		select {
+		case <-markerSeen:
+			hit = true
+		default:
+		}
+	}
+	cmd.Process.Kill()
+	cmd.Wait()
+	pr.Close()
+	<-scanDone
+	mu.Lock()
+	defer mu.Unlock()
+	return buf.String(), hit
+}
+
+func bytesContains(line, marker string) bool {
+	return len(marker) > 0 && strings.Contains(line, marker)
+}

weed/ec/ec_lifecycle_model_test.go

@@ -0,0 +1,520 @@
+package ec
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"testing"
+)
+
+// Bounded-exhaustive model check of the EC volume lifecycle.
+//
+// The randomized chaos harness samples the state space; this test enumerates
+// it. The lifecycle of one volume is modeled as a state machine whose
+// transitions mirror, step by step, the real pipelines in this package, and
+// the checker explores EVERY schedule within the bound: every operation, a
+// crash (kill, no cleanup) at every step boundary, an error return (which
+// runs the rollback, itself crashable at every step) at every step boundary,
+// a volume-server restart (running the startup reconciliation rules) in every
+// quiescent state, and every recovery re-run from every crashed state.
+//
+// Two safety invariants are asserted in every reachable state, including
+// mid-operation and post-crash:
+//
+//	I1 durability: the volume's data is always recoverable — the regular
+//	   replica exists, or some single generation holds at least dataShards
+//	   shard files.
+//	I2 single serving generation: at most one generation is ever mounted.
+//	I3 single generation on disk: the sweep-before-generate discipline keeps
+//	   at most one generation's files on disk through every schedule. (An
+//	   externally planted file breaks this by construction; that path is the
+//	   chaos harness's, not the orchestration's.)
+//
+// And one convergence property: from every quiescent state, the prescribed
+// recovery — re-running the interrupted operation, per the restart-not-resume
+// model — terminates in a clean state (a writable regular volume with no EC
+// leftovers, or exactly one complete mounted generation with no source
+// replica and no surplus copies).
+//
+// The model's fidelity contract: each step below names the code it stands
+// for. When the pipeline order changes, this file must change with it.
+//
+//	encode      = ProcessEcEncodeBatch/doEcEncode: markReadonly
+//	              (markVolumeReplicaWritable false) → sweep
+//	              (clearPreexistingEcShards, removes every prior generation) →
+//	              generate (VolumeEcShardsGenerate, a new generation's files) →
+//	              mount (VolumeEcShardsMount, registers it) → balance
+//	              (EcBalance; a crash mid-move leaves a surplus copy, never a
+//	              loss, because moves copy before they delete) → verify
+//	              (verifyEcShardsBeforeDelete) → deleteSource
+//	              (doDeleteVolumesWithLocations, the commit point).
+//	rollback    = rollbackFailedEcEncode, run only on an error return before
+//	              the commit: rollbackSweep (clearPreexistingEcShards again) →
+//	              restoreWritable. A kill runs nothing.
+//	decode      = DoEcDecode: collect (collectEcShards) → generateVolume
+//	              (VolumeEcShardsToVolume) → mountVolume (VolumeMount) →
+//	              verifyDecoded (verifyDecodedVolumeBeforeDelete) → three
+//	              per-location shard deletions (unmountAndDeleteEcShards).
+//	              Forward-only: no rollback.
+//	reconcile   = the volume server startup rules (weed/storage
+//	              disk_location_ec.go): with the source .dat present, a
+//	              partial generation fails validation and is deleted
+//	              (handleFoundEcxFile/validateEcVolume); surviving shard files
+//	              with their sidecars re-register (the #9212 adoption).
+//	dedup       = the balance's dedup phase removing surplus copies
+//	              (ecbalancer "dedup" moves + verifyEcShardOnKeepNode).
+//
+// Deliberately outside the model: multi-replica sources (the pipeline syncs
+// and reduces to one best replica before encoding), per-node shard placement
+// (I1/I2 do not depend on it), and externally planted files (the chaos
+// harness covers adoption of foreign shards).
+
+const (
+	modelDataShards  = 10
+	modelTotalShards = 14
+)
+
+type modelGen struct {
+	files   int  // shard files on disk, whether or not registered
+	mounted bool // registered and serving
+	surplus int  // duplicate copies left by an interrupted balance move
+}
+
+type modelState struct {
+	replica         bool
+	replicaReadonly bool
+	gens            []modelGen // oldest first; a new encode appends
+}
+
+func (s modelState) clone() modelState {
+	c := s
+	c.gens = append([]modelGen(nil), s.gens...)
+	return c
+}
+
+func (s modelState) key() string {
+	var b strings.Builder
+	fmt.Fprintf(&b, "r%v,ro%v", s.replica, s.replicaReadonly)
+	for _, g := range s.gens {
+		fmt.Fprintf(&b, "|f%d,m%v,s%d", g.files, g.mounted, g.surplus)
+	}
+	return b.String()
+}
+
+// dropEmptyGens removes generations with no files left.
+func (s *modelState) dropEmptyGens() {
+	kept := s.gens[:0]
+	for _, g := range s.gens {
+		if g.files > 0 {
+			kept = append(kept, g)
+		}
+	}
+	s.gens = kept
+}
+
+// ── invariants ──────────────────────────────────────────────────────────────
+
+func checkInvariants(s modelState) error {
+	// I1: durability.
+	recoverable := s.replica
+	for _, g := range s.gens {
+		if g.files >= modelDataShards {
+			recoverable = true
+		}
+	}
+	if !recoverable {
+		return fmt.Errorf("I1 violated: no replica and no generation with >= %d shard files", modelDataShards)
+	}
+	// I2: at most one mounted generation.
+	mounted := 0
+	for _, g := range s.gens {
+		if g.mounted {
+			mounted++
+		}
+	}
+	if mounted > 1 {
+		return fmt.Errorf("I2 violated: %d generations mounted", mounted)
+	}
+	// I3: at most one generation's files on disk.
+	withFiles := 0
+	for _, g := range s.gens {
+		if g.files > 0 {
+			withFiles++
+		}
+	}
+	if withFiles > 1 {
+		return fmt.Errorf("I3 violated: %d generations have files on disk", withFiles)
+	}
+	return nil
+}
+
+func isCleanEc(s modelState) bool {
+	if s.replica || len(s.gens) != 1 {
+		return false
+	}
+	g := s.gens[0]
+	return g.files == modelTotalShards && g.mounted && g.surplus == 0
+}
+
+// ── operations as step lists ────────────────────────────────────────────────
+
+// A step mutates the state; ok=false means the step's precondition failed
+// (the real command would return an error at this point).
+type modelStep struct {
+	name  string
+	apply func(*modelState) (ok bool)
+}
+
+func encodeSteps() []modelStep {
+	return []modelStep{
+		{"markReadonly", func(s *modelState) bool {
+			if !s.replica {
+				return false
+			}
+			s.replicaReadonly = true
+			return true
+		}},
+		// The orphan sweep tears down per (node, volume); a crash can land
+		// between nodes, leaving a partially removed generation.
+		{"sweepLoc0", func(s *modelState) bool { sweepLocation(s, 0); return true }},
+		{"sweepLoc1", func(s *modelState) bool { sweepLocation(s, 1); return true }},
+		{"sweepLoc2", func(s *modelState) bool { sweepLocation(s, 2); return true }},
+		// Generation writes the shard files on the generation host; a crash
+		// mid-write leaves a partial, unregistered set beside the .dat, which
+		// the startup validation removes and the next sweep also covers.
+		{"generateHalf", func(s *modelState) bool {
+			if !s.replica {
+				return false
+			}
+			s.gens = append(s.gens, modelGen{files: modelTotalShards / 2})
+			return true
+		}},
+		{"generateRest", func(s *modelState) bool {
+			s.gens[len(s.gens)-1].files = modelTotalShards
+			return true
+		}},
+		{"mount", func(s *modelState) bool {
+			s.gens[len(s.gens)-1].mounted = true
+			return true
+		}},
+		// balance: a completed run changes placement only. Its crash variant
+		// is modeled by crashing between "balanceCopy" and "balanceDelete":
+		// the copy landed, the source deletion did not.
+		{"balanceCopy", func(s *modelState) bool {
+			s.gens[len(s.gens)-1].surplus++
+			return true
+		}},
+		{"balanceDelete", func(s *modelState) bool {
+			s.gens[len(s.gens)-1].surplus--
+			return true
+		}},
+		{"verify", func(s *modelState) bool {
+			return s.gens[len(s.gens)-1].files >= modelDataShards
+		}},
+		{"deleteSource", func(s *modelState) bool {
+			s.replica = false
+			s.replicaReadonly = false
+			return true
+		}},
+	}
+}
+
+// encodeCommitIndex is the index of deleteSource: an error return at or after
+// it must not roll back (the shards are the only copy once it runs).
+func encodeCommitIndex() int { return len(encodeSteps()) - 1 }
+
+// sweepLocation removes one location's share of every generation's files —
+// the per-(node, volume) teardown of clearPreexistingEcShards. The last
+// location clears the remainder, sidecars included.
+func sweepLocation(s *modelState, loc int) {
+	for gi := range s.gens {
+		g := &s.gens[gi]
+		chunk := modelTotalShards / 3
+		if loc == 2 {
+			g.files, g.mounted, g.surplus = 0, false, 0
+		} else if g.files > chunk {
+			g.files -= chunk
+		} else {
+			g.files = 0
+		}
+	}
+	s.dropEmptyGens()
+}
+
+func rollbackSteps() []modelStep {
+	return []modelStep{
+		{"rollbackSweep", func(s *modelState) bool {
+			s.gens = nil
+			return true
+		}},
+		{"restoreWritable", func(s *modelState) bool {
+			if s.replica {
+				s.replicaReadonly = false
+			}
+			return true
+		}},
+	}
+}
+
+func decodeSteps() []modelStep {
+	steps := []modelStep{
+		{"collect", func(s *modelState) bool {
+			for _, g := range s.gens {
+				if g.mounted && g.files >= modelDataShards {
+					return true
+				}
+			}
+			return false
+		}},
+		{"generateVolume", func(s *modelState) bool {
+			s.replica = true
+			return true
+		}},
+		{"mountVolume", func(s *modelState) bool { return true }},
+		{"verifyDecoded", func(s *modelState) bool { return s.replica }},
+	}
+	// Shard deletion is per holder location; three locations model a spread
+	// volume, so a crash can land between any two of them.
+	for i := 0; i < 3; i++ {
+		i := i
+		steps = append(steps, modelStep{fmt.Sprintf("deleteShardsLoc%d", i), func(s *modelState) bool {
+			for gi := range s.gens {
+				g := &s.gens[gi]
+				if !g.mounted && g.files == 0 {
+					continue
+				}
+				// Each location holds roughly a third of the shards; the last
+				// pass clears the remainder and the generation with it.
+				chunk := modelTotalShards / 3
+				if i == 2 {
+					g.files, g.surplus, g.mounted = 0, 0, false
+				} else if g.files > chunk {
+					g.files -= chunk
+				}
+			}
+			s.dropEmptyGens()
+			return true
+		}})
+	}
+	return steps
+}
+
+// reconcile applies the volume-server startup rules after a crash-restart.
+func reconcile(s *modelState) {
+	for gi := range s.gens {
+		g := &s.gens[gi]
+		if s.replica && g.files < modelTotalShards {
+			// .dat present and the local generation is partial: validation
+			// fails and the files are removed (disk_location_ec.go).
+			g.files, g.mounted, g.surplus = 0, false, 0
+			continue
+		}
+		if g.files > 0 {
+			// Surviving files re-register through their sidecars (#9212).
+			g.mounted = true
+		}
+	}
+	s.dropEmptyGens()
+}
+
+// dedupSurplus is the balance dedup phase.
+func dedupSurplus(s *modelState) {
+	for gi := range s.gens {
+		s.gens[gi].surplus = 0
+	}
+}
+
+// ── recovery convergence ────────────────────────────────────────────────────
+
+// runToCompletion applies steps without interruption; false when a
+// precondition fails (the real re-run would error out).
+func runToCompletion(s *modelState, steps []modelStep) bool {
+	for _, st := range steps {
+		if !st.apply(s) {
+			return false
+		}
+	}
+	return true
+}
+
+// recover applies the prescribed recovery for a quiescent state and reports
+// whether it converges to a clean state. Per the restart-not-resume model:
+// a surviving replica means the encode restarts from scratch; no replica
+// means the encode had committed and only surplus cleanup can remain; a
+// mounted generation beside a replica can also be decoded away.
+func recoverModel(s modelState) (modelState, error) {
+	// A crash-restart happens before any operator action; its reconciliation
+	// must itself keep the invariants.
+	reconcile(&s)
+	if err := checkInvariants(s); err != nil {
+		return s, fmt.Errorf("after reconcile: %w", err)
+	}
+
+	if !s.replica {
+		// The encode committed: shards are the volume. Dedup cleans surplus.
+		dedupSurplus(&s)
+		if !isCleanEc(s) {
+			return s, fmt.Errorf("no replica and not clean EC")
+		}
+		return s, nil
+	}
+
+	// Replica survives: re-run the encode from scratch (sweep + regenerate),
+	// then the volume is EC; this is the restart path every interrupted
+	// encode and every leftover-orphan state takes.
+	if !runToCompletion(&s, encodeSteps()) {
+		return s, fmt.Errorf("recovery encode did not complete")
+	}
+	dedupSurplus(&s)
+	if err := checkInvariants(s); err != nil {
+		return s, fmt.Errorf("after recovery encode: %w", err)
+	}
+	if !isCleanEc(s) {
+		return s, fmt.Errorf("recovery encode did not converge to clean EC")
+	}
+	// And a decode must be able to bring it back to a regular volume.
+	if !runToCompletion(&s, decodeSteps()) {
+		return s, fmt.Errorf("decode after recovery did not complete")
+	}
+	if err := checkInvariants(s); err != nil {
+		return s, fmt.Errorf("after decode: %w", err)
+	}
+	if !s.replica || len(s.gens) != 0 {
+		return s, fmt.Errorf("decode after recovery did not converge to a bare regular volume")
+	}
+	return s, nil
+}
+
+// ── exhaustive exploration ──────────────────────────────────────────────────
+
+type trace []string
+
+// explore runs every schedule of one operation from state s: after every
+// step, the schedule may continue, crash (kill: stop, nothing else runs), or
+// error out (rollback runs, itself crashable at every step). Every resulting
+// quiescent state is handed to onQuiescent together with the trace that
+// produced it.
+func exploreOp(t *testing.T, s modelState, opName string, steps []modelStep, commitIndex int, withRollback bool, tr trace, onQuiescent func(modelState, trace)) {
+	t.Helper()
+	// Crash before the first step is the same as never starting.
+	for prefix := 1; prefix <= len(steps); prefix++ {
+		run := s.clone()
+		ok := true
+		for i := 0; i < prefix; i++ {
+			if !steps[i].apply(&run) {
+				ok = false
+				break
+			}
+			if err := checkInvariants(run); err != nil {
+				t.Fatalf("%s after %s: %v\ntrace: %v", opName, steps[i].name, err, append(tr, opName+"/"+steps[i].name))
+			}
+		}
+		if !ok {
+			continue // precondition stopped the op; same as an early error
+		}
+		stepTrace := append(append(trace{}, tr...), fmt.Sprintf("%s..%s", opName, steps[prefix-1].name))
+
+		// Kill here: nothing else runs.
+		onQuiescent(run.clone(), append(append(trace{}, stepTrace...), "KILL"))
+
+		// Error return here: the rollback runs, unless the op has committed;
+		// the rollback itself can be killed at each of its step boundaries.
+		if withRollback && prefix <= commitIndex {
+			rb := rollbackSteps()
+			for rbPrefix := 0; rbPrefix <= len(rb); rbPrefix++ {
+				rbRun := run.clone()
+				for i := 0; i < rbPrefix; i++ {
+					rb[i].apply(&rbRun)
+					if err := checkInvariants(rbRun); err != nil {
+						t.Fatalf("%s rollback after %s: %v\ntrace: %v", opName, rb[i].name, err, stepTrace)
+					}
+				}
+				suffix := "FAIL+rollback-killed"
+				if rbPrefix == len(rb) {
+					suffix = "FAIL+rollback-complete"
+				}
+				onQuiescent(rbRun.clone(), append(append(trace{}, stepTrace...), fmt.Sprintf("%s@%d", suffix, rbPrefix)))
+			}
+		}
+	}
+}
+
+// TestECLifecycleModelExhaustive enumerates every crash and failure schedule
+// of encode and decode up to two chained interrupted operations, checks the
+// safety invariants in every intermediate state, and requires the prescribed
+// recovery to converge from every quiescent state.
+func TestECLifecycleModelExhaustive(t *testing.T) {
+	start := modelState{replica: true}
+
+	visited := map[string]trace{}
+	quiescent := 0
+	schedules := 0
+	var enqueue func(s modelState, depth int, tr trace)
+
+	checkRecovery := func(s modelState, tr trace) {
+		schedules++
+		if _, seen := visited["Q"+s.key()]; seen {
+			return
+		}
+		visited["Q"+s.key()] = tr
+		quiescent++
+		if _, err := recoverModel(s.clone()); err != nil {
+			t.Fatalf("recovery does not converge: %v\nstate: %s\ntrace: %v", err, s.key(), tr)
+		}
+	}
+
+	enqueue = func(s modelState, depth int, tr trace) {
+		checkRecovery(s, tr)
+		if depth == 0 {
+			return
+		}
+		// A crash-restart may happen in any quiescent state before the next
+		// operation; explore both with and without it.
+		restarted := s.clone()
+		reconcile(&restarted)
+		if err := checkInvariants(restarted); err != nil {
+			t.Fatalf("reconcile violated invariants: %v\nstate: %s\ntrace: %v", err, s.key(), tr)
+		}
+		for _, variant := range []struct {
+			st  modelState
+			tag string
+		}{{s, ""}, {restarted, "+restart"}} {
+			st, tag := variant.st, variant.tag
+			if st.replica {
+				exploreOp(t, st, "encode"+tag, encodeSteps(), encodeCommitIndex(), true, tr, func(q modelState, qtr trace) {
+					enqueue(q, depth-1, qtr)
+				})
+			}
+			canDecode := false
+			for _, g := range st.gens {
+				if g.mounted && g.files >= modelDataShards {
+					canDecode = true
+				}
+			}
+			if canDecode {
+				exploreOp(t, st, "decode"+tag, decodeSteps(), len(decodeSteps()), false, tr, func(q modelState, qtr trace) {
+					enqueue(q, depth-1, qtr)
+				})
+			}
+		}
+	}
+
+	// Depth 2: an interrupted operation followed by another interrupted
+	// operation, recovery checked at every quiescent point in between and
+	// after. Deeper chains only revisit already-explored states (the state
+	// space is finite and memoized), which the visited count makes apparent.
+	enqueue(start, 2, trace{"start"})
+
+	// Not spec constants — collapse guards: if a refactor of the model or the
+	// explorer accidentally prunes schedules, these counts crater and this
+	// catches it. Update deliberately when the model changes fidelity.
+	if quiescent < 15 || schedules < 500 {
+		t.Fatalf("suspiciously small exploration: %d distinct quiescent states from %d schedules", quiescent, schedules)
+	}
+	keys := make([]string, 0, len(visited))
+	for k := range visited {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	t.Logf("explored %d schedules reaching %d distinct quiescent states, all recoverable", schedules, quiescent)
+}