Artifact [9f4d139b0b]

Artifact 9f4d139b0bab677d9b2aacfcf949b0ca1146890b8f2c6d0ce08ddf2662c82690:

• File internal/sync/sparse_required_local.go — part of check-in [0fc4d176c3] at 2026-10-28 22:52:43 on branch feat/chunked-blobs — Record chunked length in blob.content and verify parts on read

  Review of the chunked-blob change found a missing last part read back short with no error, and a chunked write outside a transaction could leave a torn row that looked held.

  A chunked row now keeps its stored length in blob.content as an INTEGER. Reads check the parts run unbroken from seq 0, are full-sized but for the last, and add up to that length; anything else is ErrPartsCorrupt, and a NULL content is no longer mistaken for a chunked one. StoredLenSQL reads the length off the row and needs no schema probe.

  Chunked writes, and the phantom fill around them, run in a savepoint of the caller's transaction or a transaction of their own, so a failed part leaves nothing behind. A value needing more than 4096 parts is an error, not a panic, since stored values can come from peers. The cone and sparse local payload reads use ReadStoredExact, which loads a value only at the length already budgeted for.

  MaxValueBytes is documented as process-wide, and unversioned content as not chunked. (user: [email protected], size: 4210) [annotate] [blame] [check-ins using]

package sync

import (
	"bytes"
	"compress/zlib"
	"encoding/binary"
	"fmt"
	"io"

	"github.com/danmestas/go-libfossil/db"
	"github.com/danmestas/go-libfossil/internal/blob"
	"github.com/danmestas/go-libfossil/internal/content"
	libfossil "github.com/danmestas/go-libfossil/internal/fsltype"
)

type requiredLocalFrame struct {
	rid  libfossil.FslID
	uuid string
	size int64
}

// Local bases are untrusted too. Account for compressed buffers, inflated
// programs and delta outputs before allocating any of them, not after Expand.
func (f *requiredFetch) expandLocal(q db.Querier, rid libfossil.FslID) ([]byte, error) {
	if q == nil || rid <= 0 {
		panic("sync.requiredFetch.expandLocal: nil querier or nonpositive rid")
	}
	var path []requiredLocalFrame
	seen := make(map[libfossil.FslID]bool)
	for depth := 0; rid > 0; depth++ {
		if depth >= maxRequiredDeltaDepth || seen[rid] {
			return nil, fmt.Errorf("sparse required content: cyclic or excessive local delta depth")
		}
		seen[rid] = true
		frame := requiredLocalFrame{rid: rid}
		var source libfossil.FslID
		err := q.QueryRow(`SELECT b.uuid,b.size,COALESCE(d.srcid,0)
			FROM blob b LEFT JOIN delta d ON d.rid=b.rid WHERE b.rid=?`, rid).
			Scan(&frame.uuid, &frame.size, &source)
		if err != nil {
			return nil, err
		}
		if err := f.checkName(frame.uuid); err != nil {
			return nil, err
		}
		if content.IsPrivate(q, int64(rid)) && !f.opts.Private {
			return nil, fmt.Errorf("sparse required content: private local delta basis requires permission")
		}
		if frame.size < 0 || frame.size > int64(maxRequiredBytes-f.bytes) {
			return nil, fmt.Errorf("sparse required content: local size exceeds staging budget")
		}
		path = append(path, frame)
		rid = source
	}
	var full []byte
	for i := len(path) - 1; i >= 0; i-- {
		frame := path[i]
		payload, err := f.localPayload(q, frame)
		if err != nil {
			return nil, err
		}
		if i == len(path)-1 {
			full = payload
		} else {
			full, err = f.applyDelta(frame.uuid, payload, full)
			if err != nil {
				return nil, err
			}
		}
		if int64(len(full)) != frame.size || !hashMatches(frame.uuid, full) {
			return nil, fmt.Errorf("sparse required content: invalid local size or hash for %s", frame.uuid)
		}
	}
	return full, nil
}

func (f *requiredFetch) localPayload(q db.Querier, frame requiredLocalFrame) ([]byte, error) {
	if err := f.ctx.Err(); err != nil {
		return nil, err
	}
	packedSize, prefix, err := blob.StoredHead(q, frame.rid, 4)
	if err != nil {
		return nil, err
	}
	if len(prefix) != 4 || packedSize < 5 {
		return nil, fmt.Errorf("sparse required content: invalid local compression header")
	}
	plainSize := int64(binary.BigEndian.Uint32(prefix))
	remaining := int64(maxRequiredBytes - f.bytes)
	if packedSize > remaining || plainSize >= remaining-packedSize {
		return nil, fmt.Errorf("sparse required content: local inflation exceeds staging budget")
	}
	if err := f.reserve(int(packedSize + plainSize + 1)); err != nil {
		return nil, err
	}
	// Reading only the reserved length, then re-checking the prefix, holds
	// the read to the row the budget was reserved for; size and hash are
	// checked once it is expanded.
	packed, err := blob.ReadStoredExact(q, frame.rid, packedSize)
	if err != nil {
		return nil, err
	}
	if !bytes.Equal(packed[:4], prefix) {
		return nil, fmt.Errorf("sparse required content: local content changed")
	}
	return requiredInflate(packed, int(plainSize))
}

func requiredInflate(packed []byte, size int) ([]byte, error) {
	reader, err := zlib.NewReader(bytes.NewReader(packed[4:]))
	if err != nil {
		return nil, err
	}
	// Read one extra byte to detect a forged low prefix without expanding the
	// unbounded remainder. Reaching EOF also verifies the zlib checksum.
	plain := make([]byte, size+1)
	n, readErr := io.ReadFull(reader, plain)
	closeErr := reader.Close()
	if readErr != io.EOF && readErr != io.ErrUnexpectedEOF {
		if readErr == nil {
			return nil, fmt.Errorf("sparse required content: local compression length mismatch")
		}
		return nil, readErr
	}
	if closeErr != nil {
		return nil, closeErr
	}
	if n != size {
		return nil, fmt.Errorf("sparse required content: local compression length mismatch")
	}
	return plain[:size], nil
}