From 10b64686ba8b95470b48c6cc152cae3a7678f486 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 10 Aug 2026 18:25:03 -0700 Subject: [PATCH] filer: repack chunk TTLs round up and follow the S3 expiry anchor SecondsToTTL truncates to the volume TTL grid, so 3599 remaining seconds became 59m and anything under a minute became no TTL at all - permanent chunks under an expiring entry. Round the remaining lifetime up to the smallest representable value instead, and anchor it the way FindEntry expires entries: S3-expiring entries age from Mtime, others from Crtime, so a recently overwritten S3 object is no longer treated as nearly expired. Also bind each layout to a digest of the chunk list it described. Offset writes and appends keep Extended while changing the chunks, so a same-size partial write used to leave the old playlist and extents being served over new bytes; the views now detect the mismatch and answer 404 until the file is re-ingested or repacked. --- weed/server/filer_server_format.go | 55 +++++++++++++++++-- weed/server/filer_server_format_test.go | 70 +++++++++++++++++++++++++ 2 files changed, 121 insertions(+), 4 deletions(-) create mode 100644 weed/server/filer_server_format_test.go diff --git a/weed/server/filer_server_format.go b/weed/server/filer_server_format.go index 87afbe25c..82cefde3a 100644 --- a/weed/server/filer_server_format.go +++ b/weed/server/filer_server_format.go @@ -1,11 +1,13 @@ package weed_server import ( + "bytes" "context" "crypto/md5" "errors" "fmt" "io" + "math" "net/http" "os" "path" @@ -32,6 +34,10 @@ const ( formatIngestParam = "format.ingest" formatRepackParam = "format.repack" + // formatLayoutChunksKey binds a layout to the chunk list it described, so + // any other writer that changes the chunks invalidates the views. + formatLayoutChunksKey = "x-seaweedfs-format-layout-chunks" + maxFormatSidecarBytes = 16 << 20 formatSniffBytes = 512 // defaultFormatChunkSizeMB caps extent chunks when no maxMB is configured. @@ -39,6 +45,29 @@ const ( defaultFormatChunkSizeMB = 4 ) +// formatChunkIdentity digests the chunk list a layout was written against. +func formatChunkIdentity(chunks []*filer_pb.FileChunk) []byte { + digest := md5.New() + for _, chunk := range chunks { + fmt.Fprintf(digest, "%d:%s;", chunk.Offset, chunk.GetFileIdString()) + } + return digest.Sum(nil) +} + +// roundUpToVolumeTTL returns the smallest volume-TTL-representable seconds +// value not below the argument. A volume TTL is at most 255 of one unit and +// SecondsToTTL truncates anything else downward, which would let chunks +// expire before their entry - or, under a minute, never. +func roundUpToVolumeTTL(seconds int64) int32 { + for _, unit := range []int64{60, 3600, 24 * 3600, 7 * 24 * 3600, 30 * 24 * 3600, 365 * 24 * 3600} { + count := (seconds + unit - 1) / unit + if count <= 255 && count*unit <= math.MaxInt32 { + return int32(count * unit) + } + } + return math.MaxInt32 +} + // formatChunkSizeLimit mirrors the autoChunk maxMB resolution. func (fs *FilerServer) formatChunkSizeLimit(r *http.Request) int64 { parsedMaxMB, _ := strconv.ParseInt(r.URL.Query().Get("maxMB"), 10, 32) @@ -198,8 +227,11 @@ func (fs *FilerServer) formatIngest(ctx context.Context, w http.ResponseWriter, TtlSec: so.TtlSeconds, Mime: contentType, Md5: md5Hash.Sum(nil), FileSize: uint64(written), }, - Chunks: fileChunks, - Extended: map[string][]byte{format.LayoutKey: encoded}, + Chunks: fileChunks, + Extended: map[string][]byte{ + format.LayoutKey: encoded, + formatLayoutChunksKey: formatChunkIdentity(fileChunks), + }, } copyStandardHeadersToExtended(r, entry.Extended) // commit under the entry lock like saveMetaData, so ingest overwrites @@ -289,12 +321,18 @@ func (fs *FilerServer) formatRepack(ctx context.Context, w http.ResponseWriter, // of the original span. so.TtlSeconds = entry.TtlSec if entry.TtlSec > 0 { - remaining := int64(entry.TtlSec) - int64(time.Since(entry.Crtime)/time.Second) + // mirror FindEntry's expiry anchors: S3-expiring entries age from + // Mtime, everything else from Crtime + expiresAt := entry.Crtime.Add(time.Duration(entry.TtlSec) * time.Second) + if entry.IsExpireS3Enabled() { + expiresAt = entry.GetS3ExpireTime() + } + remaining := (int64(time.Until(expiresAt)) + int64(time.Second) - 1) / int64(time.Second) if remaining <= 0 { writeJsonError(w, r, http.StatusBadRequest, errors.New("entry TTL has already expired")) return } - so.TtlSeconds = int32(remaining) + so.TtlSeconds = roundUpToVolumeTTL(remaining) } size := int64(entry.FileSize) @@ -369,6 +407,7 @@ func (fs *FilerServer) formatRepack(ctx context.Context, w http.ResponseWriter, newEntry.Extended[k] = v } newEntry.Extended[format.LayoutKey] = encoded + newEntry.Extended[formatLayoutChunksKey] = formatChunkIdentity(newChunks) if len(newEntry.Md5) == 0 { newEntry.Md5 = md5Hash.Sum(nil) } @@ -411,6 +450,14 @@ func (fs *FilerServer) serveFormatView(ctx context.Context, w http.ResponseWrite http.Error(w, "format layout is stale", http.StatusNotFound) return } + // A write outside the format endpoints (offset writes, appends, mounts) + // changes the chunks but keeps Extended, so the layout no longer + // describes the bytes even when the total size still matches. + if !bytes.Equal(entry.Extended[formatLayoutChunksKey], formatChunkIdentity(entry.GetChunks())) { + glog.WarningfCtx(ctx, "format layout on %s no longer matches its chunks", entry.FullPath) + http.Error(w, "format layout is stale", http.StatusNotFound) + return + } // The view's validator must change when the layout or the requested // representation changes, even when the media bytes and their MD5 do not: diff --git a/weed/server/filer_server_format_test.go b/weed/server/filer_server_format_test.go new file mode 100644 index 000000000..332fa739b --- /dev/null +++ b/weed/server/filer_server_format_test.go @@ -0,0 +1,70 @@ +package weed_server + +import ( + "bytes" + "math" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func TestRoundUpToVolumeTTL(t *testing.T) { + tests := []struct { + seconds int64 + want int32 + }{ + {1, 60}, + {59, 60}, + {60, 60}, + {61, 120}, + {3599, 3600}, + {3600, 3600}, + {3601, 3660}, + {255 * 60, 255 * 60}, + {255*60 + 1, 5 * 3600}, // minutes overflow 255, ceil to hours + {20_000_000, 232 * 24 * 3600}, // ~231.5 days, ceil to days + {int64(math.MaxInt32), math.MaxInt32}, // beyond every unit's 255 cap + } + for _, test := range tests { + got := roundUpToVolumeTTL(test.seconds) + if got != test.want { + t.Fatalf("roundUpToVolumeTTL(%d) = %d, want %d", test.seconds, got, test.want) + } + if int64(got) < test.seconds && got != math.MaxInt32 { + t.Fatalf("roundUpToVolumeTTL(%d) = %d shortened the lifetime", test.seconds, got) + } + // the rounded value must survive the volume TTL string conversion intact + if got != math.MaxInt32 { + ttl, err := needle.ReadTTL(needle.SecondsToTTL(got)) + if err != nil || int64(ttl.Minutes())*60 != int64(got) { + t.Fatalf("SecondsToTTL(%d) = %q does not round-trip (err %v)", got, needle.SecondsToTTL(got), err) + } + } + } +} + +func TestFormatChunkIdentity(t *testing.T) { + chunks := []*filer_pb.FileChunk{ + {FileId: "1,ab", Offset: 0, Size: 10}, + {FileId: "2,cd", Offset: 10, Size: 20}, + } + identity := formatChunkIdentity(chunks) + if !bytes.Equal(identity, formatChunkIdentity(chunks)) { + t.Fatalf("identity is not deterministic") + } + changedFid := []*filer_pb.FileChunk{ + {FileId: "1,ab", Offset: 0, Size: 10}, + {FileId: "3,ef", Offset: 10, Size: 20}, + } + if bytes.Equal(identity, formatChunkIdentity(changedFid)) { + t.Fatalf("identity ignored a chunk replacement") + } + changedOffset := []*filer_pb.FileChunk{ + {FileId: "1,ab", Offset: 0, Size: 10}, + {FileId: "2,cd", Offset: 12, Size: 20}, + } + if bytes.Equal(identity, formatChunkIdentity(changedOffset)) { + t.Fatalf("identity ignored an offset change") + } +}