Artifact [800037a59f]

Artifact 800037a59f439d51b44b1cc811b9f2a509cac14f3c114a38b1d106785865123d:

• File internal/content/compact.go — part of check-in [1fd12b3ed3] at 2026-10-28 21:53:09 on branch feat/chunked-blobs — Store blobs over a hosted value limit across blobpart rows

  A hosted SQLite that caps value size (a Durable Object's 2 MB) cannot hold a large file's stored blob in one blob.content value. With the new db.DriverConfig.MaxValueBytes set, a stored value over the limit is written in 1,000,000-byte parts to blobpart(rid, seq, content), created on first use, leaving blob.content NULL and blob.size the expanded length, so phantoms (size < 0) stay unambiguous. Chunking sits below compression: hashes, deltas and the wire format are unchanged.

  internal/blob owns the seam (ReadStored, WriteStored, InsertStored, StoredHead, DropStored, SweepParts and two scan expressions), and every direct reader or writer of blob.content now goes through it: delta rewrite and undelta, receive, clone send and walk, compaction, sparse demotion, seed drop, shun purge, horizon cost and cone preparation.

  Zero, the default, keeps every value inline and creates no table, so local stores stay byte-identical. A store with blobpart rows is not readable by stock fossil; enable it only for hosted backends.

  Closes #459 (user: [email protected], size: 4575) [annotate] [blame] [check-ins using]

package content

import (
	"context"
	"errors"
	"fmt"
	"slices"

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

// CompactStore turns proved outside content into recoverable phantoms. Dense
// stores (including lost policy) are unchanged. Structural artifacts, unknown
// content, classification provenance and retained delta dependencies stay held.
// The existing write transaction excludes competing writers and rolls back
// every content change together on cancellation, corruption or SQL failure.
func CompactStore(ctx context.Context, d *db.DB) (int, error) {
	if ctx == nil {
		panic("content.CompactStore: context must not be nil")
	}
	if d == nil {
		panic("content.CompactStore: DB must not be nil")
	}
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	count := 0
	err := d.WithTx(func(tx *db.Tx) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		c, err := StoreCone(tx)
		if err != nil {
			return err
		}
		if c == nil {
			return nil
		}
		drop, err := compactCandidates(ctx, tx, c)
		if err != nil {
			return err
		}
		if err := compactRetainBases(tx, drop); err != nil {
			return err
		}
		var ordered []libfossil.FslID
		for rid := range drop {
			ordered = append(ordered, rid)
		}
		slices.Sort(ordered)
		for _, rid := range ordered {
			if err := ctx.Err(); err != nil {
				return err
			}
			if err := blob.DropStored(tx, rid); err != nil {
				return err
			}
			if _, err := tx.Exec("INSERT OR IGNORE INTO phantom(rid) VALUES(?)", rid); err != nil {
				return err
			}
			if _, err := tx.Exec("DELETE FROM delta WHERE rid=?", rid); err != nil {
				return err
			}
			count++
		}
		return ctx.Err()
	})
	if err != nil {
		return 0, fmt.Errorf("content.CompactStore: %w", err)
	}
	if count < 0 {
		panic("content.CompactStore: negative compacted count")
	}
	return count, nil
}

func compactCandidates(ctx context.Context, tx *db.Tx, c *cone.Cone) (map[libfossil.FslID]bool, error) {
	refs, err := LoadConeReferences(ctx, tx, c)
	if err != nil {
		return nil, err
	}
	held := map[string]libfossil.FslID{}
	drop := map[libfossil.FslID]bool{}
	// Classification tolerates unavailable bytes. Destructive compaction must
	// separately verify every held payload before changing any holdings.
	err = ExpandAll(tx, func(rid libfossil.FslID, uuid string, data []byte, expandErr error) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		if expandErr != nil {
			return expandErr
		}
		held[uuid] = rid
		if refs.Outside(rid) {
			drop[rid] = true
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	for uuid := range refs.required {
		if _, ok := held[uuid]; !ok {
			return nil, fmt.Errorf("incomplete spine: artifact %s is not held", uuid)
		}
	}
	for rid, uuid := range refs.names {
		if refs.Required(rid) {
			if _, ok := held[uuid]; !ok {
				return nil, fmt.Errorf("incomplete spine: artifact %s is not held", uuid)
			}
		}
	}
	if err := refs.RetainProvenance(ctx, drop); err != nil {
		return nil, err
	}
	return drop, nil
}

// Unsent or private bytes may have no peer that can ever restore them.
func compactRetainLocal(tx *db.Tx, drop map[libfossil.FslID]bool) error {
	rows, err := tx.Query("SELECT rid FROM unsent UNION SELECT rid FROM private")
	if err != nil {
		return err
	}
	for rows.Next() {
		var rid libfossil.FslID
		if err := rows.Scan(&rid); err != nil {
			return errors.Join(err, rows.Close())
		}
		delete(drop, rid)
	}
	return errors.Join(rows.Err(), rows.Close())
}

// Preserve the transitive source closure of everything that will remain held.
// No retained delta can come to depend on a phantom as a result of compaction.
func compactRetainBases(tx *db.Tx, drop map[libfossil.FslID]bool) error {
	if err := compactRetainLocal(tx, drop); err != nil {
		return err
	}
	rows, err := tx.Query("SELECT d.rid, d.srcid FROM delta d JOIN blob b ON b.rid=d.rid WHERE b.size>=0")
	if err != nil {
		return err
	}
	sources := map[libfossil.FslID]libfossil.FslID{}
	for rows.Next() {
		var rid, source libfossil.FslID
		if err := rows.Scan(&rid, &source); err != nil {
			return errors.Join(err, rows.Close())
		}
		sources[rid] = source
	}
	if err := errors.Join(rows.Err(), rows.Close()); err != nil {
		return err
	}
	for rid := range sources {
		if drop[rid] {
			continue
		}
		for source := sources[rid]; source > 0; source = sources[source] {
			if !drop[source] {
				break
			}
			delete(drop, source)
		}
	}
	return nil
}