diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index d0372da35..27a44b444 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -157,6 +157,17 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP } }() + // Step 0: Establish start-of-task invariants before any destructive step. + // Verify the plan is complete and clear EC shards left by a prior + // interrupted encode of this volume, so the encode begins from a clean + // slate. Failing here returns before the source is marked readonly or + // copied — nothing to roll back. + t.ReportProgressWithStage(5.0, "Verifying preconditions and clearing stale EC state") + t.GetLogger().Info("Verifying preconditions and clearing stale EC state") + if err := t.ensureCleanEcStart(ctx); err != nil { + return fmt.Errorf("EC preflight failed for volume %d: %w", t.volumeID, err) + } + // Step 1: Mark all replicas readonly, then reconcile them and select the most // complete replica as the encode source. Encoding a stale replica and then // deleting the originals would silently lose entries that exist only on another @@ -197,15 +208,11 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP return fmt.Errorf("failed to generate EC shards: %w", err) } - // Clear partial EC shards left over on destinations from a prior failed - // encode so distributeEcShards' ReceiveFile is not refused by the - // mounted-volume guard. - t.ReportProgressWithStage(55.0, "Clearing stale EC shards on destinations") - t.GetLogger().Info("Clearing stale EC shards on destinations") - if err := t.cleanupStaleEcShards(ctx); err != nil { - t.rollbackReadonly(ctx) - return fmt.Errorf("failed to clear stale EC shards on destinations: %w", err) - } + // Stale EC shards from a prior interrupted encode were already cleared in + // the Step 0 preflight, before the source was marked readonly. The admin + // dedupe key (erasure_coding::) prevents a concurrent + // same-volume encode, so no destination can regain stale shards between + // the preflight and distributeEcShards below. // Delete 0-byte stub replicas left by an interrupted encode before the new // EC files land. A stub shares the _.vif path the EC @@ -219,10 +226,18 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP return fmt.Errorf("failed to remove empty stub replicas: %w", err) } - // Step 4: Distribute shards to destinations + // Step 4: Distribute shards to destinations. + // From here on a failure has written shards to destinations. Until verify + // passes we are not committed to the EC copy, so a failure must roll the + // attempt back — tear down the shards it distributed and restore the + // sources to writable — otherwise a terminally-failed encode (a + // single-attempt job, or the last of a retry series, which has no successor + // to clean up at its Step 0 preflight) strands orphan shards and a source + // fenced readonly. t.ReportProgressWithStage(60.0, "Distributing EC shards to destinations") t.GetLogger().Info("Distributing EC shards to destinations") if err := t.distributeEcShards(shardFiles); err != nil { + t.rollbackDistribute(ctx) return fmt.Errorf("failed to distribute EC shards: %w", err) } @@ -230,6 +245,7 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP t.ReportProgressWithStage(80.0, "Mounting EC shards") t.GetLogger().Info("Mounting EC shards") if err := t.mountEcShards(); err != nil { + t.rollbackDistribute(ctx) return fmt.Errorf("failed to mount EC shards: %w", err) } @@ -238,8 +254,12 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP t.ReportProgressWithStage(85.0, "Verifying EC shards across destinations") t.GetLogger().Info("Verifying EC shards across destinations") if err := t.verifyEcShardsBeforeDelete(ctx); err != nil { + t.rollbackDistribute(ctx) return fmt.Errorf("EC shard verification failed; refusing to delete source volume %d: %w", t.volumeID, err) } + // Past verify the EC copy is recoverable; a Step 7 failure must NOT tear the + // shards down — the remaining source replicas are cleaned by the next + // detection's cleanupOrphanSourceReplicas instead. // Step 7: Delete original volume t.ReportProgressWithStage(90.0, "Deleting original volume") @@ -917,16 +937,98 @@ func replicasPendingDelete(replicas []string, alreadyDeleted map[string]bool) [] return pending } -// cleanupStaleEcShards unmounts and deletes any EC shards still mounted on -// destinations from a previous failed encode of this volume. Targets every -// node we plan to write to (t.targets) plus every node detection saw EC -// shards on (t.sources with ShardIds set), and issues the cleanup over the -// full shard range so a stale topology snapshot — or shards landed by a -// prior attempt that haven't heartbeated yet — cannot leave the -// mounted-volume guard tripped during distributeEcShards. Safe by ordering: -// runs after the source .dat is in the worker's workdir and a full local -// shard set is generated. Per-destination errors are aggregated, not -// short-circuited. +// ensureCleanEcStart runs first, before any destructive step, to establish +// the invariants a fresh encode depends on: +// - a target set exists: an empty or malformed plan must fail here, not +// after the source has been marked readonly and copied; +// - a source replica exists to encode from; +// - no EC shards from a prior interrupted encode of this volume survive on +// the nodes this task will touch. Leftover partial shards trip the +// mounted-volume guard in distributeEcShards' ReceiveFile, are loaded as +// orphans on the next volume-server restart, and make detection refuse the +// volume ("Manual intervention required"). cleanupStaleEcShards blanket- +// wipes this volume's EC state on every touched node regardless of shard +// generation (a retried attempt's shards share this job's encodeTsNs, and +// an interrupted distribute often leaves shards with an unreadable +// generation — a fenced teardown would strand both). +// +// Cleaning at the start (rather than just before distribute) means the encode +// begins from a clean slate and a preflight failure leaves the source +// untouched — there is nothing to roll back. It is safe to delete stale shards +// this early: the source's regular replica still holds the data until the +// post-verify delete in Step 7. +func (t *ErasureCodingTask) ensureCleanEcStart(ctx context.Context) error { + if len(t.targets) == 0 { + return fmt.Errorf("no EC shard targets for volume %d; refusing to mark source readonly", t.volumeID) + } + // A non-empty slice is not enough: a target with an empty Node (or no + // assigned shards) is silently skipped by cleanupStaleEcShards and by + // distributeEcShards, so a plan of only such entries would pass the length + // check and mark the source readonly before failing. Reject any malformed + // target here, before the first destructive step. + for i, target := range t.targets { + if target == nil || target.Node == "" || len(target.ShardIds) == 0 { + return fmt.Errorf("malformed EC shard target %d for volume %d; refusing to mark source readonly", i, t.volumeID) + } + } + if t.server == "" && len(t.getReplicas()) == 0 { + return fmt.Errorf("no source replica for volume %d", t.volumeID) + } + return t.cleanupStaleEcShards(ctx) +} + +// rollbackDistribute undoes a failed attempt that had already begun writing EC +// shards to destinations but had not yet committed to the EC copy (verify not +// passed, so the sources are intact). It tears down the shards this attempt +// distributed and restores the sources to writable, so a terminally-failed +// encode — a single-attempt job, or the last of a retry series — leaves no +// orphan shards and no source stuck readonly. On a retry the next attempt's +// Step 0 preflight would also clear the shards, but the final attempt has no +// successor; this makes every post-distribute failure self-cleaning. +// cleanupStaleEcShards blanket-wipes this volume's EC state regardless of shard +// generation — necessary because an interrupted distribute leaves shards whose +// .vif generation is unreadable, which a fenced teardown would preserve. +// Best-effort and uses a fresh context since the caller's may already be +// cancelled (the very failure that brought us here). +func (t *ErasureCodingTask) rollbackDistribute(_ context.Context) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + if err := t.cleanupStaleEcShards(ctx); err != nil { + // The teardown could not fully clear this volume's EC shards (e.g. an + // unreachable destination, or a shard that failed to unmount). Leave the + // source readonly rather than expose it for writes while stale shards + // linger: a writable source beside mounted stale shards would let reads + // and writes diverge, and orphan cleanup will not remove a writable + // source. The next encode's Step 0 preflight (or an operator) reconciles + // the state once the shards are reachable. + glog.Warningf("rollback: EC shard teardown incomplete for volume %d; leaving source readonly for reconciliation: %v", t.volumeID, err) + return + } + t.rollbackReadonly(ctx) +} + +// cleanupStaleEcShards unmounts and deletes any EC shards for this volume on +// destinations from a previous failed encode. Targets every node we plan to +// write to (t.targets) plus every node detection saw EC shards on (t.sources +// with ShardIds set), and issues the cleanup over the full shard range so a +// stale topology snapshot — or shards landed by a prior attempt that haven't +// heartbeated yet — cannot leave the mounted-volume guard tripped during +// distributeEcShards. Called from the Step 0 preflight (ensureCleanEcStart) and +// from rollbackDistribute. +// +// Teardown is UNFENCED (encodeTsNs=0 -> the server's blanket teardown), which +// wipes every EC artifact for this volume on every disk regardless of +// generation. A generation fence is wrong here for two reasons: (1) a retried +// encode's prior attempt shares this job's encodeTsNs, and the server's fence +// preserves same-or-newer, so a fenced teardown would strand it; (2) shards +// left by an interrupted distribute often have an UNREADABLE .vif generation +// (the sidecar never landed), which the fence also preserves. This is a +// pre-encode / rollback wipe of a volume we are (re)encoding or abandoning, so +// clearing all of its EC state is correct — the admin dedupe key +// (erasure_coding::) guarantees no concurrent newer encode of +// this volume, and the blanket teardown's own replacement check aborts rather +// than clobber a live newer mount. This mirrors the shell ec.encode pre-encode +// cleanup. Per-destination errors are aggregated, not short-circuited. func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error { nodes := make(map[string]struct{}) for _, source := range t.sources { @@ -955,7 +1057,10 @@ func (t *ErasureCodingTask) cleanupStaleEcShards(ctx context.Context) error { "shard_ids": allShards, }).Info("Clearing stale EC shards on destination before re-distribute") - if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, allShards, t.encodeTsNs); err != nil { + // encodeTsNs=0 selects the server's blanket (generation-independent) + // teardown; see the function comment for why the fence is intentionally + // not used here. + if err := unmountAndDeleteEcShards(ctx, t.grpcDialOption, node, t.volumeID, t.collection, allShards, 0); err != nil { cleanupErrors = append(cleanupErrors, fmt.Sprintf("%s: %v", node, err)) t.GetLogger().WithFields(map[string]interface{}{ "volume_id": t.volumeID, diff --git a/weed/worker/tasks/erasure_coding/ec_task_preflight_test.go b/weed/worker/tasks/erasure_coding/ec_task_preflight_test.go new file mode 100644 index 000000000..4fb4a6d50 --- /dev/null +++ b/weed/worker/tasks/erasure_coding/ec_task_preflight_test.go @@ -0,0 +1,231 @@ +package erasure_coding + +import ( + "context" + "net/http" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// Reproduces the swtest interrupted-encode scenario at the task boundary: a +// previous encode of this volume was cut mid-distribute, leaving partial EC +// shards mounted on a destination. When a fresh encode task starts, its Step 0 +// preflight (ensureCleanEcStart) must clear those shards before any +// destructive step, so distribute's ReceiveFile is not refused and no orphan +// shards survive to confuse the volume-server loader or make detection refuse +// the volume. The destination is named only as a target (no source row) to +// prove the preflight reaches the full write set, not just shard-bearing +// sources. +func TestEnsureCleanEcStartClearsStaleShardsBeforeEncode(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const ( + volumeID = uint32(94782) + collection = "ec-preflight-invariant" + ) + + framework.AllocateVolume(t, grpcClient, volumeID, collection) + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 9478200, 0x9478BEEF) + upResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, + []byte("payload-for-preflight-stale-ec-cleanup")) + _ = framework.ReadAllAndClose(t, upResp) + require.Equal(t, http.StatusCreated, upResp.StatusCode) + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + _, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{ + VolumeId: volumeID, Collection: collection, + }) + require.NoError(t, err) + + // A half-finished previous distribute left a partial shard set mounted. + staleShards := []uint32{0, 1, 2} + _, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{ + VolumeId: volumeID, Collection: collection, + ShardIds: staleShards, + }) + require.NoError(t, err) + + shardPath := makeTinyEcShardFile(t) + + // Precondition for the reproduction: the mounted partial EC blocks a fresh + // ReceiveFile via the mounted-volume guard. + err = sendShardViaReceiveFile(ctx, grpcClient, volumeID, collection, 0, shardPath) + require.Error(t, err, "expected ReceiveFile to be refused while stale EC volume is mounted") + + task := NewErasureCodingTask( + "preflight-invariant", + clusterHarness.VolumeServerAddress(), + volumeID, + collection, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + // Named only as a target — no source row for this node. The preflight must + // still clear its stale shards. + task.targets = []*worker_pb.TaskTarget{ + {Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID, ShardIds: []uint32{0}}, + } + + require.NoError(t, task.ensureCleanEcStart(ctx), + "Step 0 preflight must clear stale EC shards at task start") + + _, infoErr := grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{VolumeId: volumeID}) + require.Error(t, infoErr, "stale EC volume must be gone after the preflight") + + require.NoError(t, + sendShardViaReceiveFile(ctx, grpcClient, volumeID, collection, 0, shardPath), + "ReceiveFile must succeed once the preflight has cleared the stale shards") +} + +// A post-distribute failure that has no successor to clean up after it (a +// single-attempt job, or the last of a retry series) must leave nothing behind: +// rollbackDistribute tears down the shards this attempt wrote while leaving the +// source volume intact for a future re-encode. Reproduces the swtest finding +// that a terminally-failed encode stranded orphan shards. +func TestRollbackDistributeClearsShardsAndKeepsSource(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const ( + volumeID = uint32(94783) + collection = "ec-rollback-distribute" + ) + + framework.AllocateVolume(t, grpcClient, volumeID, collection) + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 9478300, 0x9478D00D) + upResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, + []byte("payload-for-rollback-distribute")) + _ = framework.ReadAllAndClose(t, upResp) + require.Equal(t, http.StatusCreated, upResp.StatusCode) + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + _, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{ + VolumeId: volumeID, Collection: collection, + }) + require.NoError(t, err) + // Shards this attempt "distributed" and mounted on the destination. + _, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{ + VolumeId: volumeID, Collection: collection, + ShardIds: []uint32{0, 1, 2, 3}, + }) + require.NoError(t, err) + + task := NewErasureCodingTask( + "rollback-distribute", + clusterHarness.VolumeServerAddress(), + volumeID, + collection, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + task.targets = []*worker_pb.TaskTarget{ + {Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID, ShardIds: []uint32{0}}, + } + + // Simulate the failure path after distribute began. + task.rollbackDistribute(ctx) + + _, infoErr := grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{VolumeId: volumeID}) + require.Error(t, infoErr, "rollback must tear down the shards this attempt distributed") + + // The source normal volume must survive so a future re-encode can proceed. + _, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID}) + require.NoError(t, statusErr, "rollback must leave the source volume intact") +} + +// A malformed plan (no targets) must be rejected by the preflight before the +// source is marked readonly or copied, so a bad plan cannot leave the source +// fenced readonly with nothing to show for it. Pure precondition check: no +// cluster and no RPC. +func TestEnsureCleanEcStartRejectsMissingTargets(t *testing.T) { + task := NewErasureCodingTask( + "preflight-no-targets", + "10.0.0.1:8080", + 42, + "c", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + // No targets set. + err := task.ensureCleanEcStart(context.Background()) + require.Error(t, err, "preflight must reject an encode with no shard targets") + require.True(t, strings.Contains(err.Error(), "no EC shard targets"), + "error must name the missing-targets invariant, got: %v", err) +} + +// A non-empty target slice whose entries are malformed (nil, empty Node, or no +// assigned shards) must also be rejected before the source is marked readonly — +// cleanupStaleEcShards silently skips such entries, so a length check alone +// would let the encode proceed with nothing to distribute to. +func TestEnsureCleanEcStartRejectsMalformedTarget(t *testing.T) { + cases := map[string][]*worker_pb.TaskTarget{ + "nil target": {nil}, + "empty node": {{Node: "", VolumeId: 42, ShardIds: []uint32{0}}}, + "no shards": {{Node: "10.0.0.2:8080", VolumeId: 42}}, + "one good one bad": { + {Node: "10.0.0.2:8080", VolumeId: 42, ShardIds: []uint32{0}}, + {Node: "", VolumeId: 42, ShardIds: []uint32{1}}, + }, + } + for name, targets := range cases { + t.Run(name, func(t *testing.T) { + task := NewErasureCodingTask( + "preflight-malformed-target", + "10.0.0.1:8080", + 42, + "c", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + task.targets = targets + err := task.ensureCleanEcStart(context.Background()) + require.Error(t, err, "preflight must reject a malformed target") + require.True(t, strings.Contains(err.Error(), "malformed EC shard target"), + "error must name the malformed-target invariant, got: %v", err) + }) + } +} + +// A plan with targets but no source replica and no assigned server must be +// rejected before any destructive step. +func TestEnsureCleanEcStartRejectsMissingSource(t *testing.T) { + task := NewErasureCodingTask( + "preflight-no-source", + "", // no assigned server + 42, + "c", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + task.targets = []*worker_pb.TaskTarget{ + {Node: "10.0.0.2:8080", VolumeId: 42, ShardIds: []uint32{0}}, + } + // No sources and no server. + err := task.ensureCleanEcStart(context.Background()) + require.Error(t, err, "preflight must reject an encode with no source replica") + require.True(t, strings.Contains(err.Error(), "no source replica"), + "error must name the missing-source invariant, got: %v", err) +}