diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 5ae5511e4..895a7274c 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -571,6 +571,7 @@ message ReadVolumeFileStatusResponse { string disk_type = 9; VolumeInfo volume_info = 10; uint32 version = 11; + uint64 expire_at_sec = 12; // TTL expiry counted from the last write; 0 when the volume never expires } message DiskStatus { diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 6ca502385..e6586af1f 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -2032,6 +2032,7 @@ impl VolumeServer for VolumeGrpcService { disk_type: store.locations[loc_idx].disk_type.to_string(), volume_info: Some(vol.volume_info().clone()), version: vol.version().0 as u32, + expire_at_sec: vol.expire_at_sec(), }, )) } diff --git a/test/volume_server/framework/volume_fixture.go b/test/volume_server/framework/volume_fixture.go index 1ba40434a..7cf044443 100644 --- a/test/volume_server/framework/volume_fixture.go +++ b/test/volume_server/framework/volume_fixture.go @@ -16,6 +16,18 @@ import ( func AllocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string) { t.Helper() + allocateVolume(t, client, volumeID, collection, "") +} + +// AllocateVolumeWithTTL allocates a volume whose needles expire after ttl +// (e.g. "10m"). +func AllocateVolumeWithTTL(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string, ttl string) { + t.Helper() + allocateVolume(t, client, volumeID, collection, ttl) +} + +func allocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, volumeID uint32, collection string, ttl string) { + t.Helper() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -24,6 +36,7 @@ func AllocateVolume(t testing.TB, client volume_server_pb.VolumeServerClient, vo VolumeId: volumeID, Collection: collection, Replication: "000", + Ttl: ttl, Version: uint32(needle.GetCurrentVersion()), }) if err != nil { diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 7a20d5632..3a66ffff7 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -573,6 +573,7 @@ message ReadVolumeFileStatusResponse { string disk_type = 9; VolumeInfo volume_info = 10; uint32 version = 11; + uint64 expire_at_sec = 12; // TTL expiry counted from the last write; 0 when the volume never expires } message DiskStatus { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 653bea272..ce6c0c6c7 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -4678,6 +4678,7 @@ type ReadVolumeFileStatusResponse struct { DiskType string `protobuf:"bytes,9,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"` VolumeInfo *VolumeInfo `protobuf:"bytes,10,opt,name=volume_info,json=volumeInfo,proto3" json:"volume_info,omitempty"` Version uint32 `protobuf:"varint,11,opt,name=version,proto3" json:"version,omitempty"` + ExpireAtSec uint64 `protobuf:"varint,12,opt,name=expire_at_sec,json=expireAtSec,proto3" json:"expire_at_sec,omitempty"` // TTL expiry counted from the last write; 0 when the volume never expires unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -4789,6 +4790,13 @@ func (x *ReadVolumeFileStatusResponse) GetVersion() uint32 { return 0 } +func (x *ReadVolumeFileStatusResponse) GetExpireAtSec() uint64 { + if x != nil { + return x.ExpireAtSec + } + return 0 +} + type DiskStatus struct { state protoimpl.MessageState `protogen:"open.v1"` Dir string `protobuf:"bytes,1,opt,name=dir,proto3" json:"dir,omitempty"` @@ -7601,7 +7609,7 @@ const file_volume_server_proto_rawDesc = "" + "collection\x12\x1b\n" + "\tvolume_id\x18\x04 \x01(\rR\bvolumeId\":\n" + "\x1bReadVolumeFileStatusRequest\x12\x1b\n" + - "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\xe3\x03\n" + + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\"\x87\x04\n" + "\x1cReadVolumeFileStatusResponse\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12;\n" + "\x1aidx_file_timestamp_seconds\x18\x02 \x01(\x04R\x17idxFileTimestampSeconds\x12\"\n" + @@ -7618,7 +7626,8 @@ const file_volume_server_proto_rawDesc = "" + "\vvolume_info\x18\n" + " \x01(\v2\x1c.volume_server_pb.VolumeInfoR\n" + "volumeInfo\x12\x18\n" + - "\aversion\x18\v \x01(\rR\aversion\"\xd1\x01\n" + + "\aversion\x18\v \x01(\rR\aversion\x12\"\n" + + "\rexpire_at_sec\x18\f \x01(\x04R\vexpireAtSec\"\xd1\x01\n" + "\n" + "DiskStatus\x12\x10\n" + "\x03dir\x18\x01 \x01(\tR\x03dir\x12\x10\n" + diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 76c5619ce..110759f19 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -536,6 +536,7 @@ func (vs *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_se resp.DiskType = string(v.DiskType()) resp.VolumeInfo = v.GetVolumeInfo() resp.Version = uint32(v.Version()) + resp.ExpireAtSec = v.ExpireAtSec() return resp, nil } diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index e86cbf05c..043ff8677 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -56,6 +56,10 @@ type ErasureCodingTask struct { // re-delete and remove the now-EC .vif those servers share. emptyReplicasDeleted map[string]bool + // expireAtSec is the source volume's TTL expiry read with the file copy; + // it goes into the .vif so the EC volume honors the collection's TTL. + expireAtSec uint64 + // encodedBlockSize is the shard block layout WriteEcFiles actually encoded // with, read back off the EC context. Every holder must report serving the // same one before the source volume may be deleted. @@ -512,6 +516,12 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir if err != nil { return nil, fmt.Errorf("failed to read source volume file status: %w", err) } + t.expireAtSec = fileStatus.GetExpireAtSec() + if t.expireAtSec == 0 { + // Volume servers that predate the top-level field still carry the + // expiry in the nested .vif once SaveVolumeInfo has stamped it. + t.expireAtSec = fileStatus.GetVolumeInfo().GetExpireAtSec() + } t.GetLogger().WithFields(map[string]interface{}{ "volume_id": t.volumeID, @@ -749,6 +759,7 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string volumeInfo := &volume_server_pb.VolumeInfo{ Version: uint32(needle.GetCurrentVersion()), EcShardConfig: ecShardConfig, + ExpireAtSec: t.expireAtSec, } // The decoder resolves the shard block layout from the encode-time .dat // size; without it, decoding falls back to inferring the layout from the diff --git a/weed/worker/tasks/erasure_coding/ec_task_test.go b/weed/worker/tasks/erasure_coding/ec_task_test.go index 56dbc9d20..dc4548e95 100644 --- a/weed/worker/tasks/erasure_coding/ec_task_test.go +++ b/weed/worker/tasks/erasure_coding/ec_task_test.go @@ -168,6 +168,7 @@ func TestGenerateEcShardsLocallyStampsEncodeIdentity(t *testing.T) { require.Greater(t, vi.EcShardConfig.DataShards, uint32(0)) require.Greater(t, vi.EcShardConfig.ParityShards, uint32(0)) require.NotZero(t, vi.EcShardConfig.EncodeTsNs, "worker-encoded .vif must carry an encode identity") + require.Zero(t, vi.ExpireAtSec, "a source with no TTL must not gain an expiry") } // A fenced run must stamp the admin-issued generation (t.encodeTsNs) verbatim into @@ -218,6 +219,56 @@ func TestGenerateEcShardsLocallyUsesAdminGeneration(t *testing.T) { require.Equal(t, adminGeneration, vi.EcShardConfig.EncodeTsNs, "fenced run must stamp the admin generation verbatim, not a local timestamp") } +// The worker-written .vif must carry the source volume's TTL expiry, or the EC +// volume never becomes eligible for deletion while the shell ec.encode path does. +func TestGenerateEcShardsLocallyCarriesExpiry(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(958) + framework.AllocateVolumeWithTTL(t, grpcClient, volumeID, "", "10m") + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 4001, 0x6666FFFF) + resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("payload-for-ec-expiry")) + _ = framework.ReadAllAndClose(t, resp) + require.Equal(t, http.StatusCreated, resp.StatusCode) + + task := NewErasureCodingTask( + "ec-carry-expiry", + clusterHarness.VolumeServerAddress(), + volumeID, + "", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + require.NoError(t, task.markReplicasReadonly(ctx)) + + fileStatus, err := task.readSourceVolumeFileStatus(ctx) + require.NoError(t, err) + want := fileStatus.GetExpireAtSec() + require.NotZero(t, want, "a TTL'd source must report a nonzero expiry") + + localFiles, err := task.copyVolumeFilesToWorker(ctx, t.TempDir()) + require.NoError(t, err) + + shardFiles, err := task.generateEcShardsLocally(localFiles, t.TempDir()) + require.NoError(t, err) + + vi, _, found, err := volume_info.MaybeLoadVolumeInfo(shardFiles["vif"]) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, want, vi.ExpireAtSec, "worker .vif must carry the source volume's TTL expiry") +} + func compactVolumeOnce(t *testing.T, grpcClient volume_server_pb.VolumeServerClient, volumeID uint32) { t.Helper()