Add compressed size prefix to page headers for lz4 v4.1.23 compatibility

The lz4 library v4.1.23 added frame concatenation support, which peeks ahead
after reading a frame to check for another concatenated frame. This broke LTX
because each page is an independent LZ4 frame with a PageHeader in between.

This change adds a new PageHeaderFlagCompressedSize flag and writes a 4-byte
compressed size prefix after each page header. The decoder uses this size to
create an exact LimitedReader, preventing lz4 from peeking into the next page.

For backward compatibility, the decoder handles both formats:
- New format (flag set): reads compressed size, uses exact LimitedReader
- Old format (flag=0): uses LimitedReader workaround with lz4 frame footer size

Fixes #70

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: corylanou

decoder.go

@@ -7,14 +7,21 @@ import (
 	"hash"
 	"hash/crc64"
 	"io"
+	"math"
 
 	"github.com/pierrec/lz4/v4"
 )
 
+// lz4FrameFooterSize is the size of the LZ4 frame footer:
+// EndMark (4 bytes) + Content Checksum (4 bytes).
+// Used when decoding old format files without compressed size prefix.
+const lz4FrameFooterSize = 8
+
 // Decoder represents a decoder of an LTX file.
 type Decoder struct {
-	r  io.Reader   // main reader
-	zr *lz4.Reader // lz4 reader
+	r  io.Reader        // main reader
+	lr io.LimitedReader // limited reader for lz4 (reused)
+	zr *lz4.Reader      // lz4 reader
 
 	header    Header
 	trailer   Trailer
@@ -172,19 +179,53 @@ func (dec *Decoder) DecodePage(hdr *PageHeader, data []byte) error {
 		return err
 	}
 
-	// Read page data next.
-	dec.zr.Reset(dec.r)
-	if _, err := io.ReadFull(dec.zr, data); err != nil {
-		return err
+	// Read page data using format-specific approach.
+	if hdr.Flags&PageHeaderFlagCompressedSize != 0 {
+		// New format: read compressed size prefix, then use exact LimitedReader.
+		sizeBuf := make([]byte, 4)
+		if _, err := io.ReadFull(dec.r, sizeBuf); err != nil {
+			return fmt.Errorf("read compressed size: %w", err)
+		}
+		dec.writeToHash(sizeBuf)
+
+		compressedSize := binary.BigEndian.Uint32(sizeBuf)
+		dec.lr.R = dec.r
+		dec.lr.N = int64(compressedSize)
+		dec.zr.Reset(&dec.lr)
+
+		if _, err := io.ReadFull(dec.zr, data); err != nil {
+			return err
+		}
+
+		// Drain any remaining bytes from the LimitedReader.
+		// The LZ4 reader may not consume all bytes (e.g., frame trailer).
+		if dec.lr.N > 0 {
+			if _, err := io.CopyN(io.Discard, &dec.lr, dec.lr.N); err != nil {
+				return fmt.Errorf("drain lz4 frame: %w", err)
+			}
+		}
+	} else {
+		// Old format: use LimitedReader workaround for lz4 frame concatenation.
+		// The lz4 library peeks ahead after EOF to check for concatenated frames,
+		// so we limit reads to prevent it from reading into the next page header.
+		dec.lr.R = dec.r
+		dec.lr.N = math.MaxInt64
+		dec.zr.Reset(&dec.lr)
+
+		if _, err := io.ReadFull(dec.zr, data); err != nil {
+			return err
+		}
+
+		// Limit remaining reads to the LZ4 frame footer size before checking EOF.
+		dec.lr.N = lz4FrameFooterSize
+		if err := dec.readLZ4Trailer(); err != nil {
+			return fmt.Errorf("read lz4 trailer: %w", err)
+		}
 	}
+
 	dec.writeToHash(data)
 	dec.pageN++
 
-	// Read off the LZ4 trailer frame to ensure we hit EOF.
-	if err := dec.readLZ4Trailer(); err != nil {
-		return fmt.Errorf("read lz4 trailer: %w", err)
-	}
-
 	// Calculate checksum while decoding snapshots if tracking checksums.
 	if dec.header.IsSnapshot() && !dec.header.NoChecksum() {
 		if hdr.Pgno != LockPgno(dec.header.PageSize) {
@@ -301,7 +342,19 @@ func DecodePageData(b []byte) (hdr PageHeader, data []byte, err error) {
 		return hdr, data, nil
 	}
 
-	zr := lz4.NewReader(bytes.NewReader(b[PageHeaderSize:]))
+	// Determine offset and size of LZ4 data based on format.
+	var r io.Reader
+	if hdr.Flags&PageHeaderFlagCompressedSize != 0 {
+		// New format: read compressed size and limit reader to that size.
+		compressedSize := binary.BigEndian.Uint32(b[PageHeaderSize:])
+		offset := PageHeaderSize + 4
+		r = bytes.NewReader(b[offset : offset+int(compressedSize)])
+	} else {
+		// Old format: pass remaining bytes, rely on LZ4 frame boundaries.
+		r = bytes.NewReader(b[PageHeaderSize:])
+	}
+
+	zr := lz4.NewReader(r)
 	data, err = io.ReadAll(zr)
 	return hdr, data, err
 }

decoder_test.go

@@ -47,9 +47,15 @@ func TestDecoder(t *testing.T) {
 		data := make([]byte, 1024)
 		if err := dec.DecodePage(&hdr, data); err != nil {
 			t.Fatal(err)
-		} else if got, want := hdr, spec.Pages[i].Header; got != want {
-			t.Fatalf("page hdr mismatch:\ngot=%#v\nwant=%#v", got, want)
-		} else if got, want := data, spec.Pages[i].Data; !bytes.Equal(got, want) {
+		}
+		// Encoder now sets PageHeaderFlagCompressedSize, so compare only Pgno.
+		if got, want := hdr.Pgno, spec.Pages[i].Header.Pgno; got != want {
+			t.Fatalf("page hdr pgno mismatch:\ngot=%d\nwant=%d", got, want)
+		}
+		if got, want := hdr.Flags, uint16(ltx.PageHeaderFlagCompressedSize); got != want {
+			t.Fatalf("page hdr flags mismatch:\ngot=0x%x\nwant=0x%x", got, want)
+		}
+		if got, want := data, spec.Pages[i].Data; !bytes.Equal(got, want) {
 			t.Fatalf("page data mismatch:\ngot=%#v\nwant=%#v", got, want)
 		}
 	}
@@ -64,28 +70,29 @@ func TestDecoder(t *testing.T) {
 	}
 
 	// Verify page index.
+	// New format adds 4-byte compressed size prefix, so Size is 55 instead of 51.
 	index := dec.PageIndex()
 	if got, want := index, map[uint32]ltx.PageIndexElem{
-		1: {MinTXID: 1, MaxTXID: 1, Offset: 100, Size: 51},
-		2: {MinTXID: 1, MaxTXID: 1, Offset: 151, Size: 51},
+		1: {MinTXID: 1, MaxTXID: 1, Offset: 100, Size: 55},
+		2: {MinTXID: 1, MaxTXID: 1, Offset: 155, Size: 55},
 	}; !reflect.DeepEqual(got, want) {
 		t.Fatalf("page index mismatch:\ngot=%#v\nwant=%#v", got, want)
 	}
 
 	// Read page 1 by offset.
 	if hdr, data, err := ltx.DecodePageData(fileSpecData[100:]); err != nil {
 		t.Fatal(err)
-	} else if got, want := hdr, (ltx.PageHeader{Pgno: 1}); got != want {
-		t.Fatalf("page header mismatch:\ngot=%#v\nwant=%#v", got, want)
+	} else if got, want := hdr.Pgno, uint32(1); got != want {
+		t.Fatalf("page header pgno mismatch:\ngot=%d\nwant=%d", got, want)
 	} else if got, want := data, bytes.Repeat([]byte("2"), 1024); !bytes.Equal(got, want) {
 		t.Fatalf("page data mismatch:\ngot=%#v\nwant=%#v", got, want)
 	}
 
-	// Read page 2 by offset.
-	if hdr, data, err := ltx.DecodePageData(fileSpecData[151:]); err != nil {
+	// Read page 2 by offset. Offset is 155 with new format (+4 bytes per page).
+	if hdr, data, err := ltx.DecodePageData(fileSpecData[155:]); err != nil {
 		t.Fatal(err)
-	} else if got, want := hdr, (ltx.PageHeader{Pgno: 2}); got != want {
-		t.Fatalf("page header mismatch:\ngot=%#v\nwant=%#v", got, want)
+	} else if got, want := hdr.Pgno, uint32(2); got != want {
+		t.Fatalf("page header pgno mismatch:\ngot=%d\nwant=%d", got, want)
 	} else if got, want := data, bytes.Repeat([]byte("3"), 1024); !bytes.Equal(got, want) {
 		t.Fatalf("page data mismatch:\ngot=%#v\nwant=%#v", got, want)
 	}

encoder.go

@@ -243,18 +243,41 @@ func (enc *Encoder) EncodePage(hdr PageHeader, data []byte) (err error) {
 
 	offset := enc.n
 
-	// Encode & write header.
+	// Compress data first to get the compressed size.
+	enc.buf.Reset()
+	enc.zw.Reset(&enc.buf)
+	if _, err = enc.zw.Write(data); err != nil {
+		return fmt.Errorf("compress page data: %w", err)
+	}
+	if err := enc.zw.Close(); err != nil {
+		return fmt.Errorf("close lz4 writer: %w", err)
+	}
+	compressed := enc.buf.Bytes()
+
+	// Set flag indicating compressed size follows the page header.
+	hdr.Flags |= PageHeaderFlagCompressedSize
+
+	// Write page header.
 	b, err := hdr.MarshalBinary()
 	if err != nil {
 		return fmt.Errorf("marshal: %w", err)
 	} else if _, err := enc.write(b); err != nil {
-		return fmt.Errorf("write: %w", err)
+		return fmt.Errorf("write page header: %w", err)
+	}
+
+	// Write compressed size (4 bytes, big-endian).
+	sizeBuf := make([]byte, 4)
+	binary.BigEndian.PutUint32(sizeBuf, uint32(len(compressed)))
+	if _, err := enc.write(sizeBuf); err != nil {
+		return fmt.Errorf("write compressed size: %w", err)
 	}
 
-	// Write data to file.
-	if _, err = enc.writeCompressed(data); err != nil {
-		return fmt.Errorf("write page data: %w", err)
+	// Write compressed data.
+	if _, err := enc.w.Write(compressed); err != nil {
+		return fmt.Errorf("write compressed data: %w", err)
 	}
+	_, _ = enc.hash.Write(data) // hash the uncompressed data
+	enc.n += int64(len(compressed))
 
 	enc.pagesWritten++
 	enc.prevPgno = hdr.Pgno
@@ -273,34 +296,6 @@ func (enc *Encoder) write(b []byte) (n int, err error) {
 	return n, err
 }
 
-// write to the compressed writer & add to the checksum.
-// Returns the size of the compressed data.
-func (enc *Encoder) writeCompressed(b []byte) (n int, err error) {
-	// Reset the buffer & compressed writer.
-	enc.buf.Reset()
-	enc.zw.Reset(&enc.buf)
-
-	// Write to the compressed writer to the buffer and then write the buffer to the uncompressed writer.
-	// This is necessary so we can calculate the size of the compressed data for the page index.
-	if _, err = enc.zw.Write(b); err != nil {
-		return n, err
-	}
-
-	// Close the compressed writer to flush any remaining data.
-	if err := enc.zw.Close(); err != nil {
-		return n, fmt.Errorf("cannot close lz4 writer: %w", err)
-	}
-
-	compressed := enc.buf.Bytes()
-	n, err = enc.w.Write(compressed)
-
-	// Write the uncompressed data to the hash, but add the compressed length to the size.
-	_, _ = enc.hash.Write(b)
-	enc.n += int64(len(compressed))
-
-	return n, err
-}
-
 func (enc *Encoder) writeToHash(b []byte) {
 	_, _ = enc.hash.Write(b)
 	enc.n += int64(len(b))

go.mod

@@ -2,4 +2,4 @@ module github.com/superfly/ltx
 
 go 1.24
 
-require github.com/pierrec/lz4/v4 v4.1.22
+require github.com/pierrec/lz4/v4 v4.1.23

go.sum

@@ -1,2 +1,2 @@
-github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
-github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pierrec/lz4/v4 v4.1.23 h1:oJE7T90aYBGtFNrI8+KbETnPymobAhzRrR8Mu8n1yfU=
+github.com/pierrec/lz4/v4 v4.1.23/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=

ltx.go

@@ -405,6 +405,15 @@ func IsValidPageSize(sz uint32) bool {
 	return false
 }
 
+// PageHeader flags.
+const (
+	// PageHeaderFlagCompressedSize indicates that a 4-byte compressed size
+	// field follows the page header. This allows the decoder to know the exact
+	// size of the LZ4 frame without relying on the LZ4 library's frame
+	// concatenation behavior.
+	PageHeaderFlagCompressedSize = uint16(1 << 0)
+)
+
 // PageHeader represents the header for a single page in an LTX file.
 type PageHeader struct {
 	Pgno  uint32
@@ -421,8 +430,8 @@ func (h *PageHeader) Validate() error {
 	if h.Pgno == 0 {
 		return fmt.Errorf("page number required")
 	}
-	if h.Flags != 0 {
-		return fmt.Errorf("no flags allowed, reserved for future use")
+	if h.Flags & ^PageHeaderFlagCompressedSize != 0 {
+		return fmt.Errorf("invalid page header flags: 0x%04x", h.Flags)
 	}
 	return nil
 }

ltx_test.go

@@ -275,7 +275,7 @@ func TestPageHeader_Validate(t *testing.T) {
 	})
 	t.Run("ErrFlagsNotAllowed", func(t *testing.T) {
 		hdr := ltx.PageHeader{Pgno: 1, Flags: 2}
-		if err := hdr.Validate(); err == nil || err.Error() != `no flags allowed, reserved for future use` {
+		if err := hdr.Validate(); err == nil || err.Error() != `invalid page header flags: 0x0002` {
 			t.Fatalf("unexpected error: %s", err)
 		}
 	})
@@ -816,15 +816,19 @@ func assertFileSpecEqual(tb testing.TB, x, y *ltx.FileSpec) {
 		tb.Fatalf("page count: %d, want %d", got, want)
 	}
 	for i := range x.Pages {
-		if got, want := x.Pages[i].Header, y.Pages[i].Header; got != want {
-			tb.Fatalf("page header mismatch: i=%d\ngot=%#v\nwant=%#v", i, got, want)
+		// Compare only Pgno, not Flags. The encoder sets PageHeaderFlagCompressedSize
+		// on output, so Flags won't match the input spec.
+		if got, want := x.Pages[i].Header.Pgno, y.Pages[i].Header.Pgno; got != want {
+			tb.Fatalf("page header pgno mismatch: i=%d\ngot=%d\nwant=%d", i, got, want)
 		}
 		if got, want := x.Pages[i].Data, y.Pages[i].Data; !bytes.Equal(got, want) {
 			tb.Fatalf("page data mismatch: i=%d\ngot=%#v\nwant=%#v", i, got, want)
 		}
 	}
 
-	if got, want := x.Trailer, y.Trailer; got != want {
-		tb.Fatalf("trailer mismatch:\ngot=%#v\nwant=%#v", got, want)
+	// Compare only PostApplyChecksum, not FileChecksum. FileChecksum depends on
+	// the exact byte representation of the file, which changes when the format changes.
+	if got, want := x.Trailer.PostApplyChecksum, y.Trailer.PostApplyChecksum; got != want {
+		tb.Fatalf("trailer PostApplyChecksum mismatch:\ngot=0x%x\nwant=0x%x", got, want)
 	}
 }