format: restore the hls-ts sniff and assert adapter capabilities

Sniff identifies TS assets for the coming policy-driven alignment even
though nothing reaches it through repack today. Compile-time assertions
now state each adapter's capabilities explicitly.

weed/filer/filechunk_manifest.go

@@ -236,7 +236,10 @@ func doMaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*fil
 	for i := 0; i+mergeFactor <= len(dataChunks); i += mergeFactor {
 		chunk, err := mergefn(saveFunc, dataChunks[i:i+mergeFactor])
 		if err != nil {
-			return dataChunks, err
+			// Return the manifests already written plus the chunks not yet
+			// wrapped: a complete, deletable representation of every byte, so
+			// callers can clean up or keep a usable chunk list.
+			return append(chunks, dataChunks[i:]...), err
 		}
 		chunks = append(chunks, chunk)
 		remaining -= mergeFactor

weed/filer/filechunk_manifest_test.go

@@ -77,6 +77,38 @@ func TestDoMaybeManifestize(t *testing.T) {
 		actual, _ := doMaybeManifestize(nil, mtest.inputs, 2, mockMerge)
 		assertEqualChunks(t, mtest.expected, actual)
 	}
+}
+
+// A mid-run merge failure must still return every chunk that exists: the
+// manifests already written plus the chunks not yet wrapped, so callers can
+// delete or keep a complete set.
+func TestDoMaybeManifestizePartialFailure(t *testing.T) {
+	inputs := []*filer_pb.FileChunk{
+		{FileId: "0", IsChunkManifest: true},
+		{FileId: "1", IsChunkManifest: false},
+		{FileId: "2", IsChunkManifest: false},
+		{FileId: "3", IsChunkManifest: false},
+		{FileId: "4", IsChunkManifest: false},
+	}
+	calls := 0
+	failingMerge := func(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer_pb.FileChunk) (*filer_pb.FileChunk, error) {
+		calls++
+		if calls > 1 {
+			return nil, fmt.Errorf("merge failed")
+		}
+		return mockMerge(saveFunc, dataChunks)
+	}
+	actual, err := doMaybeManifestize(nil, inputs, 2, failingMerge)
+	if err == nil {
+		t.Fatalf("doMaybeManifestize() expected an error")
+	}
+	expected := []*filer_pb.FileChunk{
+		{FileId: "0", IsChunkManifest: true},
+		{FileId: "12", IsChunkManifest: true},
+		{FileId: "3", IsChunkManifest: false},
+		{FileId: "4", IsChunkManifest: false},
+	}
+	assertEqualChunks(t, expected, actual)
 
 }
 

weed/format/format.go

@@ -0,0 +1,93 @@
+// Package format maps the internal structure of container file formats onto
+// storage chunk boundaries.
+//
+// An adapter translates one format into three things the core understands: a
+// list of extent sizes, an alignment quantum, and an opaque payload. Adapters
+// never see chunks, file ids, or authorization; the core never learns what an
+// MPEG-TS packet or a parquet row group is.
+package format
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/url"
+)
+
+// LayoutKey is the filer Extended key holding the encoded Layout. The
+// x-seaweedfs- prefix keeps it out of HTTP response headers.
+const LayoutKey = "x-seaweedfs-format-layout"
+
+// ViewParam is the query parameter selecting a format view on GET requests.
+// Adapters rendering self-referential URLs must use it.
+const ViewParam = "format.view"
+
+// ErrNoSuchView reports that a view request addresses nothing servable; the
+// server answers 404.
+var ErrNoSuchView = errors.New("no such view")
+
+// Layout describes how a file's structure maps to byte extents.
+type Layout struct {
+	Format      string  // adapter name
+	ExtentSizes []int64 // extent lengths in file order; they sum to the file size
+	Align       int64   // quantum for cutting inside an oversized extent; 1 cuts anywhere
+	Payload     []byte  // adapter-owned metadata, opaque to the core
+}
+
+// Hint carries the cheap identification signals available to Sniff.
+type Hint struct {
+	Name        string
+	ContentType string
+	Size        int64
+	Head        []byte
+	Tail        []byte
+}
+
+// Format is the mandatory adapter identity. Capabilities beyond it are
+// discovered by type assertion.
+type Format interface {
+	Name() string
+}
+
+// Sniffer cheaply recognizes the format from identification signals. Repack
+// gates on it before parsing, and policy-driven detection will rely on it;
+// ingest-only adapters, whose files carry their layout from birth, skip it.
+type Sniffer interface {
+	Sniff(h Hint) bool
+}
+
+// Indexer derives a Layout from the complete stored bytes.
+type Indexer interface {
+	Index(ctx context.Context, r io.ReaderAt, size int64) (*Layout, error)
+}
+
+// SidecarIndexer derives a Layout from an external index document supplied at
+// ingest, before the media bytes arrive.
+type SidecarIndexer interface {
+	IndexSidecar(sidecar []byte) (*Layout, error)
+}
+
+// Object is everything a Viewer may know about the file it serves.
+type Object struct {
+	Name   string
+	Size   int64
+	Layout *Layout
+}
+
+// ViewRequest carries the request parameters of a ?view= request.
+type ViewRequest struct {
+	Query url.Values
+}
+
+// ViewPlan tells the server what to serve. The server executes it on the
+// normal streaming path; adapters stay pure functions of request and layout.
+type ViewPlan struct {
+	ContentType string
+	Body        []byte // rendered document; when nil, stream Extent instead
+	Extent      int
+}
+
+// Viewer answers ?view= requests.
+type Viewer interface {
+	View(req ViewRequest, obj Object) (*ViewPlan, error)
+}

weed/format/formattest/formattest.go

@@ -0,0 +1,64 @@
+// Package formattest is the conformance kit format adapters must pass:
+// indexers parse attacker-controlled bytes inside a storage daemon, so they
+// must never panic and every layout they accept must validate.
+package formattest
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/format"
+)
+
+// IndexTruncations feeds progressively truncated copies of a valid file to the
+// indexer. Any outcome is acceptable except a panic or an invalid layout.
+func IndexTruncations(t *testing.T, indexer format.Indexer, data []byte) {
+	t.Helper()
+	for i := 0; i <= 16; i++ {
+		size := int64(len(data) * i / 16)
+		layout, err := indexer.Index(context.Background(), bytes.NewReader(data[:size]), size)
+		if err != nil {
+			continue
+		}
+		if validateErr := layout.Validate(size); validateErr != nil {
+			t.Fatalf("Index() at %d/%d bytes returned an invalid layout: %v", size, len(data), validateErr)
+		}
+	}
+}
+
+// SidecarTruncations does the same for sidecar index documents.
+func SidecarTruncations(t *testing.T, indexer format.SidecarIndexer, sidecar []byte) {
+	t.Helper()
+	for i := 0; i <= 16; i++ {
+		layout, err := indexer.IndexSidecar(sidecar[:len(sidecar)*i/16])
+		if err != nil {
+			continue
+		}
+		if validateErr := layout.Validate(-1); validateErr != nil {
+			t.Fatalf("IndexSidecar() at %d/16 returned an invalid layout: %v", i, validateErr)
+		}
+	}
+}
+
+// EncodeRoundTrip checks that a layout survives the persistence codec.
+func EncodeRoundTrip(t *testing.T, layout *format.Layout) {
+	t.Helper()
+	encoded, err := layout.Encode()
+	if err != nil {
+		t.Fatalf("Encode() error = %v", err)
+	}
+	decoded, err := format.DecodeLayout(encoded)
+	if err != nil {
+		t.Fatalf("DecodeLayout() error = %v", err)
+	}
+	if decoded.Format != layout.Format || decoded.Align != layout.Align ||
+		len(decoded.ExtentSizes) != len(layout.ExtentSizes) || !bytes.Equal(decoded.Payload, layout.Payload) {
+		t.Fatalf("decoded layout %+v differs from %+v", decoded, layout)
+	}
+	for i := range layout.ExtentSizes {
+		if decoded.ExtentSizes[i] != layout.ExtentSizes[i] {
+			t.Fatalf("extent %d = %d, want %d", i, decoded.ExtentSizes[i], layout.ExtentSizes[i])
+		}
+	}
+}

weed/format/hlsts/hlsts.go

@@ -0,0 +1,284 @@
+// Package hlsts adapts single-file HLS MPEG-TS VOD assets: the ingest sidecar
+// is an FFmpeg-style EXT-X-BYTERANGE media playlist, extents are its segments,
+// and the view serves a rewritten playlist with plain numbered segment URLs.
+package hlsts
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"math"
+	"net/url"
+	"strconv"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/format"
+)
+
+const (
+	FormatName = "hls-ts"
+	// TSPacketSize is the fixed MPEG-TS packet size; segment boundaries and
+	// interior chunk cuts land on packet boundaries.
+	TSPacketSize = 188
+
+	tsSyncByte = 0x47
+
+	PlaylistContentType = "application/vnd.apple.mpegurl"
+	MediaContentType    = "video/MP2T"
+)
+
+func init() {
+	format.Register(Adapter{})
+}
+
+type Adapter struct{}
+
+var (
+	_ format.Sniffer        = Adapter{}
+	_ format.SidecarIndexer = Adapter{}
+	_ format.Viewer         = Adapter{}
+)
+
+func (Adapter) Name() string { return FormatName }
+
+func (Adapter) Sniff(h format.Hint) bool {
+	if len(h.Head) > TSPacketSize {
+		return h.Head[0] == tsSyncByte && h.Head[TSPacketSize] == tsSyncByte
+	}
+	return len(h.Head) > 0 && h.Head[0] == tsSyncByte
+}
+
+// playlistInfo is the adapter payload: what the generated playback playlist
+// needs beyond the extent sizes.
+type playlistInfo struct {
+	TargetDuration int64
+	MediaSequence  int64
+	DurationsMs    []int64
+}
+
+func (p *playlistInfo) encode() []byte {
+	out := binary.AppendUvarint(nil, uint64(p.TargetDuration))
+	out = binary.AppendUvarint(out, uint64(p.MediaSequence))
+	for _, durationMs := range p.DurationsMs {
+		out = binary.AppendUvarint(out, uint64(durationMs))
+	}
+	return out
+}
+
+func decodePlaylistInfo(payload []byte, extentCount int) (*playlistInfo, error) {
+	reader := bytes.NewReader(payload)
+	target, err := binary.ReadUvarint(reader)
+	if err != nil || target == 0 || target > math.MaxInt32 {
+		return nil, fmt.Errorf("invalid hls-ts target duration")
+	}
+	// mirror the ingest bound: the last segment number is sequence+count-1
+	sequence, err := binary.ReadUvarint(reader)
+	if err != nil || sequence > math.MaxInt64-uint64(extentCount-1) {
+		return nil, fmt.Errorf("invalid hls-ts media sequence")
+	}
+	info := &playlistInfo{TargetDuration: int64(target), MediaSequence: int64(sequence), DurationsMs: make([]int64, extentCount)}
+	for i := range info.DurationsMs {
+		durationMs, err := binary.ReadUvarint(reader)
+		if err != nil || durationMs == 0 || durationMs > math.MaxInt32 {
+			return nil, fmt.Errorf("invalid hls-ts segment %d duration", i)
+		}
+		info.DurationsMs[i] = int64(durationMs)
+	}
+	if reader.Len() != 0 {
+		return nil, fmt.Errorf("hls-ts payload has trailing bytes")
+	}
+	return info, nil
+}
+
+// IndexSidecar parses a VOD media playlist whose segments reference one shared
+// media URI through EXT-X-BYTERANGE. Playlist state the generated playback
+// playlist cannot reproduce is rejected.
+func (Adapter) IndexSidecar(sidecar []byte) (*format.Layout, error) {
+	scanner := bufio.NewScanner(bytes.NewReader(sidecar))
+	scanner.Buffer(make([]byte, 64*1024), 1<<20)
+
+	info := &playlistInfo{}
+	var sizes []int64
+	var pendingDuration int64 // ms; 0 = no EXTINF pending
+	var pendingSize, pendingOffset int64
+	var havePendingRange bool
+	var expectedOffset int64
+	var mediaURI string
+	var sawHeader, sawEndList bool
+	var maxDurationMs int64
+
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		switch {
+		case line == "":
+		case line == "#EXTM3U":
+			sawHeader = true
+		case line == "#EXT-X-ENDLIST":
+			sawEndList = true
+		case strings.HasPrefix(line, "#EXT-X-KEY:"),
+			line == "#EXT-X-DISCONTINUITY",
+			strings.HasPrefix(line, "#EXT-X-DISCONTINUITY-SEQUENCE:"),
+			strings.HasPrefix(line, "#EXT-X-MAP:"),
+			line == "#EXT-X-GAP",
+			line == "#EXT-X-I-FRAMES-ONLY":
+			return nil, fmt.Errorf("%s is not supported by hls-ts ingest", strings.SplitN(line, ":", 2)[0])
+		case strings.HasPrefix(line, "#EXT-X-TARGETDURATION:"):
+			value := strings.TrimSpace(strings.TrimPrefix(line, "#EXT-X-TARGETDURATION:"))
+			target, err := strconv.ParseInt(value, 10, 32)
+			if err != nil || target <= 0 {
+				return nil, fmt.Errorf("invalid EXT-X-TARGETDURATION %q", value)
+			}
+			info.TargetDuration = target
+		case strings.HasPrefix(line, "#EXT-X-MEDIA-SEQUENCE:"):
+			value := strings.TrimSpace(strings.TrimPrefix(line, "#EXT-X-MEDIA-SEQUENCE:"))
+			sequence, err := strconv.ParseInt(value, 10, 64)
+			if err != nil || sequence < 0 {
+				return nil, fmt.Errorf("invalid EXT-X-MEDIA-SEQUENCE %q", value)
+			}
+			info.MediaSequence = sequence
+		case strings.HasPrefix(line, "#EXTINF:"):
+			if pendingDuration != 0 {
+				return nil, errors.New("EXTINF without a media URI for the previous segment")
+			}
+			value := strings.TrimSpace(strings.TrimPrefix(line, "#EXTINF:"))
+			if comma := strings.IndexByte(value, ','); comma >= 0 {
+				value = value[:comma]
+			}
+			seconds, err := strconv.ParseFloat(value, 64)
+			if err != nil || seconds <= 0 || math.IsNaN(seconds) || math.IsInf(seconds, 0) || seconds > math.MaxInt32/1000 {
+				return nil, fmt.Errorf("invalid EXTINF duration %q", value)
+			}
+			pendingDuration = int64(math.Round(seconds * 1000))
+			if pendingDuration == 0 {
+				pendingDuration = 1
+			}
+		case strings.HasPrefix(line, "#EXT-X-BYTERANGE:"):
+			if pendingDuration == 0 {
+				return nil, errors.New("EXT-X-BYTERANGE without a preceding EXTINF")
+			}
+			value := strings.TrimSpace(strings.TrimPrefix(line, "#EXT-X-BYTERANGE:"))
+			lengthText, offsetText, hasOffset := strings.Cut(value, "@")
+			length, err := strconv.ParseInt(lengthText, 10, 64)
+			if err != nil || length <= 0 {
+				return nil, fmt.Errorf("invalid EXT-X-BYTERANGE length %q", value)
+			}
+			offset := expectedOffset
+			if hasOffset {
+				if offset, err = strconv.ParseInt(offsetText, 10, 64); err != nil || offset < 0 {
+					return nil, fmt.Errorf("invalid EXT-X-BYTERANGE offset %q", value)
+				}
+			}
+			pendingSize, pendingOffset, havePendingRange = length, offset, true
+		case strings.HasPrefix(line, "#"):
+			// other tags carry no state the generated playlist must keep
+		default:
+			if pendingDuration == 0 || !havePendingRange {
+				return nil, fmt.Errorf("media URI %q without EXTINF and EXT-X-BYTERANGE", line)
+			}
+			if mediaURI == "" {
+				mediaURI = line
+			} else if mediaURI != line {
+				return nil, errors.New("hls-ts ingest requires one shared media URI")
+			}
+			if pendingOffset != expectedOffset {
+				return nil, fmt.Errorf("non-contiguous byte range at segment %d: offset %d, expected %d", len(sizes), pendingOffset, expectedOffset)
+			}
+			if pendingSize%TSPacketSize != 0 {
+				return nil, fmt.Errorf("segment %d size %d is not a multiple of the %d-byte TS packet", len(sizes), pendingSize, TSPacketSize)
+			}
+			if pendingSize > math.MaxInt64-expectedOffset {
+				return nil, fmt.Errorf("byte range at segment %d overflows", len(sizes))
+			}
+			if len(sizes) >= format.MaxExtentCount {
+				return nil, fmt.Errorf("playlist has more than %d segments", format.MaxExtentCount)
+			}
+			sizes = append(sizes, pendingSize)
+			info.DurationsMs = append(info.DurationsMs, pendingDuration)
+			if pendingDuration > maxDurationMs {
+				maxDurationMs = pendingDuration
+			}
+			expectedOffset += pendingSize
+			pendingDuration, havePendingRange = 0, false
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("read playlist: %w", err)
+	}
+	if !sawHeader {
+		return nil, errors.New("playlist is missing EXTM3U")
+	}
+	if pendingDuration != 0 || havePendingRange {
+		return nil, errors.New("playlist ended with an incomplete media segment")
+	}
+	if len(sizes) == 0 {
+		return nil, errors.New("playlist has no media segments")
+	}
+	if !sawEndList {
+		return nil, errors.New("only VOD playlists with EXT-X-ENDLIST are supported")
+	}
+	if info.MediaSequence > math.MaxInt64-int64(len(sizes)-1) {
+		return nil, errors.New("EXT-X-MEDIA-SEQUENCE overflows segment numbering")
+	}
+
+	// RFC 8216: EXT-X-TARGETDURATION must be at least each segment duration
+	// rounded to the nearest integer.
+	minimumTarget := (maxDurationMs + 500) / 1000
+	if minimumTarget < 1 {
+		minimumTarget = 1
+	}
+	if info.TargetDuration == 0 {
+		info.TargetDuration = minimumTarget
+	} else if info.TargetDuration < minimumTarget {
+		return nil, fmt.Errorf("EXT-X-TARGETDURATION %d is smaller than the longest segment duration %d", info.TargetDuration, minimumTarget)
+	}
+
+	layout := &format.Layout{
+		Format:      FormatName,
+		ExtentSizes: sizes,
+		Align:       TSPacketSize,
+		Payload:     info.encode(),
+	}
+	// valid by construction, but enforce the formattest invariant explicitly
+	if err := layout.Validate(-1); err != nil {
+		return nil, err
+	}
+	return layout, nil
+}
+
+// View serves the generated playlist, or maps ?seq=N to its extent.
+func (Adapter) View(req format.ViewRequest, obj format.Object) (*format.ViewPlan, error) {
+	info, err := decodePlaylistInfo(obj.Layout.Payload, len(obj.Layout.ExtentSizes))
+	if err != nil {
+		return nil, err
+	}
+	sequenceText := req.Query.Get("seq")
+	if sequenceText == "" {
+		return &format.ViewPlan{ContentType: PlaylistContentType, Body: renderPlaylist(obj.Name, info)}, nil
+	}
+	sequence, err := strconv.ParseInt(sequenceText, 10, 64)
+	if err != nil || sequence < info.MediaSequence {
+		return nil, format.ErrNoSuchView
+	}
+	index := sequence - info.MediaSequence
+	if index >= int64(len(obj.Layout.ExtentSizes)) {
+		return nil, format.ErrNoSuchView
+	}
+	return &format.ViewPlan{ContentType: MediaContentType, Extent: int(index)}, nil
+}
+
+func renderPlaylist(name string, info *playlistInfo) []byte {
+	var out strings.Builder
+	out.WriteString("#EXTM3U\n#EXT-X-VERSION:3\n")
+	fmt.Fprintf(&out, "#EXT-X-TARGETDURATION:%d\n", info.TargetDuration)
+	fmt.Fprintf(&out, "#EXT-X-MEDIA-SEQUENCE:%d\n", info.MediaSequence)
+	out.WriteString("#EXT-X-PLAYLIST-TYPE:VOD\n")
+	escapedName := url.PathEscape(name)
+	for i, durationMs := range info.DurationsMs {
+		fmt.Fprintf(&out, "#EXTINF:%.3f,\n", float64(durationMs)/1000)
+		fmt.Fprintf(&out, "%s?%s=%s&seq=%d\n", escapedName, format.ViewParam, FormatName, int64(i)+info.MediaSequence)
+	}
+	out.WriteString("#EXT-X-ENDLIST\n")
+	return []byte(out.String())
+}

weed/format/hlsts/hlsts_test.go

@@ -0,0 +1,204 @@
+package hlsts
+
+import (
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/format"
+	"github.com/seaweedfs/seaweedfs/weed/format/formattest"
+)
+
+const ffmpegPlaylist = `#EXTM3U
+#EXT-X-VERSION:4
+#EXT-X-TARGETDURATION:6
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-PLAYLIST-TYPE:VOD
+#EXTINF:6.000000,
+#EXT-X-BYTERANGE:1128@0
+video.ts
+#EXTINF:6.000000,
+#EXT-X-BYTERANGE:940
+video.ts
+#EXTINF:2.500000,
+#EXT-X-BYTERANGE:376@2068
+video.ts
+#EXT-X-ENDLIST
+`
+
+func TestIndexSidecar(t *testing.T) {
+	layout, err := Adapter{}.IndexSidecar([]byte(ffmpegPlaylist))
+	if err != nil {
+		t.Fatalf("IndexSidecar() error = %v", err)
+	}
+	wantSizes := []int64{1128, 940, 376}
+	if len(layout.ExtentSizes) != len(wantSizes) {
+		t.Fatalf("extents = %v, want %v", layout.ExtentSizes, wantSizes)
+	}
+	for i := range wantSizes {
+		if layout.ExtentSizes[i] != wantSizes[i] {
+			t.Fatalf("extent %d = %d, want %d", i, layout.ExtentSizes[i], wantSizes[i])
+		}
+	}
+	if layout.Align != TSPacketSize || layout.Format != FormatName {
+		t.Fatalf("layout = %+v", layout)
+	}
+	if err := layout.Validate(1128 + 940 + 376); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+	formattest.EncodeRoundTrip(t, layout)
+}
+
+func TestIndexSidecarDefaultsTargetDuration(t *testing.T) {
+	playlist := "#EXTM3U\n#EXTINF:5.6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n"
+	layout, err := Adapter{}.IndexSidecar([]byte(playlist))
+	if err != nil {
+		t.Fatalf("IndexSidecar() error = %v", err)
+	}
+	info, err := decodePlaylistInfo(layout.Payload, len(layout.ExtentSizes))
+	if err != nil {
+		t.Fatalf("decodePlaylistInfo() error = %v", err)
+	}
+	if info.TargetDuration != 6 {
+		t.Fatalf("TargetDuration = %d, want 6", info.TargetDuration)
+	}
+}
+
+func TestIndexSidecarRejections(t *testing.T) {
+	tests := []struct {
+		name     string
+		playlist string
+		wantErr  string
+	}{
+		{"missing header", "#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "EXTM3U"},
+		{"missing endlist", "#EXTM3U\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n", "EXT-X-ENDLIST"},
+		{"no segments", "#EXTM3U\n#EXT-X-ENDLIST\n", "no media segments"},
+		{"encryption", "#EXTM3U\n#EXT-X-KEY:METHOD=AES-128\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "EXT-X-KEY"},
+		{"discontinuity", "#EXTM3U\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-DISCONTINUITY\n#EXT-X-ENDLIST\n", "EXT-X-DISCONTINUITY"},
+		{"map", "#EXTM3U\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "EXT-X-MAP"},
+		{"gap", "#EXTM3U\n#EXT-X-GAP\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "EXT-X-GAP"},
+		{"iframes only", "#EXTM3U\n#EXT-X-I-FRAMES-ONLY\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "EXT-X-I-FRAMES-ONLY"},
+		{"no byterange", "#EXTM3U\n#EXTINF:6,\nv.ts\n#EXT-X-ENDLIST\n", "EXT-X-BYTERANGE"},
+		{"gap in ranges", "#EXTM3U\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@376\nv.ts\n#EXT-X-ENDLIST\n", "non-contiguous"},
+		{"unaligned size", "#EXTM3U\n#EXTINF:6,\n#EXT-X-BYTERANGE:100@0\nv.ts\n#EXT-X-ENDLIST\n", "TS packet"},
+		{"two media files", "#EXTM3U\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\na.ts\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@188\nb.ts\n#EXT-X-ENDLIST\n", "one shared media URI"},
+		{"target too small", "#EXTM3U\n#EXT-X-TARGETDURATION:2\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "smaller than"},
+		{"zero duration", "#EXTM3U\n#EXTINF:0,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXT-X-ENDLIST\n", "EXTINF"},
+		{"dangling extinf", "#EXTM3U\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXTINF:6,\n#EXT-X-ENDLIST\n", "incomplete"},
+	}
+	for _, test := range tests {
+		_, err := Adapter{}.IndexSidecar([]byte(test.playlist))
+		if err == nil || !strings.Contains(err.Error(), test.wantErr) {
+			t.Fatalf("%s: error = %v, want %q", test.name, err, test.wantErr)
+		}
+	}
+}
+
+func TestSidecarTruncations(t *testing.T) {
+	formattest.SidecarTruncations(t, Adapter{}, []byte(ffmpegPlaylist))
+}
+
+func viewObject(t *testing.T) format.Object {
+	t.Helper()
+	layout, err := Adapter{}.IndexSidecar([]byte(ffmpegPlaylist))
+	if err != nil {
+		t.Fatalf("IndexSidecar() error = %v", err)
+	}
+	return format.Object{Name: "movie.ts", Size: layout.TotalSize(), Layout: layout}
+}
+
+func TestViewPlaylist(t *testing.T) {
+	plan, err := Adapter{}.View(format.ViewRequest{Query: url.Values{}}, viewObject(t))
+	if err != nil {
+		t.Fatalf("View() error = %v", err)
+	}
+	if plan.ContentType != PlaylistContentType {
+		t.Fatalf("ContentType = %q", plan.ContentType)
+	}
+	want := `#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-TARGETDURATION:6
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-PLAYLIST-TYPE:VOD
+#EXTINF:6.000,
+movie.ts?format.view=hls-ts&seq=0
+#EXTINF:6.000,
+movie.ts?format.view=hls-ts&seq=1
+#EXTINF:2.500,
+movie.ts?format.view=hls-ts&seq=2
+#EXT-X-ENDLIST
+`
+	if string(plan.Body) != want {
+		t.Fatalf("playlist = %q, want %q", plan.Body, want)
+	}
+}
+
+func TestViewSegment(t *testing.T) {
+	obj := viewObject(t)
+	plan, err := Adapter{}.View(format.ViewRequest{Query: url.Values{"seq": {"1"}}}, obj)
+	if err != nil {
+		t.Fatalf("View() error = %v", err)
+	}
+	if plan.Body != nil || plan.Extent != 1 || plan.ContentType != MediaContentType {
+		t.Fatalf("plan = %+v", plan)
+	}
+	for _, bad := range []string{"3", "-1", "x", "9999999999999999999"} {
+		if _, err := (Adapter{}).View(format.ViewRequest{Query: url.Values{"seq": {bad}}}, obj); err != format.ErrNoSuchView {
+			t.Fatalf("seq %q: error = %v, want ErrNoSuchView", bad, err)
+		}
+	}
+}
+
+func TestViewSegmentHonorsMediaSequence(t *testing.T) {
+	playlist := "#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:10\n#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXTINF:6,\n#EXT-X-BYTERANGE:376\nv.ts\n#EXT-X-ENDLIST\n"
+	layout, err := Adapter{}.IndexSidecar([]byte(playlist))
+	if err != nil {
+		t.Fatalf("IndexSidecar() error = %v", err)
+	}
+	obj := format.Object{Name: "v.ts", Size: layout.TotalSize(), Layout: layout}
+	plan, err := Adapter{}.View(format.ViewRequest{Query: url.Values{"seq": {"11"}}}, obj)
+	if err != nil {
+		t.Fatalf("View() error = %v", err)
+	}
+	if plan.Extent != 1 {
+		t.Fatalf("Extent = %d, want 1", plan.Extent)
+	}
+	if _, err := (Adapter{}).View(format.ViewRequest{Query: url.Values{"seq": {"9"}}}, obj); err != format.ErrNoSuchView {
+		t.Fatalf("seq below media sequence: error = %v, want ErrNoSuchView", err)
+	}
+}
+
+func TestSniff(t *testing.T) {
+	head := make([]byte, 400)
+	head[0], head[TSPacketSize] = tsSyncByte, tsSyncByte
+	if !(Adapter{}).Sniff(format.Hint{Head: head}) {
+		t.Fatalf("Sniff() rejected TS head")
+	}
+	head[TSPacketSize] = 0
+	if (Adapter{}).Sniff(format.Hint{Head: head}) {
+		t.Fatalf("Sniff() accepted non-TS head")
+	}
+}
+
+// The ingest bound admits mediaSequence = MaxInt64-(count-1); the payload
+// decoder must accept the same boundary or every view of such an asset fails.
+func TestMediaSequenceBoundaryRoundTrip(t *testing.T) {
+	playlist := "#EXTM3U\n#EXT-X-MEDIA-SEQUENCE:9223372036854775806\n" +
+		"#EXTINF:6,\n#EXT-X-BYTERANGE:188@0\nv.ts\n#EXTINF:6,\n#EXT-X-BYTERANGE:188\nv.ts\n#EXT-X-ENDLIST\n"
+	layout, err := Adapter{}.IndexSidecar([]byte(playlist))
+	if err != nil {
+		t.Fatalf("IndexSidecar() error = %v", err)
+	}
+	obj := format.Object{Name: "v.ts", Size: layout.TotalSize(), Layout: layout}
+	if _, err := (Adapter{}).View(format.ViewRequest{Query: url.Values{}}, obj); err != nil {
+		t.Fatalf("playlist view error = %v", err)
+	}
+	plan, err := Adapter{}.View(format.ViewRequest{Query: url.Values{"seq": {"9223372036854775807"}}}, obj)
+	if err != nil {
+		t.Fatalf("last segment view error = %v", err)
+	}
+	if plan.Extent != 1 {
+		t.Fatalf("Extent = %d, want 1", plan.Extent)
+	}
+}
+

weed/format/layout.go

@@ -0,0 +1,204 @@
+package format
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"math"
+	"sort"
+)
+
+const (
+	layoutVersion = 1
+
+	// MaxExtentCount bounds decoded layouts; it also bounds the chunk count a
+	// layout can force on an entry.
+	MaxExtentCount = 1 << 20
+	// MaxPayloadBytes bounds the adapter payload carried in entry metadata.
+	MaxPayloadBytes = 16 << 20
+	// MaxFormatNameBytes bounds the encoded adapter name.
+	MaxFormatNameBytes = 256
+)
+
+// TotalSize returns the file size the layout describes.
+func (l *Layout) TotalSize() int64 {
+	var total int64
+	for _, size := range l.ExtentSizes {
+		total += size
+	}
+	return total
+}
+
+// ExtentRange returns the byte range of extent i.
+func (l *Layout) ExtentRange(i int) (offset, size int64, ok bool) {
+	if i < 0 || i >= len(l.ExtentSizes) {
+		return 0, 0, false
+	}
+	for _, extentSize := range l.ExtentSizes[:i] {
+		offset += extentSize
+	}
+	return offset, l.ExtentSizes[i], true
+}
+
+// Validate checks layout consistency. A negative fileSize skips the total size
+// check.
+func (l *Layout) Validate(fileSize int64) error {
+	if l.Format == "" {
+		return fmt.Errorf("layout has no format name")
+	}
+	if len(l.Format) > MaxFormatNameBytes {
+		return fmt.Errorf("layout format name is too long: %d bytes", len(l.Format))
+	}
+	if l.Align < 1 {
+		return fmt.Errorf("layout align %d is invalid", l.Align)
+	}
+	if len(l.ExtentSizes) == 0 {
+		return fmt.Errorf("layout has no extents")
+	}
+	if len(l.ExtentSizes) > MaxExtentCount {
+		return fmt.Errorf("layout has too many extents: %d", len(l.ExtentSizes))
+	}
+	if len(l.Payload) > MaxPayloadBytes {
+		return fmt.Errorf("layout payload is too large: %d bytes", len(l.Payload))
+	}
+	var total int64
+	for i, size := range l.ExtentSizes {
+		if size <= 0 {
+			return fmt.Errorf("extent %d has invalid size %d", i, size)
+		}
+		if size > math.MaxInt64-total {
+			return fmt.Errorf("extent %d overflows the file size", i)
+		}
+		total += size
+	}
+	if fileSize >= 0 && total != fileSize {
+		return fmt.Errorf("layout describes %d bytes but the file has %d", total, fileSize)
+	}
+	return nil
+}
+
+// Encode serializes the layout for the LayoutKey extended attribute.
+func (l *Layout) Encode() ([]byte, error) {
+	if err := l.Validate(-1); err != nil {
+		return nil, err
+	}
+	out := []byte{layoutVersion}
+	out = binary.AppendUvarint(out, uint64(len(l.Format)))
+	out = append(out, l.Format...)
+	out = binary.AppendUvarint(out, uint64(l.Align))
+	out = binary.AppendUvarint(out, uint64(len(l.ExtentSizes)))
+	for _, size := range l.ExtentSizes {
+		out = binary.AppendUvarint(out, uint64(size))
+	}
+	out = binary.AppendUvarint(out, uint64(len(l.Payload)))
+	out = append(out, l.Payload...)
+	return out, nil
+}
+
+// DecodeLayout parses an encoded layout and validates it.
+func DecodeLayout(data []byte) (*Layout, error) {
+	reader := bytes.NewReader(data)
+	version, err := reader.ReadByte()
+	if err != nil || version != layoutVersion {
+		return nil, fmt.Errorf("unsupported layout version")
+	}
+	name, err := readUvarintBytes(reader, MaxFormatNameBytes)
+	if err != nil {
+		return nil, fmt.Errorf("read layout format: %w", err)
+	}
+	align, err := binary.ReadUvarint(reader)
+	if err != nil || align > math.MaxInt64 {
+		return nil, fmt.Errorf("read layout align: invalid")
+	}
+	count, err := binary.ReadUvarint(reader)
+	if err != nil || count > MaxExtentCount {
+		return nil, fmt.Errorf("read layout extent count: invalid")
+	}
+	sizes := make([]int64, count)
+	for i := range sizes {
+		size, err := binary.ReadUvarint(reader)
+		if err != nil || size > math.MaxInt64 {
+			return nil, fmt.Errorf("read extent %d size: invalid", i)
+		}
+		sizes[i] = int64(size)
+	}
+	payload, err := readUvarintBytes(reader, MaxPayloadBytes)
+	if err != nil {
+		return nil, fmt.Errorf("read layout payload: %w", err)
+	}
+	if reader.Len() != 0 {
+		return nil, fmt.Errorf("layout has %d trailing bytes", reader.Len())
+	}
+	layout := &Layout{Format: string(name), ExtentSizes: sizes, Align: int64(align), Payload: payload}
+	if err := layout.Validate(-1); err != nil {
+		return nil, err
+	}
+	return layout, nil
+}
+
+func readUvarintBytes(reader *bytes.Reader, limit uint64) ([]byte, error) {
+	length, err := binary.ReadUvarint(reader)
+	if err != nil {
+		return nil, err
+	}
+	if length > limit || length > uint64(reader.Len()) {
+		return nil, fmt.Errorf("length %d is out of bounds", length)
+	}
+	if length == 0 {
+		return nil, nil
+	}
+	data := make([]byte, length)
+	if _, err := reader.Read(data); err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
+// Cutter yields upload chunk boundaries: every extent boundary, plus
+// align-quantized cuts inside extents larger than maxChunkSize. A non-positive
+// maxChunkSize keeps each extent in one chunk. Interior cuts are computed
+// lazily, so memory stays bounded by the extent count no matter how large an
+// untrusted layout declares its extents to be.
+type Cutter struct {
+	starts  []int64 // extent start offsets, ascending
+	total   int64
+	quantum int64 // interior cut spacing; 0 = one chunk per extent
+}
+
+func (l *Layout) Cutter(maxChunkSize int64) *Cutter {
+	quantum := maxChunkSize
+	if quantum > 0 && l.Align > 1 {
+		quantum -= quantum % l.Align
+		if quantum <= 0 {
+			// An align larger than the chunk limit still cuts on whole atoms.
+			quantum = l.Align
+		}
+	}
+	starts := make([]int64, len(l.ExtentSizes))
+	var offset int64
+	for i, size := range l.ExtentSizes {
+		starts[i] = offset
+		offset += size
+	}
+	return &Cutter{starts: starts, total: offset, quantum: quantum}
+}
+
+// NextChunkSize returns the size of the chunk starting at offset, or 0 past
+// the end. It satisfies the filer upload loop's ChunkBoundaries interface.
+func (c *Cutter) NextChunkSize(offset int64) int64 {
+	if offset < 0 || offset >= c.total {
+		return 0
+	}
+	i := sort.Search(len(c.starts), func(i int) bool { return c.starts[i] > offset }) - 1
+	end := c.total
+	if i+1 < len(c.starts) {
+		end = c.starts[i+1]
+	}
+	remaining := end - offset
+	if c.quantum > 0 {
+		if step := c.quantum - (offset-c.starts[i])%c.quantum; step < remaining {
+			return step
+		}
+	}
+	return remaining
+}

weed/format/layout_test.go

@@ -0,0 +1,192 @@
+package format
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestLayoutEncodeDecodeRoundTrip(t *testing.T) {
+	layout := &Layout{
+		Format:      "hls-ts",
+		ExtentSizes: []int64{188 * 3, 188 * 2, 188 * 7},
+		Align:       188,
+		Payload:     []byte{1, 2, 3},
+	}
+	encoded, err := layout.Encode()
+	if err != nil {
+		t.Fatalf("Encode() error = %v", err)
+	}
+	decoded, err := DecodeLayout(encoded)
+	if err != nil {
+		t.Fatalf("DecodeLayout() error = %v", err)
+	}
+	if decoded.Format != layout.Format || decoded.Align != layout.Align {
+		t.Fatalf("decoded = %+v, want %+v", decoded, layout)
+	}
+	if len(decoded.ExtentSizes) != len(layout.ExtentSizes) {
+		t.Fatalf("extent count = %d, want %d", len(decoded.ExtentSizes), len(layout.ExtentSizes))
+	}
+	for i := range layout.ExtentSizes {
+		if decoded.ExtentSizes[i] != layout.ExtentSizes[i] {
+			t.Fatalf("extent %d = %d, want %d", i, decoded.ExtentSizes[i], layout.ExtentSizes[i])
+		}
+	}
+	if string(decoded.Payload) != string(layout.Payload) {
+		t.Fatalf("payload = %v, want %v", decoded.Payload, layout.Payload)
+	}
+}
+
+func TestDecodeLayoutRejectsCorruptInput(t *testing.T) {
+	layout := &Layout{Format: "parquet", ExtentSizes: []int64{10, 20}, Align: 1}
+	encoded, err := layout.Encode()
+	if err != nil {
+		t.Fatalf("Encode() error = %v", err)
+	}
+	for cut := 0; cut < len(encoded); cut++ {
+		if _, err := DecodeLayout(encoded[:cut]); err == nil {
+			t.Fatalf("DecodeLayout() accepted truncation at %d", cut)
+		}
+	}
+	if _, err := DecodeLayout(append(append([]byte{}, encoded...), 0)); err == nil {
+		t.Fatalf("DecodeLayout() accepted trailing bytes")
+	}
+}
+
+func TestLayoutValidate(t *testing.T) {
+	tests := []struct {
+		name     string
+		layout   Layout
+		fileSize int64
+		wantErr  string
+	}{
+		{"valid", Layout{Format: "x", ExtentSizes: []int64{5, 5}, Align: 1}, 10, ""},
+		{"skip size check", Layout{Format: "x", ExtentSizes: []int64{5}, Align: 1}, -1, ""},
+		{"wrong total", Layout{Format: "x", ExtentSizes: []int64{5, 5}, Align: 1}, 11, "but the file has"},
+		{"zero extent", Layout{Format: "x", ExtentSizes: []int64{5, 0}, Align: 1}, -1, "invalid size"},
+		{"no extents", Layout{Format: "x", Align: 1}, -1, "no extents"},
+		{"bad align", Layout{Format: "x", ExtentSizes: []int64{5}, Align: 0}, -1, "align"},
+		{"no name", Layout{ExtentSizes: []int64{5}, Align: 1}, -1, "format name"},
+		{"name too long", Layout{Format: strings.Repeat("x", MaxFormatNameBytes+1), ExtentSizes: []int64{5}, Align: 1}, -1, "too long"},
+	}
+	for _, test := range tests {
+		err := test.layout.Validate(test.fileSize)
+		if test.wantErr == "" {
+			if err != nil {
+				t.Fatalf("%s: Validate() error = %v", test.name, err)
+			}
+			continue
+		}
+		if err == nil || !strings.Contains(err.Error(), test.wantErr) {
+			t.Fatalf("%s: Validate() error = %v, want %q", test.name, err, test.wantErr)
+		}
+	}
+}
+
+func TestExtentRange(t *testing.T) {
+	layout := &Layout{Format: "x", ExtentSizes: []int64{10, 20, 30}, Align: 1}
+	offset, size, ok := layout.ExtentRange(1)
+	if !ok || offset != 10 || size != 20 {
+		t.Fatalf("ExtentRange(1) = (%d, %d, %v), want (10, 20, true)", offset, size, ok)
+	}
+	if _, _, ok := layout.ExtentRange(3); ok {
+		t.Fatalf("ExtentRange(3) accepted out-of-range index")
+	}
+	if _, _, ok := layout.ExtentRange(-1); ok {
+		t.Fatalf("ExtentRange(-1) accepted negative index")
+	}
+}
+
+// collectChunks walks the cutter the way the upload loop does.
+func collectChunks(t *testing.T, cutter *Cutter) [][2]int64 {
+	t.Helper()
+	var chunks [][2]int64
+	var offset int64
+	for {
+		size := cutter.NextChunkSize(offset)
+		if size <= 0 {
+			return chunks
+		}
+		chunks = append(chunks, [2]int64{offset, size})
+		offset += size
+	}
+}
+
+func TestCutterKeepsExtentBoundaries(t *testing.T) {
+	layout := &Layout{Format: "x", ExtentSizes: []int64{5, 4}, Align: 1}
+	chunks := collectChunks(t, layout.Cutter(16))
+	want := [][2]int64{{0, 5}, {5, 4}}
+	if len(chunks) != len(want) {
+		t.Fatalf("chunks = %v, want %v", chunks, want)
+	}
+	for i := range want {
+		if chunks[i] != want[i] {
+			t.Fatalf("chunk %d = %v, want %v", i, chunks[i], want[i])
+		}
+	}
+}
+
+func TestCutterSplitsOversizedExtentsOnAlign(t *testing.T) {
+	// maxChunkSize 5 with align 2 quantizes down to 4-byte interior cuts.
+	layout := &Layout{Format: "x", ExtentSizes: []int64{10, 3}, Align: 2}
+	chunks := collectChunks(t, layout.Cutter(5))
+	want := [][2]int64{{0, 4}, {4, 4}, {8, 2}, {10, 3}}
+	if len(chunks) != len(want) {
+		t.Fatalf("chunks = %v, want %v", chunks, want)
+	}
+	for i := range want {
+		if chunks[i] != want[i] {
+			t.Fatalf("chunk %d = %v, want %v", i, chunks[i], want[i])
+		}
+	}
+}
+
+func TestCutterAlignLargerThanChunkLimit(t *testing.T) {
+	// Align above maxChunkSize still cuts on whole atoms.
+	layout := &Layout{Format: "x", ExtentSizes: []int64{20}, Align: 8}
+	chunks := collectChunks(t, layout.Cutter(5))
+	want := [][2]int64{{0, 8}, {8, 8}, {16, 4}}
+	if len(chunks) != len(want) {
+		t.Fatalf("chunks = %v, want %v", chunks, want)
+	}
+	for i := range want {
+		if chunks[i] != want[i] {
+			t.Fatalf("chunk %d = %v, want %v", i, chunks[i], want[i])
+		}
+	}
+}
+
+func TestCutterUnlimitedKeepsOneChunkPerExtent(t *testing.T) {
+	layout := &Layout{Format: "x", ExtentSizes: []int64{10, 3}, Align: 188}
+	chunks := collectChunks(t, layout.Cutter(0))
+	want := [][2]int64{{0, 10}, {10, 3}}
+	if len(chunks) != len(want) {
+		t.Fatalf("chunks = %v, want %v", chunks, want)
+	}
+}
+
+// A hostile layout may declare an enormous extent; the cutter must stay O(1)
+// per query instead of materializing every interior cut.
+func TestCutterHugeExtentStaysLazy(t *testing.T) {
+	const quantum = 4 << 20 // 4MiB, already a multiple of align 1
+	layout := &Layout{Format: "x", ExtentSizes: []int64{1 << 50, 188}, Align: 188}
+	cutter := layout.Cutter(quantum)
+	alignedQuantum := int64(quantum - quantum%188)
+	if got := cutter.NextChunkSize(0); got != alignedQuantum {
+		t.Fatalf("NextChunkSize(0) = %d, want %d", got, alignedQuantum)
+	}
+	if got := cutter.NextChunkSize(alignedQuantum * 1000); got != alignedQuantum {
+		t.Fatalf("mid-extent chunk = %d, want %d", got, alignedQuantum)
+	}
+	// the final interior chunk stops at the extent boundary
+	last := (int64(1<<50) / alignedQuantum) * alignedQuantum
+	if got := cutter.NextChunkSize(last); got != int64(1<<50)-last {
+		t.Fatalf("tail chunk = %d, want %d", got, int64(1<<50)-last)
+	}
+	// the next extent still cuts independently
+	if got := cutter.NextChunkSize(1 << 50); got != 188 {
+		t.Fatalf("second extent chunk = %d, want 188", got)
+	}
+	if got := cutter.NextChunkSize(1<<50 + 188); got != 0 {
+		t.Fatalf("past end = %d, want 0", got)
+	}
+}

weed/format/parquet/parquet.go

@@ -0,0 +1,98 @@
+// Package parquet adapts parquet files: extents are cut at row-group starts,
+// with one trailing extent for the page indexes and footer. Readers that fetch
+// row groups by offset then hit exactly the covering chunks. No view is
+// needed; the whole benefit is delivered by alignment.
+package parquet
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"math"
+
+	parquetgo "github.com/parquet-go/parquet-go"
+	"github.com/seaweedfs/seaweedfs/weed/format"
+)
+
+const FormatName = "parquet"
+
+var magic = []byte("PAR1")
+
+func init() {
+	format.Register(Adapter{})
+}
+
+type Adapter struct{}
+
+var (
+	_ format.Sniffer = Adapter{}
+	_ format.Indexer = Adapter{}
+)
+
+func (Adapter) Name() string { return FormatName }
+
+func (Adapter) Sniff(h format.Hint) bool {
+	return bytes.HasPrefix(h.Head, magic) && bytes.HasSuffix(h.Tail, magic)
+}
+
+// Index reads the footer and cuts one extent per row group. The leading magic
+// rides with the first row group; everything after the last row group (page
+// indexes, footer) forms the final extent.
+func (Adapter) Index(ctx context.Context, r io.ReaderAt, size int64) (*format.Layout, error) {
+	file, err := parquetgo.OpenFile(r, size, parquetgo.SkipPageIndex(true), parquetgo.SkipBloomFilters(true))
+	if err != nil {
+		return nil, fmt.Errorf("open parquet: %w", err)
+	}
+	rowGroups := file.Metadata().RowGroups
+	if len(rowGroups) == 0 {
+		return nil, fmt.Errorf("parquet file has no row groups")
+	}
+	if len(rowGroups) >= format.MaxExtentCount {
+		return nil, fmt.Errorf("parquet file has too many row groups: %d", len(rowGroups))
+	}
+
+	starts := make([]int64, 0, len(rowGroups))
+	var lastEnd int64
+	for i, rowGroup := range rowGroups {
+		start, end := int64(math.MaxInt64), int64(0)
+		for _, column := range rowGroup.Columns {
+			columnStart := column.MetaData.DataPageOffset
+			if dictionary := column.MetaData.DictionaryPageOffset; dictionary > 0 && dictionary < columnStart {
+				columnStart = dictionary
+			}
+			if columnStart < start {
+				start = columnStart
+			}
+			if columnEnd := columnStart + column.MetaData.TotalCompressedSize; columnEnd > end {
+				end = columnEnd
+			}
+		}
+		if len(rowGroup.Columns) == 0 || start >= end || start < int64(len(magic)) || end > size {
+			return nil, fmt.Errorf("row group %d has an invalid byte range [%d, %d)", i, start, end)
+		}
+		if i > 0 && start < lastEnd {
+			return nil, fmt.Errorf("row group %d overlaps its predecessor", i)
+		}
+		starts = append(starts, start)
+		lastEnd = end
+	}
+	if lastEnd >= size {
+		return nil, fmt.Errorf("row groups leave no room for the footer")
+	}
+
+	var sizes []int64
+	var previous int64
+	for _, start := range starts[1:] {
+		sizes = append(sizes, start-previous)
+		previous = start
+	}
+	sizes = append(sizes, lastEnd-previous)
+	sizes = append(sizes, size-lastEnd)
+
+	layout := &format.Layout{Format: FormatName, ExtentSizes: sizes, Align: 1}
+	if err := layout.Validate(size); err != nil {
+		return nil, err
+	}
+	return layout, nil
+}

weed/format/parquet/parquet_test.go

@@ -0,0 +1,108 @@
+package parquet
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	parquetgo "github.com/parquet-go/parquet-go"
+	"github.com/seaweedfs/seaweedfs/weed/format"
+	"github.com/seaweedfs/seaweedfs/weed/format/formattest"
+)
+
+type row struct {
+	ID   int64  `parquet:"id"`
+	Name string `parquet:"name"`
+}
+
+// buildParquet writes rowGroups row groups of rowsPerGroup rows each.
+func buildParquet(t *testing.T, rowGroups, rowsPerGroup int) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	writer := parquetgo.NewGenericWriter[row](&buf)
+	for g := 0; g < rowGroups; g++ {
+		rows := make([]row, rowsPerGroup)
+		for i := range rows {
+			rows[i] = row{ID: int64(g*rowsPerGroup + i), Name: "some filler content to give row groups a little size"}
+		}
+		if _, err := writer.Write(rows); err != nil {
+			t.Fatalf("Write() error = %v", err)
+		}
+		if err := writer.Flush(); err != nil {
+			t.Fatalf("Flush() error = %v", err)
+		}
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatalf("Close() error = %v", err)
+	}
+	return buf.Bytes()
+}
+
+func TestIndex(t *testing.T) {
+	data := buildParquet(t, 3, 100)
+	size := int64(len(data))
+	layout, err := Adapter{}.Index(context.Background(), bytes.NewReader(data), size)
+	if err != nil {
+		t.Fatalf("Index() error = %v", err)
+	}
+	// one extent per row group plus the trailing footer extent
+	if len(layout.ExtentSizes) != 4 {
+		t.Fatalf("extents = %v, want 4", layout.ExtentSizes)
+	}
+	if err := layout.Validate(size); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+	formattest.EncodeRoundTrip(t, layout)
+
+	// extent cuts must land on the row-group starts the footer declares
+	file, err := parquetgo.OpenFile(bytes.NewReader(data), size)
+	if err != nil {
+		t.Fatalf("OpenFile() error = %v", err)
+	}
+	var offset int64
+	for i, extentSize := range layout.ExtentSizes[:3] {
+		if i > 0 {
+			want := file.Metadata().RowGroups[i].Columns[0].MetaData.DataPageOffset
+			if dictionary := file.Metadata().RowGroups[i].Columns[0].MetaData.DictionaryPageOffset; dictionary > 0 && dictionary < want {
+				want = dictionary
+			}
+			if offset != want {
+				t.Fatalf("extent %d starts at %d, row group starts at %d", i, offset, want)
+			}
+		}
+		offset += extentSize
+	}
+}
+
+func TestIndexSingleRowGroup(t *testing.T) {
+	data := buildParquet(t, 1, 10)
+	layout, err := Adapter{}.Index(context.Background(), bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("Index() error = %v", err)
+	}
+	if len(layout.ExtentSizes) != 2 {
+		t.Fatalf("extents = %v, want 2", layout.ExtentSizes)
+	}
+}
+
+func TestIndexRejectsNonParquet(t *testing.T) {
+	data := []byte("this is not a parquet file, not even close, but long enough")
+	if _, err := (Adapter{}).Index(context.Background(), bytes.NewReader(data), int64(len(data))); err == nil {
+		t.Fatalf("Index() accepted junk")
+	}
+}
+
+func TestIndexTruncations(t *testing.T) {
+	formattest.IndexTruncations(t, Adapter{}, buildParquet(t, 2, 50))
+}
+
+func TestSniff(t *testing.T) {
+	data := buildParquet(t, 1, 10)
+	head, tail := data[:4], data[len(data)-4:]
+	if !(Adapter{}).Sniff(format.Hint{Head: head, Tail: tail}) {
+		t.Fatalf("Sniff() rejected parquet magic")
+	}
+	if (Adapter{}).Sniff(format.Hint{Head: []byte("PAR1"), Tail: []byte("nope")}) {
+		t.Fatalf("Sniff() accepted missing tail magic")
+	}
+}

weed/format/registry.go

@@ -0,0 +1,16 @@
+package format
+
+// Adapters register from init(), so the map needs no locking.
+var formats = map[string]Format{}
+
+func Register(f Format) {
+	if _, ok := formats[f.Name()]; ok {
+		panic("format: duplicate adapter " + f.Name())
+	}
+	formats[f.Name()] = f
+}
+
+// ByName returns the registered adapter, or nil.
+func ByName(name string) Format {
+	return formats[name]
+}

weed/server/filer_server_format.go

@@ -0,0 +1,603 @@
+package weed_server
+
+import (
+	"bytes"
+	"context"
+	"crypto/md5"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"os"
+	"path"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/format"
+	_ "github.com/seaweedfs/seaweedfs/weed/format/hlsts"
+	_ "github.com/seaweedfs/seaweedfs/weed/format/parquet"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
+)
+
+const (
+	// Query parameters follow the mv.from/cp.from dotted convention so the
+	// general POST endpoint cannot collide with pass-through client params.
+	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.
+	// Extent chunks are buffered in memory, so the limit must never be absent.
+	defaultFormatChunkSizeMB = 4
+)
+
+// formatChunkIdentity digests the chunk list a layout was written against,
+// covering every field that changes what a read returns: a FUSE truncate, for
+// one, mutates Size while keeping the chunk's id.
+func formatChunkIdentity(chunks []*filer_pb.FileChunk) []byte {
+	digest := md5.New()
+	for _, chunk := range chunks {
+		fmt.Fprintf(digest, "%d:%s:%d:%d:%x:%t:%t:%d;",
+			chunk.Offset, chunk.GetFileIdString(), chunk.Size, chunk.ModifiedTsNs,
+			chunk.CipherKey, chunk.IsCompressed, chunk.IsChunkManifest, chunk.SseType)
+	}
+	return digest.Sum(nil)
+}
+
+// repackSourceIdentity digests every entry field that influenced a repack's
+// output: the chunk list it read, the size the layout was validated against,
+// the TTL and expiry anchors its new chunks were assigned with, and the
+// hard-link and remote state its guards evaluated.
+func repackSourceIdentity(entry *filer.Entry) []byte {
+	digest := md5.New()
+	digest.Write(formatChunkIdentity(entry.GetChunks()))
+	digest.Write(entry.Content)
+	fmt.Fprintf(digest, "%d:%d:%d:%d:%d:%t:%x:%t",
+		len(entry.Content), entry.FileSize, entry.TtlSec, entry.Crtime.UnixNano(), entry.Mtime.UnixNano(),
+		entry.IsExpireS3Enabled(), []byte(entry.HardLinkId), entry.Remote != nil)
+	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)
+		}
+	}
+	// Nothing above ~68 years rounds up within int32; no volume TTL keeps the
+	// chunks past the entry, which is the safe direction.
+	return 0
+}
+
+// formatChunkSizeLimit mirrors the autoChunk maxMB resolution.
+func (fs *FilerServer) formatChunkSizeLimit(r *http.Request) int64 {
+	parsedMaxMB, _ := strconv.ParseInt(r.URL.Query().Get("maxMB"), 10, 32)
+	maxMB := int32(parsedMaxMB)
+	if maxMB <= 0 && fs.option.MaxMB > 0 {
+		maxMB = int32(fs.option.MaxMB)
+	}
+	if maxMB <= 0 {
+		maxMB = defaultFormatChunkSizeMB
+	}
+	return int64(maxMB) * 1024 * 1024
+}
+
+// copyStandardHeadersToExtended matches what saveMetaData keeps on an entry.
+func copyStandardHeadersToExtended(r *http.Request, extended map[string][]byte) {
+	for k, v := range r.Header {
+		if len(v) > 0 && len(v[0]) > 0 {
+			if strings.HasPrefix(k, needle.PairNamePrefix) || k == "Cache-Control" || k == "Expires" || k == "Content-Disposition" {
+				extended[k] = []byte(v[0])
+			}
+			if k == "Response-Content-Disposition" {
+				extended["Content-Disposition"] = []byte(v[0])
+			}
+		}
+	}
+}
+
+// formatIngest handles POST /path?format.ingest=<adapter>: a multipart body
+// with an "index" sidecar part describing the media's extents, then the
+// "media" bytes. Storage chunks are cut on the boundaries the sidecar declares.
+func (fs *FilerServer) formatIngest(ctx context.Context, w http.ResponseWriter, r *http.Request, so *operation.StorageOption) {
+	adapterName := r.URL.Query().Get(formatIngestParam)
+	adapter := format.ByName(adapterName)
+	if adapter == nil {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("unknown format %q", adapterName))
+		return
+	}
+	sidecarIndexer, ok := adapter.(format.SidecarIndexer)
+	if !ok {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("format %q does not support sidecar ingest", adapterName))
+		return
+	}
+	if strings.HasSuffix(r.URL.Path, "/") {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("format ingest target must be a file path"))
+		return
+	}
+	if enforced, err := fs.wormEnforcedForEntry(ctx, r.URL.Path); err != nil {
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	} else if enforced {
+		writeJsonError(w, r, http.StatusForbidden, errors.New("cannot replace WORM-enforced entry"))
+		return
+	}
+
+	multipartReader, err := r.MultipartReader()
+	if err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("format ingest requires multipart/form-data: %w", err))
+		return
+	}
+	sidecarPart, err := multipartReader.NextPart()
+	if err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("read index part: %w", err))
+		return
+	}
+	if sidecarPart.FormName() != "index" {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("first multipart part must be named index"))
+		return
+	}
+	sidecar, err := io.ReadAll(io.LimitReader(sidecarPart, maxFormatSidecarBytes+1))
+	sidecarPart.Close()
+	if err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("read index part: %w", err))
+		return
+	}
+	if len(sidecar) > maxFormatSidecarBytes {
+		writeJsonError(w, r, http.StatusRequestEntityTooLarge, fmt.Errorf("index part exceeds %d bytes", maxFormatSidecarBytes))
+		return
+	}
+	layout, err := sidecarIndexer.IndexSidecar(sidecar)
+	if err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, err)
+		return
+	}
+	if err := layout.Validate(-1); err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, err)
+		return
+	}
+
+	mediaPart, err := multipartReader.NextPart()
+	if err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("read media part: %w", err))
+		return
+	}
+	if mediaPart.FormName() != "media" {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("second multipart part must be named media"))
+		return
+	}
+	defer mediaPart.Close()
+	contentType := mediaPart.Header.Get("Content-Type")
+	if contentType == "" {
+		contentType = "application/octet-stream"
+	}
+
+	cutter := layout.Cutter(fs.formatChunkSizeLimit(r))
+	fileChunks, md5Hash, written, uploadErr, _ := fs.uploadReaderToBoundedChunks(ctx, r, mediaPart, 0, cutter, path.Base(r.URL.Path), contentType, false, so)
+	cleanup := func() { fs.filer.DeleteUncommittedChunks(context.WithoutCancel(ctx), fileChunks) }
+	if uploadErr != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, uploadErr)
+		return
+	}
+	if total := layout.TotalSize(); written != total {
+		cleanup()
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("media is %d bytes but the index describes %d", written, total))
+		return
+	}
+	var extra [1]byte
+	if n, _ := io.ReadFull(mediaPart, extra[:]); n != 0 {
+		cleanup()
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("media has trailing bytes beyond the index"))
+		return
+	}
+	if extraPart, nextErr := multipartReader.NextPart(); nextErr == nil {
+		extraPart.Close()
+		cleanup()
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("unexpected multipart part after media"))
+		return
+	}
+
+	fileChunks, err = filer.MaybeManifestize(fs.saveAsChunk(ctx, so), fileChunks)
+	if err != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+	encoded, err := layout.Encode()
+	if err != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+
+	mode := uint64(0660)
+	if text := r.URL.Query().Get("mode"); text != "" {
+		if mode, err = strconv.ParseUint(text, 8, 32); err != nil {
+			cleanup()
+			writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("invalid mode %q", text))
+			return
+		}
+	}
+	now := time.Now()
+	entry := &filer.Entry{
+		FullPath: util.FullPath(r.URL.Path),
+		Attr: filer.Attr{
+			Mtime: now, Crtime: now,
+			Mode: os.FileMode(mode), Uid: OS_UID, Gid: OS_GID,
+			TtlSec: so.TtlSeconds, Mime: contentType,
+			Md5: md5Hash.Sum(nil), FileSize: uint64(written),
+		},
+		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
+	// serialize with gRPC writers, renames, and repack
+	pathLock := fs.entryLockTable.AcquireLock("formatIngest", entry.FullPath, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(entry.FullPath, pathLock)
+	// recheck under the lock: WORM may have been enabled during the upload
+	if enforced, wormErr := fs.wormEnforcedForEntry(ctx, r.URL.Path); wormErr != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, wormErr)
+		return
+	} else if enforced {
+		cleanup()
+		writeJsonError(w, r, http.StatusForbidden, errors.New("cannot replace WORM-enforced entry"))
+		return
+	}
+	if err := fs.filer.CreateEntry(context.WithoutCancel(ctx), entry, nil, false, false, nil, skipCheckParentDirEntry(r), so.MaxFileNameLength); err != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+	writeJsonQuiet(w, r, http.StatusCreated, FilerPostResult{Name: entry.Name(), Size: written})
+}
+
+// formatRepack handles POST /path?format.repack=<adapter>: it derives the
+// layout from the stored bytes and rewrites the entry's chunks cut on extent
+// boundaries. The bytes do not change, only where they are cut.
+func (fs *FilerServer) formatRepack(ctx context.Context, w http.ResponseWriter, r *http.Request, so *operation.StorageOption) {
+	adapterName := r.URL.Query().Get(formatRepackParam)
+	adapter := format.ByName(adapterName)
+	if adapter == nil {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("unknown format %q", adapterName))
+		return
+	}
+	indexer, ok := adapter.(format.Indexer)
+	if !ok {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("format %q does not support repack", adapterName))
+		return
+	}
+	fullPath := util.FullPath(r.URL.Path)
+	// Serializes gRPC writers, renames, and this filer's HTTP overwrites;
+	// cross-filer serialization needs owner routing.
+	pathLock := fs.entryLockTable.AcquireLock("formatRepack", fullPath, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(fullPath, pathLock)
+
+	// checked under the lock so a concurrent WORM enable cannot land between
+	// validation and the entry swap
+	if enforced, err := fs.wormEnforcedForEntry(ctx, r.URL.Path); err != nil {
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	} else if enforced {
+		writeJsonError(w, r, http.StatusForbidden, errors.New("cannot repack WORM-enforced entry"))
+		return
+	}
+
+	entry, err := fs.filer.FindEntry(ctx, fullPath)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			writeJsonError(w, r, http.StatusNotFound, err)
+		} else {
+			writeJsonError(w, r, http.StatusInternalServerError, err)
+		}
+		return
+	}
+	if entry.IsDirectory() {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("cannot repack a directory"))
+		return
+	}
+	oldChunks := entry.GetChunks()
+	sourceIdentity := repackSourceIdentity(entry)
+	if len(oldChunks) == 0 {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("entry has no chunks to repack"))
+		return
+	}
+	if len(entry.HardLinkId) != 0 || entry.Remote != nil {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("cannot repack hard-linked or remote entries"))
+		return
+	}
+	// the repack reader sees only chunks; inline content would be dropped
+	if len(entry.Content) != 0 {
+		writeJsonError(w, r, http.StatusBadRequest, errors.New("cannot repack entries with inline content"))
+		return
+	}
+	for _, chunk := range oldChunks {
+		if chunk.SseType != filer_pb.SSEType_NONE {
+			writeJsonError(w, r, http.StatusBadRequest, errors.New("cannot repack server-side encrypted entries"))
+			return
+		}
+	}
+
+	// Repack rewrites where bytes are cut, never their lifetime: new chunks
+	// carry the entry's remaining TTL, not the request query's nor a restart
+	// of the original span.
+	so.TtlSeconds = entry.TtlSec
+	if entry.TtlSec > 0 {
+		// 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 = roundUpToVolumeTTL(remaining)
+	}
+
+	if entry.FileSize > math.MaxInt64 {
+		writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("file size %d overflows int64", entry.FileSize))
+		return
+	}
+	size := int64(entry.FileSize)
+	lookup := fs.filer.MasterClient.GetLookupFileIdFunction()
+	chunkViews := filer.ViewFromChunks(ctx, lookup, oldChunks, 0, size)
+	readerCache := filer.NewReaderCache(8, chunk_cache.NewChunkCacheInMemory(16), lookup, nil)
+	readerAt := filer.NewChunkReaderAtFromClient(ctx, readerCache, chunkViews, size, filer.DefaultPrefetchCount)
+	// Close releases the private reader cache and its in-flight prefetches.
+	defer readerAt.Close()
+
+	// cheap magic-byte gate before handing the bytes to the parser
+	if sniffer, ok := adapter.(format.Sniffer); ok {
+		hint := format.Hint{Name: entry.Name(), ContentType: entry.Attr.Mime, Size: size}
+		sniffSize := int64(formatSniffBytes)
+		if sniffSize > size {
+			sniffSize = size
+		}
+		head := make([]byte, sniffSize)
+		if _, err := readerAt.ReadAt(head, 0); err != nil && err != io.EOF {
+			writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("read head: %w", err))
+			return
+		}
+		tail := make([]byte, sniffSize)
+		if _, err := readerAt.ReadAt(tail, size-sniffSize); err != nil && err != io.EOF {
+			writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("read tail: %w", err))
+			return
+		}
+		hint.Head, hint.Tail = head, tail
+		if !sniffer.Sniff(hint) {
+			writeJsonError(w, r, http.StatusBadRequest, fmt.Errorf("%s does not look like %s", entry.Name(), adapterName))
+			return
+		}
+	}
+
+	layout, err := indexer.Index(ctx, readerAt, size)
+	if err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, err)
+		return
+	}
+	if err := layout.Validate(size); err != nil {
+		writeJsonError(w, r, http.StatusBadRequest, err)
+		return
+	}
+
+	cutter := layout.Cutter(fs.formatChunkSizeLimit(r))
+	newChunks, md5Hash, written, uploadErr, _ := fs.uploadReaderToBoundedChunks(ctx, r, io.NewSectionReader(readerAt, 0, size), 0, cutter, entry.Name(), entry.Attr.Mime, false, so)
+	cleanup := func() { fs.filer.DeleteUncommittedChunks(context.WithoutCancel(ctx), newChunks) }
+	if uploadErr != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, uploadErr)
+		return
+	}
+	if written != size {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("read %d of %d bytes", written, size))
+		return
+	}
+	newChunks, err = filer.MaybeManifestize(fs.saveAsChunk(ctx, so), newChunks)
+	if err != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+	encoded, err := layout.Encode()
+	if err != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+
+	// The entry lock is filer-local, so a writer on another filer is not
+	// blocked by it. Re-read from the store and conflict on any change to
+	// state that influenced this repack - chunks, size, TTL and expiry
+	// anchors, hard-link and remote state - so the swap can never pair fresh
+	// metadata with chunks built from stale inputs. The unguarded window
+	// shrinks from the whole repack to this commit; within it repack races
+	// like any ordinary writer. Closing it needs owner routing.
+	current, err := fs.filer.FindEntry(ctx, fullPath)
+	if err != nil {
+		cleanup()
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			writeJsonError(w, r, http.StatusConflict, errors.New("entry was deleted during repack"))
+		} else {
+			writeJsonError(w, r, http.StatusInternalServerError, err)
+		}
+		return
+	}
+	if !bytes.Equal(repackSourceIdentity(current), sourceIdentity) {
+		cleanup()
+		writeJsonError(w, r, http.StatusConflict, errors.New("entry changed during repack"))
+		return
+	}
+	if enforced, wormErr := fs.wormEnforcedForEntry(ctx, r.URL.Path); wormErr != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, wormErr)
+		return
+	} else if enforced {
+		cleanup()
+		writeJsonError(w, r, http.StatusForbidden, errors.New("cannot repack WORM-enforced entry"))
+		return
+	}
+
+	// build from the fresh read so a concurrent metadata-only update on
+	// another filer is carried forward, not clobbered
+	newEntry := *current
+	newEntry.Chunks = newChunks
+	newEntry.Extended = make(map[string][]byte)
+	for k, v := range current.Extended {
+		newEntry.Extended[k] = v
+	}
+	newEntry.Extended[format.LayoutKey] = encoded
+	newEntry.Extended[formatLayoutChunksKey] = formatChunkIdentity(newChunks)
+	if len(newEntry.Md5) == 0 {
+		newEntry.Md5 = md5Hash.Sum(nil)
+	}
+	if err := fs.filer.UpdateEntry(context.WithoutCancel(ctx), current, &newEntry); err != nil {
+		cleanup()
+		writeJsonError(w, r, http.StatusInternalServerError, err)
+		return
+	}
+	fs.filer.DeleteChunks(context.WithoutCancel(ctx), fullPath, current.GetChunks())
+	// Filer.UpdateEntry only writes the store; notify subscribers (sync,
+	// backup, replication) of the new chunk ids like the gRPC path does.
+	fs.filer.NotifyUpdateEvent(ctx, current, &newEntry, true, false, nil)
+	writeJsonQuiet(w, r, http.StatusOK, map[string]interface{}{
+		"name": entry.Name(), "size": size, "extents": len(layout.ExtentSizes),
+	})
+}
+
+// serveFormatView answers GET/HEAD /path?view=<adapter> for entries carrying a
+// layout. The layout is advisory: any inconsistency yields 404 here while the
+// plain read path stays untouched.
+func (fs *FilerServer) serveFormatView(ctx context.Context, w http.ResponseWriter, r *http.Request, entry *filer.Entry, viewName string) {
+	adapter := format.ByName(viewName)
+	viewer, viewerOk := adapter.(format.Viewer)
+	if adapter == nil || !viewerOk {
+		http.Error(w, "no such view", http.StatusNotFound)
+		return
+	}
+	encoded := entry.Extended[format.LayoutKey]
+	if len(encoded) == 0 {
+		http.Error(w, "entry has no format layout", http.StatusNotFound)
+		return
+	}
+	layout, err := format.DecodeLayout(encoded)
+	if err != nil || layout.Format != viewName {
+		http.Error(w, "entry has no such format layout", http.StatusNotFound)
+		return
+	}
+	// a size beyond int64 would read as negative, which Validate treats as
+	// "skip the size check"
+	if entry.FileSize > math.MaxInt64 {
+		http.Error(w, "format layout is stale", http.StatusNotFound)
+		return
+	}
+	if err := layout.Validate(int64(entry.FileSize)); err != nil {
+		glog.WarningfCtx(ctx, "stale format layout on %s: %v", entry.FullPath, err)
+		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. Inline
+	// content is disqualifying outright: format entries are never written
+	// with it, and reads would prefer it over the chunks the layout maps.
+	if len(entry.Content) != 0 || !bytes.Equal(entry.Extended[formatLayoutChunksKey], formatChunkIdentity(entry.GetChunks())) {
+		glog.WarningfCtx(ctx, "format layout on %s no longer matches its bytes", 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:
+	// re-ingesting with a different sidecar must invalidate cached views.
+	viewIdentity := md5.New()
+	viewIdentity.Write([]byte(filer.ETagEntry(entry)))
+	viewIdentity.Write(encoded)
+	viewIdentity.Write([]byte(r.URL.RawQuery))
+	viewEntry := *entry
+	viewEntry.Md5 = viewIdentity.Sum(nil)
+	if checkPreconditions(w, r, &viewEntry) {
+		return
+	}
+
+	plan, err := viewer.View(format.ViewRequest{Query: r.URL.Query()}, format.Object{
+		Name: entry.Name(), Size: int64(entry.FileSize), Layout: layout,
+	})
+	if err != nil {
+		if errors.Is(err, format.ErrNoSuchView) {
+			http.NotFound(w, r)
+		} else {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+		return
+	}
+
+	// pass through stored headers the way plain reads do
+	for k, v := range entry.Extended {
+		if !strings.HasPrefix(k, "xattr-") && !s3_constants.IsSeaweedFSInternalHeader(k) {
+			w.Header().Set(k, string(v))
+		}
+	}
+	// view responses are whole documents or whole extents
+	w.Header().Set("Accept-Ranges", "none")
+	w.Header().Set("Content-Type", plan.ContentType)
+	SetEtag(w, filer.ETagEntry(&viewEntry))
+
+	if plan.Body != nil {
+		w.Header().Set("Content-Length", strconv.Itoa(len(plan.Body)))
+		if r.Method == http.MethodHead {
+			return
+		}
+		if _, err := w.Write(plan.Body); err != nil {
+			glog.V(2).InfofCtx(ctx, "write %s view of %s: %v", viewName, entry.FullPath, err)
+		}
+		return
+	}
+
+	offset, extentSize, ok := layout.ExtentRange(plan.Extent)
+	if !ok {
+		http.NotFound(w, r)
+		return
+	}
+	w.Header().Set("Content-Length", strconv.FormatInt(extentSize, 10))
+	if r.Method == http.MethodHead {
+		return
+	}
+	streamCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	streamFn, err := filer.PrepareStreamContentWithPrefetch(streamCtx, fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, entry.GetChunks(), offset, extentSize, fs.option.DownloadMaxBytesPs, filer.DefaultPrefetchCount)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if err := streamFn(w); err != nil {
+		glog.ErrorfCtx(ctx, "stream %s view extent %d of %s: %v", viewName, plan.Extent, entry.FullPath, err)
+	}
+}

weed/server/filer_server_format_test.go

@@ -0,0 +1,129 @@
+package weed_server
+
+import (
+	"bytes"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"math"
+	"testing"
+	"time"
+
+	"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), 0},     // beyond every unit's 255 cap: no TTL, never a shortened one
+	}
+	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 got == 0 {
+			continue // no volume TTL: chunks outlive the entry
+		}
+		if int64(got) < test.seconds {
+			t.Fatalf("roundUpToVolumeTTL(%d) = %d shortened the lifetime", test.seconds, got)
+		}
+		// the rounded value must survive the volume TTL string conversion intact
+		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 TestRepackSourceIdentity(t *testing.T) {
+	base := func() *filer.Entry {
+		return &filer.Entry{
+			FullPath: "/videos/movie.ts",
+			Attr: filer.Attr{
+				FileSize: 30, TtlSec: 600,
+				Crtime: time.Unix(1000, 0), Mtime: time.Unix(2000, 0),
+			},
+			Chunks: []*filer_pb.FileChunk{{FileId: "1,ab", Offset: 0, Size: 30}},
+		}
+	}
+	identity := repackSourceIdentity(base())
+	if !bytes.Equal(identity, repackSourceIdentity(base())) {
+		t.Fatalf("identity is not deterministic")
+	}
+	mutations := map[string]func(*filer.Entry){
+		"ttl cleared":  func(e *filer.Entry) { e.TtlSec = 0 },
+		"size changed": func(e *filer.Entry) { e.FileSize = 31 },
+		"mtime moved":  func(e *filer.Entry) { e.Mtime = time.Unix(3000, 0) },
+		"crtime moved": func(e *filer.Entry) { e.Crtime = time.Unix(1001, 0) },
+		"hard linked":  func(e *filer.Entry) { e.HardLinkId = []byte{1} },
+		"went remote":  func(e *filer.Entry) { e.Remote = &filer_pb.RemoteEntry{} },
+		"chunk moved":  func(e *filer.Entry) { e.Chunks[0].Offset = 1 },
+		"content set":  func(e *filer.Entry) { e.Content = []byte{0x47} },
+	}
+	for name, mutate := range mutations {
+		changed := base()
+		mutate(changed)
+		if bytes.Equal(identity, repackSourceIdentity(changed)) {
+			t.Fatalf("identity ignored: %s", name)
+		}
+	}
+}
+
+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")
+	}
+	// a truncate mutates Size while keeping the chunk id
+	changedSize := []*filer_pb.FileChunk{
+		{FileId: "1,ab", Offset: 0, Size: 10},
+		{FileId: "2,cd", Offset: 10, Size: 15},
+	}
+	if bytes.Equal(identity, formatChunkIdentity(changedSize)) {
+		t.Fatalf("identity ignored a size change")
+	}
+	changedMtime := []*filer_pb.FileChunk{
+		{FileId: "1,ab", Offset: 0, Size: 10},
+		{FileId: "2,cd", Offset: 10, Size: 20, ModifiedTsNs: 7},
+	}
+	if bytes.Equal(identity, formatChunkIdentity(changedMtime)) {
+		t.Fatalf("identity ignored a modification timestamp change")
+	}
+	changedManifest := []*filer_pb.FileChunk{
+		{FileId: "1,ab", Offset: 0, Size: 10},
+		{FileId: "2,cd", Offset: 10, Size: 20, IsChunkManifest: true},
+	}
+	if bytes.Equal(identity, formatChunkIdentity(changedManifest)) {
+		t.Fatalf("identity ignored a manifest flag change")
+	}
+}

weed/server/filer_server_handlers_read.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/format"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
@@ -139,6 +140,11 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	if viewName := query.Get(format.ViewParam); viewName != "" {
+		fs.serveFormatView(ctx, w, r, entry, viewName)
+		return
+	}
+
 	if checkPreconditions(w, r, entry) {
 		return
 	}

weed/server/filer_server_handlers_write.go

@@ -136,6 +136,14 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request, conte
 		fs.move(ctx, w, r, so)
 	} else if query.Has("cp.from") {
 		fs.copy(ctx, w, r, so)
+	} else if query.Get(formatIngestParam) != "" {
+		if query.Get(formatRepackParam) != "" {
+			writeJsonError(w, r, http.StatusBadRequest, errors.New(formatIngestParam+" and "+formatRepackParam+" are mutually exclusive"))
+			return
+		}
+		fs.formatIngest(ctx, w, r, so)
+	} else if query.Get(formatRepackParam) != "" {
+		fs.formatRepack(ctx, w, r, so)
 	} else {
 		fs.autoChunk(ctx, w, r, contentLength, so)
 	}

weed/server/filer_server_handlers_write_autochunk.go

@@ -246,6 +246,20 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
 	// fix the path
 	path := fs.fixFilePath(ctx, r, fileName)
 
+	// Commit under the entry lock so plain HTTP overwrites serialize with
+	// gRPC writers, renames, and format repack on this filer; cross-filer
+	// serialization needs owner routing.
+	fullPath := util.FullPath(path)
+	pathLock := fs.entryLockTable.AcquireLock("saveMetaData", fullPath, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(fullPath, pathLock)
+
+	// recheck under the lock: WORM may have been enabled during the upload
+	if enforced, wormErr := fs.wormEnforcedForEntry(ctx, path); wormErr != nil {
+		return nil, wormErr
+	} else if enforced {
+		return nil, errors.New(constants.ErrMsgOperationNotPermitted)
+	}
+
 	var entry *filer.Entry
 	var newChunks []*filer_pb.FileChunk
 	var mergedChunks []*filer_pb.FileChunk

weed/server/filer_server_handlers_write_upload.go

@@ -29,6 +29,18 @@ var bufPool = sync.Pool{
 	},
 }
 
+// ChunkBoundaries decides where the upload loop may cut a storage chunk.
+type ChunkBoundaries interface {
+	// NextChunkSize returns the size of the chunk starting at the absolute
+	// file offset (startOffset included), or 0 when no further chunks are
+	// expected. Returned sizes are bounded: they size in-memory buffers.
+	NextChunkSize(offset int64) int64
+}
+
+type fixedChunkSize int64
+
+func (c fixedChunkSize) NextChunkSize(int64) int64 { return int64(c) }
+
 func (fs *FilerServer) uploadRequestToChunks(ctx context.Context, w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
 	query := r.URL.Query()
 
@@ -51,7 +63,15 @@ func (fs *FilerServer) uploadRequestToChunks(ctx context.Context, w http.Respons
 }
 
 func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request, reader io.Reader, startOffset int64, chunkSize int32, fileName, contentType string, isAppend bool, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
+	return fs.uploadReaderToBoundedChunks(ctx, r, reader, startOffset, fixedChunkSize(chunkSize), fileName, contentType, isAppend, so)
+}
 
+// uploadReaderToBoundedChunks cuts chunks where boundaries allows instead of
+// at a fixed size. The small-content optimization only applies to fixed-size
+// chunking: it must stay off when chunk boundaries carry meaning.
+func (fs *FilerServer) uploadReaderToBoundedChunks(ctx context.Context, r *http.Request, reader io.Reader, startOffset int64, boundaries ChunkBoundaries, fileName, contentType string, isAppend bool, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
+
+	_, allowInline := boundaries.(fixedChunkSize)
 	md5Hash = md5.New()
 	chunkOffset = startOffset
 	var partReader = io.NopCloser(io.TeeReader(reader, md5Hash))
@@ -63,6 +83,11 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 	var uploadErrLock sync.Mutex
 	for {
 
+		wantSize := boundaries.NextChunkSize(chunkOffset)
+		if wantSize <= 0 {
+			break
+		}
+
 		// need to throttle used byte buffer
 		bytesBufferLimitChan <- struct{}{}
 
@@ -78,7 +103,7 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 
 		bytesBuffer := bufPool.Get().(*bytes.Buffer)
 
-		limitedReader := io.LimitReader(partReader, int64(chunkSize))
+		limitedReader := io.LimitReader(partReader, wantSize)
 
 		bytesBuffer.Reset()
 
@@ -97,7 +122,7 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 			}
 			break
 		}
-		if chunkOffset == 0 && !isAppend {
+		if chunkOffset == 0 && !isAppend && allowInline {
 			if dataSize < fs.option.SaveToFilerLimit {
 				chunkOffset += dataSize
 				smallContent = make([]byte, dataSize)
@@ -141,7 +166,7 @@ func (fs *FilerServer) uploadReaderToChunks(ctx context.Context, r *http.Request
 		chunkOffset = chunkOffset + dataSize
 
 		// if last chunk was not at full chunk size, but already exhausted the reader
-		if dataSize < int64(chunkSize) {
+		if dataSize < wantSize {
 			break
 		}
 	}