iceberg maintenance: keep the snapshots that branches and tags pin (#10774)

* iceberg maintenance: keep the snapshots that branches and tags pin

expireSnapshots only ever protected the current snapshot, so a snapshot
held by a tag or a non-main branch was expired once it aged out of the
retention window. iceberg-go's RemoveSnapshots drops any ref whose
snapshot is gone without complaint, so the tag disappeared and the files
behind it were deleted as unreferenced.

Protect every ref target, and honour a branch's own
min-snapshots-to-keep / max-snapshot-age-ms over the ancestors behind its
head. Detection skips pinned snapshots for the same reason: proposing a
job whose only outcome is a no-op keeps the worker busy forever.

* iceberg maintenance: re-plan when a ref appears mid-commit, and stop proposing no-op expiry

Three follow-ups from review of the ref-aware expiry:

The commit guard only compared the table head, so a tag created between
planning and commit could pin a snapshot the plan was about to expire.
Re-check the refs against the metadata the commit actually reads.

Detection now asks snapshotsToExpire what execution would remove instead
of approximating with its own count-and-age rules. Expiry always requires
a snapshot past the retention window, so a table over the quota whose
snapshots are all young was being proposed for a job that could only
no-op.

The branch retention test could not tell "retained the whole lineage"
from "honoured min-snapshots-to-keep", because the branch had exactly as
many ancestors as the count. Give it one more, and cover
max-snapshot-age-ms too. Both need snapshots genuinely older than a
retention window, which iceberg-go will not accept at build time, so the
fixture backdates the metadata after building it.

* iceberg maintenance: fold the metadata test builders back into one

buildTestMetadata, buildTestMetadataWithRefs, buildTestMetadataAged and
buildTestMetadataNow were four names for one thing. Keep the original and
give it the refs and age it needs.

weed/worker/tasks/iceberg/detection.go

@@ -486,27 +486,12 @@ func compactionMinInputFiles(minInputFiles int64) (int, error) {
 
 // needsMaintenance checks whether snapshot expiration work is needed based on
 // metadata-only thresholds.
+// It asks execution what it would do rather than reimplementing the rules:
+// expiry always requires a snapshot to be past the retention window, so a
+// table over the snapshot quota whose snapshots are all young, or all pinned by
+// refs, would otherwise be proposed for a job that can only no-op.
 func needsMaintenance(meta table.Metadata, config Config) bool {
-	snapshots := meta.Snapshots()
-	if len(snapshots) == 0 {
-		return false
-	}
-
-	// Check snapshot count
-	if int64(len(snapshots)) > config.MaxSnapshotsToKeep {
-		return true
-	}
-
-	// Check oldest snapshot age
-	retentionMs := config.SnapshotRetentionMs
-	nowMs := time.Now().UnixMilli()
-	for _, snap := range snapshots {
-		if nowMs-snap.TimestampMs > retentionMs {
-			return true
-		}
-	}
-
-	return false
+	return len(snapshotsToExpire(meta, config, time.Now().UnixMilli())) > 0
 }
 
 // buildMaintenanceProposal creates a JobProposal for a table needing maintenance.

weed/worker/tasks/iceberg/exec_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -40,6 +41,7 @@ type fakeFilerServer struct {
 	mu           sync.Mutex
 	entries      map[string]map[string]*filer_pb.Entry // dir → name → entry
 	beforeUpdate func(*fakeFilerServer, *filer_pb.UpdateEntryRequest) error
+	beforeLookup func(*fakeFilerServer, *filer_pb.LookupDirectoryEntryRequest)
 
 	// Set by enableAssign to serve AssignVolume against a fake volume server.
 	assignVolumeServer string
@@ -93,6 +95,13 @@ func (f *fakeFilerServer) listDir(dir string) []*filer_pb.Entry {
 }
 
 func (f *fakeFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	f.mu.Lock()
+	beforeLookup := f.beforeLookup
+	f.mu.Unlock()
+	if beforeLookup != nil {
+		beforeLookup(f, req)
+	}
+
 	entry := f.getEntry(req.Directory, req.Name)
 	if entry == nil {
 		return nil, status.Errorf(codes.NotFound, "entry not found: %s/%s", req.Directory, req.Name)
@@ -287,6 +296,12 @@ type tableSetup struct {
 	// that created it would record the table's running totals.
 	SnapshotSummary map[string]string
 	Snapshots       []table.Snapshot
+	// Refs are branches and tags beyond main, which always points at the last
+	// snapshot.
+	Refs map[string]table.SnapshotRef
+	// Age backdates the whole table so its snapshots can sit outside a
+	// retention window.
+	Age time.Duration
 }
 
 func (ts tableSetup) tablePath() string {
@@ -314,7 +329,7 @@ func (ts tableSetup) fileRef(elem ...string) string {
 func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Metadata {
 	t.Helper()
 
-	meta := buildTestMetadata(t, setup.Snapshots)
+	meta := buildTestMetadata(t, setup.Snapshots, setup.Refs, setup.Age)
 	fullMetadataJSON, err := json.Marshal(meta)
 	if err != nil {
 		t.Fatalf("marshal metadata: %v", err)
@@ -582,6 +597,244 @@ func TestExpireSnapshotsExecution(t *testing.T) {
 	}
 }
 
+func TestExpireSnapshotsKeepsTaggedSnapshot(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().Add(-10 * time.Second).UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+		},
+		Refs: map[string]table.SnapshotRef{
+			"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
+		},
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
+	}
+
+	if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("reload metadata: %v", err)
+	}
+
+	remaining := map[int64]bool{}
+	for _, snap := range state.Metadata.Snapshots() {
+		remaining[snap.SnapshotID] = true
+	}
+	if !remaining[1] {
+		t.Error("tagged snapshot 1 was expired")
+	}
+	if remaining[2] {
+		t.Error("expected untagged snapshot 2 to be expired")
+	}
+
+	var tagged bool
+	for name, ref := range state.Metadata.Refs() {
+		if name == "release" {
+			tagged = true
+			if ref.SnapshotID != 1 {
+				t.Errorf("tag release points at snapshot %d, want 1", ref.SnapshotID)
+			}
+		}
+	}
+	if !tagged {
+		t.Error("tag release was dropped from the metadata")
+	}
+
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
+	if fs.getEntry(metaDir, "snap-1.avro") == nil {
+		t.Error("manifest list of the tagged snapshot was deleted")
+	}
+	if fs.getEntry(metaDir, "snap-2.avro") != nil {
+		t.Error("expected the expired snapshot's manifest list to be deleted")
+	}
+}
+
+func TestExpireSnapshotsHonorsBranchRetention(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	// The branch chain is 1 <- 2 <- 3, with main on 4. Backdating the table by
+	// Age gives every snapshot a real age; the ten-second spacing keeps them
+	// distinct and stays inside what iceberg-go accepts at build time.
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	stepMs := int64(10 * time.Second / time.Millisecond)
+	first, second := int64(1), int64(2)
+	minKeep := 2
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
+			{SnapshotID: 2, TimestampMs: now + stepMs, ManifestList: "metadata/snap-2.avro", ParentSnapshotID: &first, SequenceNumber: 2},
+			{SnapshotID: 3, TimestampMs: now + 2*stepMs, ManifestList: "metadata/snap-3.avro", ParentSnapshotID: &second, SequenceNumber: 3},
+			{SnapshotID: 4, TimestampMs: now + 3*stepMs, ManifestList: "metadata/snap-4.avro", SequenceNumber: 4},
+		},
+		Refs: map[string]table.SnapshotRef{
+			"audit": {SnapshotID: 3, SnapshotRefType: table.BranchRef, MinSnapshotsToKeep: &minKeep},
+		},
+		Age: 5 * time.Hour,
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
+	}
+
+	if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+
+	// min-snapshots-to-keep=2 reaches from the branch head back to its parent,
+	// and stops there: the grandparent is past the count and expires.
+	assertSnapshots(t, client, setup, []int64{2, 3, 4}, []int64{1})
+}
+
+func TestExpireSnapshotsHonorsBranchMaxSnapshotAge(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	stepMs := int64(10 * time.Second / time.Millisecond)
+	first, second := int64(1), int64(2)
+	// Between the branch head's parent (5h20s) and its grandparent (5h30s), so
+	// the window has to reach past the head to be observable.
+	maxAgeMs := int64(5*time.Hour/time.Millisecond) + 25*1000
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
+			{SnapshotID: 2, TimestampMs: now + stepMs, ManifestList: "metadata/snap-2.avro", ParentSnapshotID: &first, SequenceNumber: 2},
+			{SnapshotID: 3, TimestampMs: now + 2*stepMs, ManifestList: "metadata/snap-3.avro", ParentSnapshotID: &second, SequenceNumber: 3},
+			{SnapshotID: 4, TimestampMs: now + 3*stepMs, ManifestList: "metadata/snap-4.avro", SequenceNumber: 4},
+		},
+		Refs: map[string]table.SnapshotRef{
+			"audit": {SnapshotID: 3, SnapshotRefType: table.BranchRef, MaxSnapshotAgeMs: &maxAgeMs},
+		},
+		Age: 5 * time.Hour,
+	}
+	populateTable(t, fs, setup)
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    3,
+		Operations:          "expire_snapshots",
+	}
+
+	if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+
+	// The window reaches from the branch head back over its parent but stops
+	// short of the grandparent.
+	assertSnapshots(t, client, setup, []int64{2, 3, 4}, []int64{1})
+}
+
+// assertSnapshots reloads the table and checks exactly which snapshots survived.
+func assertSnapshots(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, want, gone []int64) {
+	t.Helper()
+
+	state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("reload metadata: %v", err)
+	}
+	remaining := map[int64]bool{}
+	for _, snap := range state.Metadata.Snapshots() {
+		remaining[snap.SnapshotID] = true
+	}
+	for _, id := range want {
+		if !remaining[id] {
+			t.Errorf("snapshot %d was expired, want kept", id)
+		}
+	}
+	for _, id := range gone {
+		if remaining[id] {
+			t.Errorf("snapshot %d survived, want expired", id)
+		}
+	}
+}
+
+// A tag created between planning and commit pins a snapshot the plan was going
+// to expire. The head has not moved, so only a refs check can catch it.
+func TestExpireSnapshotsRejectsPlanWhenARefAppears(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+		},
+		Age: 200 * time.Hour,
+	}
+	populateTable(t, fs, setup)
+
+	// Re-publish the table with a tag on snapshot 1 while the commit is
+	// re-reading it, which is the window the guard closes.
+	tagged := setup
+	tagged.Refs = map[string]table.SnapshotRef{
+		"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
+	}
+	lookups := 0
+	fs.beforeLookup = func(f *fakeFilerServer, req *filer_pb.LookupDirectoryEntryRequest) {
+		if req.Name != setup.TableName {
+			return
+		}
+		lookups++
+		if lookups == 2 {
+			populateTable(t, f, tagged)
+		}
+	}
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+		MaxCommitRetries:    1,
+		Operations:          "expire_snapshots",
+	}
+
+	_, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err == nil {
+		t.Fatal("expireSnapshots() error = nil, want the plan rejected as stale")
+	}
+	if !errors.Is(err, errStalePlan) {
+		t.Fatalf("expireSnapshots() error = %v, want errStalePlan", err)
+	}
+
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
+	if fs.getEntry(metaDir, "snap-1.avro") == nil {
+		t.Error("the newly tagged snapshot's manifest list was deleted")
+	}
+}
+
 func TestExpireSnapshotsNothingToExpire(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -1239,6 +1492,7 @@ func TestDetectWithFakeFiler(t *testing.T) {
 			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
 			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
 		},
+		Age: 200 * time.Hour, // past the default 7-day retention
 	}
 	populateTable(t, fs, setup)
 
@@ -1286,6 +1540,7 @@ func TestDetectWithFilters(t *testing.T) {
 			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
 			{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
 		},
+		Age: 200 * time.Hour,
 	}
 	setup2 := tableSetup{
 		BucketName: "bucket-b",
@@ -1296,6 +1551,7 @@ func TestDetectWithFilters(t *testing.T) {
 			{SnapshotID: 5, TimestampMs: now + 4, ManifestList: "metadata/snap-5.avro"},
 			{SnapshotID: 6, TimestampMs: now + 5, ManifestList: "metadata/snap-6.avro"},
 		},
+		Age: 200 * time.Hour,
 	}
 	populateTable(t, fs, setup1)
 	populateTable(t, fs, setup2)
@@ -1523,6 +1779,7 @@ func TestDetectSchedulesSnapshotExpiryDespiteCompactionEvaluationError(t *testin
 			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", SequenceNumber: 1},
 			{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro", SequenceNumber: 2},
 		},
+		Age: 400 * 24 * time.Hour, // past the year-long retention below
 	}
 	populateTable(t, fs, setup)
 

weed/worker/tasks/iceberg/handler_test.go

@@ -3,9 +3,11 @@ package iceberg
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"path"
+	"strconv"
 	"testing"
 	"time"
 
@@ -119,13 +121,33 @@ func TestNeedsMaintenanceNoSnapshots(t *testing.T) {
 		MaxSnapshotsToKeep:  2,
 	}
 
-	meta := buildTestMetadata(t, nil)
+	meta := buildTestMetadata(t, nil, nil, 0)
 	if needsMaintenance(meta, config) {
 		t.Error("expected no maintenance for table with no snapshots")
 	}
 }
 
 func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(24),
+		MaxSnapshotsToKeep:  2,
+	}
+
+	now := time.Now().UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+		{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
+	}
+	meta := buildTestMetadata(t, snapshots, nil, 48*time.Hour)
+	if !needsMaintenance(meta, config) {
+		t.Error("expected maintenance for table exceeding max snapshots")
+	}
+}
+
+// Expiry always requires a snapshot to be past the retention window, so a table
+// over the quota whose snapshots are all young has nothing to do.
+func TestNeedsMaintenanceExceedsMaxSnapshotsWithinRetention(t *testing.T) {
 	config := Config{
 		SnapshotRetentionMs: hoursToMs(24 * 365), // very long retention
 		MaxSnapshotsToKeep:  2,
@@ -137,9 +159,31 @@ func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
 		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
 		{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
 	}
-	meta := buildTestMetadata(t, snapshots)
-	if !needsMaintenance(meta, config) {
-		t.Error("expected maintenance for table exceeding max snapshots")
+	if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0), config) {
+		t.Error("expected no maintenance while every snapshot is inside the retention window")
+	}
+}
+
+// Every expirable snapshot is pinned by a tag, so a job could only no-op.
+func TestNeedsMaintenanceSkipsRefPinnedSnapshots(t *testing.T) {
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
+		MaxSnapshotsToKeep:  1,
+	}
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+	}
+	refs := map[string]table.SnapshotRef{
+		"release": {SnapshotID: 1, SnapshotRefType: table.TagRef},
+	}
+	if needsMaintenance(buildTestMetadata(t, snapshots, refs, 0), config) {
+		t.Error("expected no maintenance when the only old snapshot is tagged")
+	}
+	if !needsMaintenance(buildTestMetadata(t, snapshots, nil, 0), config) {
+		t.Error("expected maintenance for the same table without the tag")
 	}
 }
 
@@ -153,7 +197,7 @@ func TestNeedsMaintenanceWithinLimits(t *testing.T) {
 	snapshots := []table.Snapshot{
 		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
 	}
-	meta := buildTestMetadata(t, snapshots)
+	meta := buildTestMetadata(t, snapshots, nil, 0)
 	if needsMaintenance(meta, config) {
 		t.Error("expected no maintenance for table within limits")
 	}
@@ -163,17 +207,34 @@ func TestNeedsMaintenanceOldSnapshot(t *testing.T) {
 	// Use a retention of 0 hours so that any snapshot is considered "old"
 	config := Config{
 		SnapshotRetentionMs: hoursToMs(0), // instant expiry
+		MaxSnapshotsToKeep:  1,
+	}
+
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
+	snapshots := []table.Snapshot{
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
+	}
+	meta := buildTestMetadata(t, snapshots, nil, 0)
+	if !needsMaintenance(meta, config) {
+		t.Error("expected maintenance for table with expired snapshot")
+	}
+}
+
+// The current snapshot is never expirable, so a single-snapshot table has no
+// work no matter how far past retention it is.
+func TestNeedsMaintenanceSingleSnapshot(t *testing.T) {
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(0),
 		MaxSnapshotsToKeep:  10,
 	}
 
-	now := time.Now().UnixMilli()
+	now := time.Now().Add(-30 * time.Second).UnixMilli()
 	snapshots := []table.Snapshot{
-		{SnapshotID: 1, TimestampMs: now - 1, ManifestList: "metadata/snap-1.avro"},
+		{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
 	}
-	meta := buildTestMetadata(t, snapshots)
-	// With 0 retention, any snapshot with timestamp < now should need maintenance
-	if !needsMaintenance(meta, config) {
-		t.Error("expected maintenance for table with expired snapshot")
+	if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0), config) {
+		t.Error("expected no maintenance for a table with only the current snapshot")
 	}
 }
 
@@ -217,7 +278,7 @@ func TestBuildMaintenanceProposal(t *testing.T) {
 		{SnapshotID: 1, TimestampMs: now},
 		{SnapshotID: 2, TimestampMs: now + 1},
 	}
-	meta := buildTestMetadata(t, snapshots)
+	meta := buildTestMetadata(t, snapshots, nil, 0)
 
 	info := tableInfo{
 		BucketName: "my-bucket",
@@ -1408,9 +1469,14 @@ func TestExecuteNilRequest(t *testing.T) {
 // Test helpers
 // ---------------------------------------------------------------------------
 
-// buildTestMetadata creates a minimal Iceberg metadata for testing.
-// When snapshots is nil or empty, the metadata has no snapshots.
-func buildTestMetadata(t *testing.T, snapshots []table.Snapshot) table.Metadata {
+// buildTestMetadata creates a minimal Iceberg metadata for testing. When
+// snapshots is nil or empty the metadata has no snapshots; refs adds branches
+// and tags on top of the main branch, which always points at the last snapshot.
+// A positive age backdates the result, letting a test describe snapshots that
+// are genuinely past a retention window - iceberg-go refuses to add a snapshot
+// stamped more than a minute before the metadata's last-updated time, so the
+// shift has to happen after the build.
+func buildTestMetadata(t *testing.T, snapshots []table.Snapshot, refs map[string]table.SnapshotRef, age time.Duration) table.Metadata {
 	t.Helper()
 
 	schema := newTestSchema()
@@ -1441,11 +1507,80 @@ func buildTestMetadata(t *testing.T, snapshots []table.Snapshot) table.Metadata
 		t.Fatalf("failed to set snapshot ref: %v", err)
 	}
 
+	// The option type is unexported, so each combination is spelled out.
+	for name, ref := range refs {
+		var refErr error
+		switch {
+		case ref.MinSnapshotsToKeep != nil && ref.MaxSnapshotAgeMs != nil:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType,
+				table.WithMinSnapshotsToKeep(*ref.MinSnapshotsToKeep), table.WithMaxSnapshotAgeMs(*ref.MaxSnapshotAgeMs))
+		case ref.MinSnapshotsToKeep != nil:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType,
+				table.WithMinSnapshotsToKeep(*ref.MinSnapshotsToKeep))
+		case ref.MaxSnapshotAgeMs != nil:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType,
+				table.WithMaxSnapshotAgeMs(*ref.MaxSnapshotAgeMs))
+		default:
+			refErr = builder.SetSnapshotRef(name, ref.SnapshotID, ref.SnapshotRefType)
+		}
+		if refErr != nil {
+			t.Fatalf("failed to set ref %s: %v", name, refErr)
+		}
+	}
+
 	result, err := builder.Build()
 	if err != nil {
 		t.Fatalf("failed to build metadata: %v", err)
 	}
-	return result
+	metadata := result
+	if age <= 0 {
+		return metadata
+	}
+
+	encoded, err := json.Marshal(metadata)
+	if err != nil {
+		t.Fatalf("marshal metadata: %v", err)
+	}
+
+	decoder := json.NewDecoder(bytes.NewReader(encoded))
+	decoder.UseNumber()
+	var doc map[string]any
+	if err := decoder.Decode(&doc); err != nil {
+		t.Fatalf("decode metadata: %v", err)
+	}
+
+	shiftMs := age.Milliseconds()
+	shift := func(container map[string]any, key string) {
+		raw, ok := container[key].(json.Number)
+		if !ok {
+			return
+		}
+		value, err := raw.Int64()
+		if err != nil {
+			t.Fatalf("parse %s: %v", key, err)
+		}
+		container[key] = json.Number(strconv.FormatInt(value-shiftMs, 10))
+	}
+
+	shift(doc, "last-updated-ms")
+	for _, listKey := range []string{"snapshots", "snapshot-log", "metadata-log"} {
+		entries, _ := doc[listKey].([]any)
+		for _, entry := range entries {
+			if item, ok := entry.(map[string]any); ok {
+				shift(item, "timestamp-ms")
+			}
+		}
+	}
+
+	aged, err := json.Marshal(doc)
+	if err != nil {
+		t.Fatalf("marshal aged metadata: %v", err)
+	}
+	parsed, err := table.ParseMetadataBytes(aged)
+	if err != nil {
+		t.Fatalf("parse aged metadata: %v", err)
+	}
+	return parsed
 }
 
 func newTestSchema() *iceberg.Schema {

weed/worker/tasks/iceberg/operations.go

@@ -51,47 +51,13 @@ func (h *Handler) expireSnapshots(
 		return "no snapshots", nil, nil
 	}
 
-	// Determine which snapshots to expire
 	currentSnap := meta.CurrentSnapshot()
 	var currentSnapID int64
 	if currentSnap != nil {
 		currentSnapID = currentSnap.SnapshotID
 	}
 
-	retentionMs := config.SnapshotRetentionMs
-	nowMs := time.Now().UnixMilli()
-
-	// Sort snapshots by timestamp descending (most recent first) so that
-	// the keep-count logic always preserves the newest snapshots.
-	sorted := make([]table.Snapshot, len(snapshots))
-	copy(sorted, snapshots)
-	sort.Slice(sorted, func(i, j int) bool {
-		return sorted[i].TimestampMs > sorted[j].TimestampMs
-	})
-
-	// Walk from newest to oldest. The current snapshot is always kept.
-	// Among the remaining, keep up to MaxSnapshotsToKeep-1 (since current
-	// counts toward the quota). Expire the rest only if they exceed the
-	// retention window; snapshots within the window are kept regardless.
-	var toExpire []int64
-	var kept int64
-	for _, snap := range sorted {
-		if snap.SnapshotID == currentSnapID {
-			kept++
-			continue
-		}
-		age := nowMs - snap.TimestampMs
-		if kept < config.MaxSnapshotsToKeep {
-			kept++
-			continue
-		}
-		if age > retentionMs {
-			toExpire = append(toExpire, snap.SnapshotID)
-		} else {
-			kept++
-		}
-	}
-
+	toExpire := snapshotsToExpire(meta, config, time.Now().UnixMilli())
 	if len(toExpire) == 0 {
 		return "no snapshots expired", nil, nil
 	}
@@ -102,7 +68,7 @@ func (h *Handler) expireSnapshots(
 		expireSet[id] = struct{}{}
 	}
 	var expiredSnaps, keptSnaps []table.Snapshot
-	for _, snap := range sorted {
+	for _, snap := range snapshots {
 		if _, ok := expireSet[snap.SnapshotID]; ok {
 			expiredSnaps = append(expiredSnaps, snap)
 		} else {
@@ -136,6 +102,15 @@ func (h *Handler) expireSnapshots(
 		if (cs == nil) != (currentSnapID == 0) || (cs != nil && cs.SnapshotID != currentSnapID) {
 			return errStalePlan
 		}
+		// A tag or branch created since planning can pin a snapshot this plan
+		// expires without moving the head, and RemoveSnapshots would drop that
+		// ref along with it. Re-plan instead.
+		nowProtected := protectedSnapshots(currentMeta, time.Now().UnixMilli())
+		for _, id := range toExpire {
+			if _, pinned := nowProtected[id]; pinned {
+				return errStalePlan
+			}
+		}
 		return builder.RemoveSnapshots(toExpire, false)
 	})
 	if err != nil {
@@ -168,6 +143,106 @@ func (h *Handler) expireSnapshots(
 	return fmt.Sprintf("expired %d snapshot(s), deleted %d unreferenced file(s)", len(toExpire), deletedCount), metrics, nil
 }
 
+// snapshotsToExpire returns the snapshot IDs that fall outside the configured
+// retention. Detection and execution both call it so a table is only proposed
+// for expiry when the run would actually remove something.
+func snapshotsToExpire(meta table.Metadata, config Config, nowMs int64) []int64 {
+	var currentSnapID int64
+	if currentSnap := meta.CurrentSnapshot(); currentSnap != nil {
+		currentSnapID = currentSnap.SnapshotID
+	}
+	retentionMs := config.SnapshotRetentionMs
+	protected := protectedSnapshots(meta, nowMs)
+
+	// Sort snapshots by timestamp descending (most recent first) so that
+	// the keep-count logic always preserves the newest snapshots.
+	sorted := make([]table.Snapshot, len(meta.Snapshots()))
+	copy(sorted, meta.Snapshots())
+	sort.Slice(sorted, func(i, j int) bool {
+		return sorted[i].TimestampMs > sorted[j].TimestampMs
+	})
+
+	// Walk from newest to oldest. The current snapshot is always kept.
+	// Among the remaining, keep up to MaxSnapshotsToKeep-1 (since current
+	// counts toward the quota). Expire the rest only if they exceed the
+	// retention window; snapshots within the window are kept regardless.
+	var toExpire []int64
+	var kept int64
+	for _, snap := range sorted {
+		if snap.SnapshotID == currentSnapID {
+			kept++
+			continue
+		}
+		if _, isProtected := protected[snap.SnapshotID]; isProtected {
+			kept++
+			continue
+		}
+		if kept < config.MaxSnapshotsToKeep {
+			kept++
+			continue
+		}
+		if nowMs-snap.TimestampMs > retentionMs {
+			toExpire = append(toExpire, snap.SnapshotID)
+		} else {
+			kept++
+		}
+	}
+	return toExpire
+}
+
+// protectedSnapshots returns the snapshots that named refs hold in place.
+// A branch head or a tag pins its snapshot no matter how old it is, and a
+// branch may carry its own retention overrides for the ancestors behind it.
+// iceberg-go's RemoveSnapshots drops any ref whose snapshot is gone without
+// complaint, so expiring one of these would silently delete the tag and then
+// the files it pointed at.
+func protectedSnapshots(meta table.Metadata, nowMs int64) map[int64]struct{} {
+	protected := make(map[int64]struct{})
+	byID := make(map[int64]table.Snapshot)
+	for _, snap := range meta.Snapshots() {
+		byID[snap.SnapshotID] = snap
+	}
+
+	for _, ref := range meta.Refs() {
+		protected[ref.SnapshotID] = struct{}{}
+		if ref.SnapshotRefType != table.BranchRef {
+			continue
+		}
+		// Without overrides the branch keeps only its head here; the ancestors
+		// behind it stay under the worker's own retention config.
+		if ref.MinSnapshotsToKeep == nil && ref.MaxSnapshotAgeMs == nil {
+			continue
+		}
+		minToKeep := 1
+		if ref.MinSnapshotsToKeep != nil {
+			minToKeep = *ref.MinSnapshotsToKeep
+		}
+		seen := make(map[int64]struct{})
+		kept := 0
+		for id := ref.SnapshotID; ; {
+			if _, looped := seen[id]; looped {
+				break
+			}
+			seen[id] = struct{}{}
+			snap, ok := byID[id]
+			if !ok {
+				break
+			}
+			withinAge := ref.MaxSnapshotAgeMs != nil && nowMs-snap.TimestampMs <= *ref.MaxSnapshotAgeMs
+			if kept >= minToKeep && !withinAge {
+				break
+			}
+			protected[snap.SnapshotID] = struct{}{}
+			kept++
+			if snap.ParentSnapshotID == nil {
+				break
+			}
+			id = *snap.ParentSnapshotID
+		}
+	}
+	return protected
+}
+
 // collectSnapshotFiles returns all file paths (manifest lists, manifest files,
 // data files) referenced by the given snapshots. It returns an error if any
 // manifest list or manifest cannot be read/parsed, to prevent delete decisions