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

@@ -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:

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")
+	}
+}