ec: report expiry from the Rust volume server and cover the .vif carry

The seaweed-volume drop-in server omitted expire_at_sec from its
ReadVolumeFileStatus response, so worker-encoded volumes sourced from it
lost the TTL. Fall back to the nested VolumeInfo expiry for servers that
predate the top-level field, and pin the nonzero/zero carry with a test.

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 {

seaweed-volume/src/server/grpc_server.rs

@@ -2005,6 +2005,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(),
             },
         ))
     }

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 {

weed/worker/tasks/erasure_coding/ec_task.go

@@ -517,6 +517,11 @@ func (t *ErasureCodingTask) copyVolumeFilesToWorker(ctx context.Context, workDir
 		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,

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()