Artifact [974b4842eb]

Artifact 974b4842eb0248c319694a6a95f652f55c0c506f712a538a108343af6247308e:

• File internal/sync/sparse_receive_safety_test.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: 4199) [annotate] [blame] [check-ins using]

package sync

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/danmestas/go-libfossil/db"
	"github.com/danmestas/go-libfossil/internal/cone"
	"github.com/danmestas/go-libfossil/internal/content"
	"github.com/danmestas/go-libfossil/internal/deck"
)

func TestSparseReceiveRetainsProtectedHoldings(t *testing.T) {
	r := setupSyncTestRepo(t)
	uuids := map[string]string{}
	for _, name := range []string{"corrupt", "existing", "private", "structural", "unsent", "wanted-drop"} {
		uuids[name] = storeAndLink(t, r, []byte("receive safety "+name))
	}
	var files []deck.FileCard
	for _, name := range []string{"corrupt", "existing", "private", "structural", "unsent", "wanted-drop"} {
		files = append(files, deck.FileCard{Name: "zzz/" + name, UUID: uuids[name]})
	}
	data, err := (&deck.Deck{C: "protected receive", D: time.Now().UTC(), U: deck.User("u"), F: files, P: []string{uuids["structural"]}}).Marshal()
	if err != nil {
		t.Fatal(err)
	}
	storeAndLink(t, r, data)
	unknown := storeAndLink(t, r, []byte("unreferenced ordinary bytes"))
	for _, table := range []string{"private", "unsent"} {
		if _, err := r.DB().Exec("INSERT OR IGNORE INTO "+table+"(rid) SELECT rid FROM blob WHERE uuid=?", uuids[table]); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := r.DB().Exec("UPDATE blob SET content=x'00' WHERE uuid=?", uuids["corrupt"]); err != nil {
		t.Fatal(err)
	}
	c, err := cone.New([]string{"src"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if err := content.SetStoreCone(r.DB(), c); err != nil {
		t.Fatal(err)
	}
	received := map[string]bool{unknown: true}
	for name, uuid := range uuids {
		if name != "existing" {
			received[uuid] = true
		}
	}
	elided := 0
	if err := r.WithTx(func(tx *db.Tx) error {
		var err error
		elided, err = filterSparseReceived(context.Background(), tx, received, nil)
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if elided != 1 {
		t.Fatalf("final elision count=%d, want only the one demoted payload", elided)
	}
	for name, uuid := range uuids {
		_, held := content.AvailableByUUID(r.DB(), uuid)
		if held != (name != "wanted-drop") {
			t.Fatalf("%s held=%v", name, held)
		}
	}
	if _, held := content.AvailableByUUID(r.DB(), unknown); !held {
		t.Fatal("unknown artifact discarded")
	}
}

type sparseCancelAfterDemotion struct {
	db.Querier
	cancel  context.CancelFunc
	demoted bool
}

func (q *sparseCancelAfterDemotion) Exec(query string, args ...any) (sql.Result, error) {
	res, err := q.Querier.Exec(query, args...)
	if err == nil && strings.HasPrefix(query, "UPDATE blob SET size=-1, content=NULL") {
		q.demoted = true
		q.cancel()
	}
	return res, err
}

func TestSparseReceiveCancellationRollsBackDemotion(t *testing.T) {
	r := setupSyncTestRepo(t)
	first := storeAndLink(t, r, []byte("first rollback payload"))
	second := storeAndLink(t, r, []byte("second rollback payload"))
	data, err := (&deck.Deck{C: "rollback receive", D: time.Now().UTC(), U: deck.User("u"), F: []deck.FileCard{
		{Name: "zzz/a", UUID: first}, {Name: "zzz/b", UUID: second},
	}}).Marshal()
	if err != nil {
		t.Fatal(err)
	}
	storeAndLink(t, r, data)
	c, err := cone.New([]string{"src"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if err := content.SetStoreCone(r.DB(), c); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var demoted bool
	err = r.WithTx(func(tx *db.Tx) error {
		q := &sparseCancelAfterDemotion{Querier: tx, cancel: cancel}
		_, err := filterSparseReceived(ctx, q, map[string]bool{first: true, second: true}, nil)
		demoted = q.demoted
		return err
	})
	if !demoted || !errors.Is(err, context.Canceled) {
		t.Fatalf("demoted=%v, err=%v; want actual mutation followed by cancellation", demoted, err)
	}
	for _, uuid := range []string{first, second} {
		rid, held := content.AvailableByUUID(r.DB(), uuid)
		if !held {
			t.Fatalf("canceled receive lost %s", uuid)
		}
		if _, err := content.Expand(r.DB(), rid); err != nil {
			t.Fatal(err)
		}
		var phantoms int
		if err := r.DB().QueryRow("SELECT count(*) FROM phantom WHERE rid=?", rid).Scan(&phantoms); err != nil || phantoms != 0 {
			t.Fatalf("canceled receive phantom=%d, err=%v", phantoms, err)
		}
	}
}