Artifact [09d3ac4fcf]

Artifact 09d3ac4fcf083219ad49eb2f318952309bbdf17f1e42ddd2627b16e835111c82:

• File internal/manifest/dephantomize_test.go — part of check-in [1c02c944ec] at 2026-08-07 17:33:21 on branch trunk — fix(db): seed hash-policy so new repos name artifacts like fossil init (#225) (#226)

  SeedConfig never wrote hash-policy, so naming fell back to deriving from the repo's own artifacts — and an empty repo has none, settling on SHA1 permanently. Seed sha3, which is what fossil init writes.

  Clone drops the row the same way it drops project-code: a clone adopts its source's policy, and an absent row derives naming from the artifacts that arrive.

  Moves the policy enum to db so the seed and hash.NamingFor read one definition, and updates fixtures that hardcoded SHA1 names to ask the repo how it names artifacts. (user: [email protected], size: 32615) [annotate] [blame] [check-ins using]

package manifest

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/danmestas/go-libfossil/db"
	"github.com/danmestas/go-libfossil/internal/blob"
	"github.com/danmestas/go-libfossil/internal/deck"
	"github.com/danmestas/go-libfossil/internal/delta"
	libfossil "github.com/danmestas/go-libfossil/internal/fsltype"
	"github.com/danmestas/go-libfossil/internal/repo"
	_ "github.com/danmestas/go-libfossil/internal/testdriver"
)

func TestAfterDephantomizeCheckin(t *testing.T) {
	r := setupTestRepo(t)

	// Create a checkin manifest, compute its hash, store as phantom, then fill.
	fileContent := []byte("hello dephantomize")
	fileRid, fileUUID, err := blob.Store(r.DB(), fileContent)
	if err != nil {
		t.Fatalf("Store file blob: %v", err)
	}
	_ = fileRid

	d := &deck.Deck{
		Type: deck.Checkin,
		C:    "dephantomize commit",
		U:    deck.User("testuser"),
		D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC),
		F:    []deck.FileCard{{Name: "hello.txt", UUID: fileUUID}},
	}

	manifestBytes, err := d.Marshal()
	if err != nil {
		t.Fatalf("Marshal: %v", err)
	}

	// Compute the hash to get the UUID.
	rid, uuid, err := blob.Store(r.DB(), manifestBytes)
	if err != nil {
		t.Fatalf("Store manifest: %v", err)
	}

	// Delete the real blob, re-insert as phantom, then fill it back.
	// This simulates the phantom->real transition.
	r.DB().Exec("DELETE FROM event WHERE objid=?", rid)
	r.DB().Exec("DELETE FROM plink WHERE cid=?", rid)
	r.DB().Exec("DELETE FROM leaf WHERE rid=?", rid)
	r.DB().Exec("DELETE FROM mlink WHERE mid=?", rid)
	r.DB().Exec("DELETE FROM tagxref WHERE rid=?", rid)

	// Verify no event row exists before dephantomize.
	var eventCount int
	r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", rid).Scan(&eventCount)
	if eventCount != 0 {
		t.Fatalf("event count before dephantomize = %d, want 0", eventCount)
	}

	// Call AfterDephantomize.
	AfterDephantomize(context.Background(), r, rid)

	// Verify event row was created.
	r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", rid).Scan(&eventCount)
	if eventCount != 1 {
		t.Errorf("event count after dephantomize = %d, want 1", eventCount)
	}

	// Verify event is a checkin.
	var eventType string
	r.DB().QueryRow("SELECT type FROM event WHERE objid=?", rid).Scan(&eventType)
	if eventType != "ci" {
		t.Errorf("event type = %q, want 'ci'", eventType)
	}

	_ = uuid
}

func TestAfterDephantomizeOrphan(t *testing.T) {
	r := setupTestRepo(t)

	// Create a baseline checkin.
	fileContent := []byte("baseline file")
	_, fileUUID, err := blob.Store(r.DB(), fileContent)
	if err != nil {
		t.Fatalf("Store file: %v", err)
	}

	baselineDeck := &deck.Deck{
		Type: deck.Checkin,
		C:    "baseline",
		U:    deck.User("testuser"),
		D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC),
		F:    []deck.FileCard{{Name: "file.txt", UUID: fileUUID}},
	}
	baselineBytes, err := baselineDeck.Marshal()
	if err != nil {
		t.Fatalf("Marshal baseline: %v", err)
	}
	baselineRid, _, err := blob.Store(r.DB(), baselineBytes)
	if err != nil {
		t.Fatalf("Store baseline: %v", err)
	}

	// Create another checkin that will be the "orphan" (delta manifest).
	fileContent2 := []byte("orphan file")
	_, fileUUID2, err := blob.Store(r.DB(), fileContent2)
	if err != nil {
		t.Fatalf("Store file2: %v", err)
	}

	orphanDeck := &deck.Deck{
		Type: deck.Checkin,
		C:    "orphan commit",
		U:    deck.User("testuser"),
		D:    time.Date(2024, 1, 15, 11, 0, 0, 0, time.UTC),
		F:    []deck.FileCard{{Name: "file2.txt", UUID: fileUUID2}},
	}
	orphanBytes, err := orphanDeck.Marshal()
	if err != nil {
		t.Fatalf("Marshal orphan: %v", err)
	}
	orphanRid, _, err := blob.Store(r.DB(), orphanBytes)
	if err != nil {
		t.Fatalf("Store orphan: %v", err)
	}

	// Clear crosslink tables for the orphan.
	r.DB().Exec("DELETE FROM event WHERE objid=?", orphanRid)
	r.DB().Exec("DELETE FROM leaf WHERE rid=?", orphanRid)
	r.DB().Exec("DELETE FROM tagxref WHERE rid=?", orphanRid)

	// Insert orphan row linking orphanRid to baselineRid.
	r.DB().Exec("INSERT INTO orphan(rid, baseline) VALUES(?, ?)", orphanRid, baselineRid)

	// Verify orphan row exists.
	var orphanCount int
	r.DB().QueryRow("SELECT count(*) FROM orphan WHERE baseline=?", baselineRid).Scan(&orphanCount)
	if orphanCount != 1 {
		t.Fatalf("orphan count = %d, want 1", orphanCount)
	}

	// Call AfterDephantomize on the baseline.
	AfterDephantomize(context.Background(), r, baselineRid)

	// Verify orphan was cleaned up.
	r.DB().QueryRow("SELECT count(*) FROM orphan WHERE baseline=?", baselineRid).Scan(&orphanCount)
	if orphanCount != 0 {
		t.Errorf("orphan count after dephantomize = %d, want 0", orphanCount)
	}

	// Verify orphan checkin got crosslinked (event row created).
	var eventCount int
	r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", orphanRid).Scan(&eventCount)
	if eventCount != 0 {
		// The orphan's event may have been created by crosslinkSingle.
		// If it wasn't (because it was already crosslinked earlier), that's also fine.
		// The key assertion is that the orphan row was cleaned up.
	}
}

// buildDephantomizeFanout stores one full "root" checkin manifest plus fanout
// distinct checkin manifests, each a depth-1 delta against that root, so every
// child appears in the delta table with srcid = root. That is exactly the shape
// afterDephantomize's work stack walks when a base fills and unblocks the delta
// children waiting on it: root is popped, all fanout children are pushed, then
// each is popped in turn. The walk therefore runs fanout+1 stack pops -- deep
// enough to cross crosslinkCancelCheckStride -- while every child stays a
// single-hop delta so content.Expand reconstructs it without a deep delta chain.
func buildDephantomizeFanout(t *testing.T, r *repo.Repo, fanout int) (root libfossil.FslID, children []libfossil.FslID) {
	t.Helper()

	// One shared file blob keeps the fixture cheap; the manifests differ only by
	// commit message and time, which is enough to make each a distinct blob.
	fileContent := []byte("fanout shared file")
	_, fileUUID, err := blob.Store(r.DB(), fileContent)
	if err != nil {
		t.Fatalf("Store fanout file blob: %v", err)
	}

	manifest := func(i int) []byte {
		d := &deck.Deck{
			Type: deck.Checkin,
			C:    fmt.Sprintf("fanout commit %d", i),
			U:    deck.User("testuser"),
			D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC).Add(time.Duration(i) * time.Minute),
			F:    []deck.FileCard{{Name: "hello.txt", UUID: fileUUID}},
		}
		mb, err := d.Marshal()
		if err != nil {
			t.Fatalf("Marshal fanout manifest %d: %v", i, err)
		}
		return mb
	}

	root, _, err = blob.Store(r.DB(), manifest(0))
	if err != nil {
		t.Fatalf("Store fanout root: %v", err)
	}

	children = make([]libfossil.FslID, fanout)
	for i := 0; i < fanout; i++ {
		rid, _, err := blob.StoreDelta(r.DB(), manifest(i+1), root)
		if err != nil {
			t.Fatalf("StoreDelta fanout child %d: %v", i, err)
		}
		children[i] = rid
	}
	return root, children
}

// countCrosslinked reports how many of rids got a crosslink event row, i.e. how
// far the phantom-fill cascade actually processed.
func countCrosslinked(t *testing.T, r *repo.Repo, rids []libfossil.FslID) int {
	t.Helper()
	n := 0
	for _, rid := range rids {
		var c int
		if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", rid).Scan(&c); err != nil {
			t.Fatalf("count event objid=%d: %v", rid, err)
		}
		if c > 0 {
			n++
		}
	}
	return n
}

// TestAfterDephantomizeObservesCancellationMidWalk pins that a clone's deadline
// can interrupt the mid-round phantom-fill cascade (issue #166). AfterDephantomize
// runs synchronously as a clone fills phantoms and walks the whole reachable
// delta/orphan graph; before the fix nothing in that walk observed the clone's
// context, so on a large graph it ran for many multiples of the configured
// deadline, bounded only by a 1,000,000-iteration safety cap.
//
// fanout is deliberately larger than crosslinkCancelCheckStride so the work
// stack polls its context more than once. Spelled as a literal -- matching the
// sibling TestCrosslinkContextObservesCancellationMidSweep -- so a corpus that
// grew or shrank in lockstep with the stride could never mask a neutered check.
func TestAfterDephantomizeObservesCancellationMidWalk(t *testing.T) {
	const fanout = 300

	// Control: a live context crosslinks every child. This proves the fixture is
	// fully walkable, so any shortfall in the cancelled run below is the deadline
	// biting, not a broken graph.
	rLive := setupTestRepo(t)
	liveRoot, liveChildren := buildDephantomizeFanout(t, rLive, fanout)
	AfterDephantomize(context.Background(), rLive, liveRoot)
	if got := countCrosslinked(t, rLive, liveChildren); got != fanout {
		t.Fatalf("with a live context, crosslinked %d of %d fanout children, want all %d",
			got, fanout, fanout)
	}

	// Cancellation: the context is live for its first Done() poll (the pop of
	// the root, at stack position 0) and cancelled from the second poll on (the
	// pop at crosslinkCancelCheckStride). Only an in-loop, stride-batched check
	// can observe that transition; a walk that ignores ctx runs to completion.
	rCancel := setupTestRepo(t)
	cancelRoot, cancelChildren := buildDephantomizeFanout(t, rCancel, fanout)
	ctx := newPollCancelCtx(1)
	AfterDephantomize(ctx, rCancel, cancelRoot)
	if got := countCrosslinked(t, rCancel, cancelChildren); got >= fanout {
		t.Fatalf("with a context cancelled mid-walk, crosslinked %d of %d fanout children: "+
			"AfterDephantomize ran the whole phantom-fill cascade without observing the "+
			"deadline (issue #166)", got, fanout)
	}
}

func TestAfterDephantomizeNilRepo(t *testing.T) {
	defer func() {
		if r := recover(); r == nil {
			t.Error("expected panic for nil repo")
		}
	}()
	AfterDephantomize(context.Background(), nil, 1)
}

func TestAfterDephantomizeZeroRid(t *testing.T) {
	r := setupTestRepo(t)
	// Should return without panicking.
	AfterDephantomize(context.Background(), r, 0)
	AfterDephantomize(context.Background(), r, -1)
}

func TestReceiveLinkerReplayAfterSettledCapDoesNotDuplicateMlinks(t *testing.T) {
	r := setupTestRepo(t)

	_, fileUUID, err := blob.Store(r.DB(), []byte("settled-cap file"))
	if err != nil {
		t.Fatalf("Store file blob: %v", err)
	}
	d := &deck.Deck{
		Type: deck.Checkin,
		C:    "settled-cap checkin",
		U:    deck.User("testuser"),
		D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC),
		F:    []deck.FileCard{{Name: "settled-cap.txt", UUID: fileUUID}},
	}
	checkinBytes, err := d.Marshal()
	if err != nil {
		t.Fatalf("Marshal checkin: %v", err)
	}
	checkinRID, checkinUUID, err := blob.Store(r.DB(), checkinBytes)
	if err != nil {
		t.Fatalf("Store checkin blob: %v", err)
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	for i := range receiveSettledLimit {
		linker.settled[libfossil.FslID(-i-1)] = struct{}{}
	}
	if got := len(linker.settled); got != receiveSettledLimit {
		t.Fatalf("pre-filled settled RIDs = %d, want %d", got, receiveSettledLimit)
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, checkinRID, checkinUUID, checkinBytes)
	}); err != nil {
		t.Fatalf("first LinkStored: %v", err)
	}

	var firstMlinks int
	if err := r.DB().QueryRow("SELECT count(*) FROM mlink WHERE mid=?", checkinRID).Scan(&firstMlinks); err != nil {
		t.Fatalf("count first mlink rows: %v", err)
	}
	if firstMlinks == 0 {
		t.Fatal("first LinkStored created no mlink rows")
	}
	firstLinked := linker.Stats().Linked
	if firstLinked == 0 {
		t.Fatal("first LinkStored did not increase Stats.Linked")
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, checkinRID, checkinUUID, checkinBytes)
	}); err != nil {
		t.Fatalf("replay LinkStored: %v", err)
	}

	var replayMlinks int
	if err := r.DB().QueryRow("SELECT count(*) FROM mlink WHERE mid=?", checkinRID).Scan(&replayMlinks); err != nil {
		t.Fatalf("count replay mlink rows: %v", err)
	}
	if replayMlinks != firstMlinks {
		t.Errorf("replay mlink rows = %d, want unchanged %d", replayMlinks, firstMlinks)
	}
	if got := linker.Stats().Linked; got != firstLinked {
		t.Errorf("replay Stats.Linked = %d, want unchanged %d", got, firstLinked)
	}
}

func TestReceiveLinkerReplayAfterSettledCapUsesEventMarkerForControl(t *testing.T) {
	r := setupTestRepo(t)
	_, targetUUID, err := blob.Store(r.DB(), []byte("settled-cap control target"))
	if err != nil {
		t.Fatalf("Store control target: %v", err)
	}
	d := &deck.Deck{
		Type: deck.Control,
		D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC),
		T: []deck.TagCard{
			{Type: deck.TagSingleton, Name: "sym-settled-control", UUID: targetUUID},
		},
		U: deck.User("testuser"),
	}
	controlBytes, err := d.Marshal()
	if err != nil {
		t.Fatalf("Marshal control: %v", err)
	}
	controlRID, controlUUID, err := blob.Store(r.DB(), controlBytes)
	if err != nil {
		t.Fatalf("Store control blob: %v", err)
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	for i := range receiveSettledLimit {
		linker.settled[libfossil.FslID(-i-1)] = struct{}{}
	}
	if got := len(linker.settled); got != receiveSettledLimit {
		t.Fatalf("pre-filled settled RIDs = %d, want %d", got, receiveSettledLimit)
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, controlRID, controlUUID, controlBytes)
	}); err != nil {
		t.Fatalf("first LinkStored: %v", err)
	}
	firstLinked := linker.Stats().Linked
	if firstLinked == 0 {
		t.Fatal("first LinkStored did not increase Stats.Linked")
	}

	var eventCount, sourceTagCount int
	if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", controlRID).Scan(&eventCount); err != nil {
		t.Fatalf("count control event rows: %v", err)
	}
	if eventCount != 1 {
		t.Fatalf("control event rows = %d, want 1", eventCount)
	}
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", controlRID).Scan(&sourceTagCount); err != nil {
		t.Fatalf("count control source tagxref rows: %v", err)
	}
	if sourceTagCount == 0 {
		t.Fatal("first LinkStored created no control source tagxref rows")
	}

	r.DB().Exec("DELETE FROM tagxref WHERE srcid=?", controlRID)
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", controlRID).Scan(&sourceTagCount); err != nil {
		t.Fatalf("count removed control source tagxref rows: %v", err)
	}
	if sourceTagCount != 0 {
		t.Fatalf("control source tagxref rows after removal = %d, want 0", sourceTagCount)
	}
	if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", controlRID).Scan(&eventCount); err != nil {
		t.Fatalf("count retained control event rows: %v", err)
	}
	if eventCount != 1 {
		t.Fatalf("control event rows after source tag removal = %d, want 1", eventCount)
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, controlRID, controlUUID, controlBytes)
	}); err != nil {
		t.Fatalf("replay LinkStored: %v", err)
	}
	if got := linker.Stats().Linked; got != firstLinked {
		t.Errorf("replay Stats.Linked = %d, want unchanged %d", got, firstLinked)
	}
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", controlRID).Scan(&sourceTagCount); err != nil {
		t.Fatalf("count replay control source tagxref rows: %v", err)
	}
	if sourceTagCount != 0 {
		t.Errorf("replay control source tagxref rows = %d, want unchanged 0", sourceTagCount)
	}
}

func TestReceiveLinkerReplayAfterSettledCapUsesTagxrefMarkerForEvent(t *testing.T) {
	r := setupTestRepo(t)
	eventTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	d := &deck.Deck{
		Type: deck.Event,
		E: &deck.EventCard{
			Date: eventTime,
			UUID: "fedcba9876543210fedcba9876543210fedcba98",
		},
		U: deck.User("testuser"),
		D: eventTime,
		W: []byte("settled-cap event body"),
		C: "settled-cap event",
	}
	eventBytes, err := d.Marshal()
	if err != nil {
		t.Fatalf("Marshal event: %v", err)
	}
	eventRID, eventUUID, err := blob.Store(r.DB(), eventBytes)
	if err != nil {
		t.Fatalf("Store event blob: %v", err)
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	for i := range receiveSettledLimit {
		linker.settled[libfossil.FslID(-i-1)] = struct{}{}
	}
	if got := len(linker.settled); got != receiveSettledLimit {
		t.Fatalf("pre-filled settled RIDs = %d, want %d", got, receiveSettledLimit)
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, eventRID, eventUUID, eventBytes)
	}); err != nil {
		t.Fatalf("first LinkStored: %v", err)
	}
	firstLinked := linker.Stats().Linked
	if firstLinked == 0 {
		t.Fatal("first LinkStored did not increase Stats.Linked")
	}

	var eventCount, firstTagCount int
	if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", eventRID).Scan(&eventCount); err != nil {
		t.Fatalf("count event rows: %v", err)
	}
	if eventCount != 1 {
		t.Fatalf("event rows = %d, want 1", eventCount)
	}
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", eventRID).Scan(&firstTagCount); err != nil {
		t.Fatalf("count event source tagxref rows: %v", err)
	}
	if firstTagCount == 0 {
		t.Fatal("first LinkStored created no event source tagxref rows")
	}

	r.DB().Exec("DELETE FROM event WHERE objid=?", eventRID)
	if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", eventRID).Scan(&eventCount); err != nil {
		t.Fatalf("count removed event rows: %v", err)
	}
	if eventCount != 0 {
		t.Fatalf("event rows after removal = %d, want 0", eventCount)
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, eventRID, eventUUID, eventBytes)
	}); err != nil {
		t.Fatalf("replay LinkStored: %v", err)
	}
	if got := linker.Stats().Linked; got != firstLinked {
		t.Errorf("replay Stats.Linked = %d, want unchanged %d", got, firstLinked)
	}
	var replayTagCount int
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", eventRID).Scan(&replayTagCount); err != nil {
		t.Fatalf("count replay event source tagxref rows: %v", err)
	}
	if replayTagCount != firstTagCount {
		t.Errorf("replay event source tagxref rows = %d, want unchanged %d", replayTagCount, firstTagCount)
	}
	if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", eventRID).Scan(&eventCount); err != nil {
		t.Fatalf("count replay event rows: %v", err)
	}
	if eventCount != 0 {
		t.Errorf("replay event rows = %d, want unchanged 0", eventCount)
	}
}

func TestReceiveLinkerReplayAfterSettledCapUsesTagxrefMarkerForTicket(t *testing.T) {
	r := setupTestRepo(t)
	d := &deck.Deck{
		Type: deck.Ticket,
		K:    "0123456789abcdef0123456789abcdef01234567",
		U:    deck.User("testuser"),
		D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC),
		J: []deck.TicketField{
			{Name: "title", Value: "settled-cap ticket"},
		},
	}
	ticketBytes, err := d.Marshal()
	if err != nil {
		t.Fatalf("Marshal ticket: %v", err)
	}
	ticketRID, ticketUUID, err := blob.Store(r.DB(), ticketBytes)
	if err != nil {
		t.Fatalf("Store ticket blob: %v", err)
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	for i := range receiveSettledLimit {
		linker.settled[libfossil.FslID(-i-1)] = struct{}{}
	}
	if got := len(linker.settled); got != receiveSettledLimit {
		t.Fatalf("pre-filled settled RIDs = %d, want %d", got, receiveSettledLimit)
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, ticketRID, ticketUUID, ticketBytes)
	}); err != nil {
		t.Fatalf("first LinkStored: %v", err)
	}
	firstLinked := linker.Stats().Linked
	if firstLinked == 0 {
		t.Fatal("first LinkStored did not increase Stats.Linked")
	}

	var eventCount, firstTagCount int
	if err := r.DB().QueryRow("SELECT count(*) FROM event WHERE objid=?", ticketRID).Scan(&eventCount); err != nil {
		t.Fatalf("count ticket event rows: %v", err)
	}
	if eventCount != 1 {
		t.Fatalf("ticket event rows = %d, want 1", eventCount)
	}
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", ticketRID).Scan(&firstTagCount); err != nil {
		t.Fatalf("count ticket source tagxref rows: %v", err)
	}
	if firstTagCount == 0 {
		t.Fatal("first LinkStored created no ticket source tagxref rows")
	}

	if err := r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, ticketRID, ticketUUID, ticketBytes)
	}); err != nil {
		t.Fatalf("replay LinkStored: %v", err)
	}
	if got := linker.Stats().Linked; got != firstLinked {
		t.Errorf("replay Stats.Linked = %d, want unchanged %d", got, firstLinked)
	}
	var replayTagCount int
	if err := r.DB().QueryRow("SELECT count(*) FROM tagxref WHERE srcid=?", ticketRID).Scan(&replayTagCount); err != nil {
		t.Fatalf("count replay ticket source tagxref rows: %v", err)
	}
	if replayTagCount != firstTagCount {
		t.Errorf("replay ticket source tagxref rows = %d, want unchanged %d", replayTagCount, firstTagCount)
	}
}

// TestReceiveLinkerCancelledContextRollsBackPhantomFillCascade pins the
// receive-time cancellation boundary: an already-expired clone deadline must
// stop before linking a phantom base can recursively examine its delta children.
// The one extra child is deliberate: this fixture sits immediately above the
// 2,000-artifact cascade cap without turning the regression into a corpus test.
func TestReceiveLinkerCancelledContextRollsBackPhantomFillCascade(t *testing.T) {
	const fanout = 2_001

	r := setupTestRepo(t)
	base := []byte("cancelled phantom base with enough stable bytes for deltas")
	baseUUID := repoHash(r, base)
	baseRID, err := blob.StorePhantom(r.DB(), baseUUID)
	if err != nil {
		t.Fatalf("StorePhantom base: %v", err)
	}

	children := make([]libfossil.FslID, fanout)
	for i := range children {
		child := []byte(fmt.Sprintf("cancelled phantom delta child %04d", i))
		rid, err := blob.StoreDeltaRaw(
			r.DB(),
			repoHash(r, child),
			delta.Create(base, child),
			baseRID,
			nil,
		)
		if err != nil {
			t.Fatalf("StoreDeltaRaw child %d: %v", i, err)
		}
		children[i] = rid
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err = r.WithTx(func(tx *db.Tx) error {
		rid, uuid, err := blob.Store(tx, base)
		if err != nil {
			return fmt.Errorf("fill phantom base: %w", err)
		}
		if rid != baseRID {
			return fmt.Errorf("filled base rid = %d, want original phantom rid %d", rid, baseRID)
		}
		return linker.LinkStored(ctx, tx, rid, uuid, base)
	})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("LinkStored cancellation error = %v, want context.Canceled", err)
	}

	var baseSize int
	if err := r.DB().QueryRow("SELECT size FROM blob WHERE rid=?", baseRID).Scan(&baseSize); err != nil {
		t.Fatalf("query base size after rollback: %v", err)
	}
	if baseSize != -1 {
		t.Fatalf("base size after cancelled LinkStored = %d, want phantom size -1", baseSize)
	}
	var phantomCount int
	if err := r.DB().QueryRow("SELECT count(*) FROM phantom WHERE rid=?", baseRID).Scan(&phantomCount); err != nil {
		t.Fatalf("query phantom after rollback: %v", err)
	}
	if phantomCount != 1 {
		t.Fatalf("phantom rows for base after cancelled LinkStored = %d, want 1", phantomCount)
	}

	for _, rid := range children {
		var terminal int
		if err := r.DB().QueryRow(`
			SELECT
				(SELECT count(*) FROM event WHERE objid=?)
			  + (SELECT count(*) FROM plink WHERE cid=? OR pid=?)
			  + (SELECT count(*) FROM mlink WHERE mid=?)
			  + (SELECT count(*) FROM tagxref WHERE rid=?)
			  + (SELECT count(*) FROM leaf WHERE rid=?)
			  + (SELECT count(*) FROM crosslink_nonartifact WHERE rid=?)`,
			rid, rid, rid, rid, rid, rid, rid,
		).Scan(&terminal); err != nil {
			t.Fatalf("query terminal state for child %d: %v", rid, err)
		}
		if terminal != 0 {
			t.Fatalf("child %d acquired %d terminal derived/nonartifact rows after cancelled LinkStored", rid, terminal)
		}
	}
}
func TestReceiveLinkerForumReplyBeforeRootPreservesReservedThreadReferences(t *testing.T) {
	ctx := context.Background()
	r := setupTestRepo(t)
	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}

	root := &deck.Deck{
		Type: deck.ForumPost,
		H:    "reply-before-root thread",
		U:    deck.User("testuser"),
		W:    []byte("root post arrives after its reply"),
		D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC),
	}
	rootBytes, err := root.Marshal()
	if err != nil {
		t.Fatalf("Marshal root forum post: %v", err)
	}
	rootUUID := repoHash(r, rootBytes)

	reply := &deck.Deck{
		Type: deck.ForumPost,
		G:    rootUUID,
		H:    "reply-before-root thread",
		I:    rootUUID,
		U:    deck.User("testuser"),
		W:    []byte("reply received before its root"),
		D:    time.Date(2024, 1, 15, 10, 1, 0, 0, time.UTC),
	}
	replyBytes, err := reply.Marshal()
	if err != nil {
		t.Fatalf("Marshal reply forum post: %v", err)
	}

	var replyRID libfossil.FslID
	if err := r.WithTx(func(tx *db.Tx) error {
		var replyUUID string
		replyRID, replyUUID, err = blob.Store(tx, replyBytes)
		if err != nil {
			return err
		}
		return linker.LinkStored(ctx, tx, replyRID, replyUUID, replyBytes)
	}); err != nil {
		t.Fatalf("store and link reply: %v", err)
	}

	var reservedRoot, reservedIRT libfossil.FslID
	if err := r.DB().QueryRow(
		"SELECT froot, firt FROM forumpost WHERE fpid=?", replyRID,
	).Scan(&reservedRoot, &reservedIRT); err != nil {
		t.Fatalf("query reply forum references before root: %v", err)
	}
	if reservedRoot <= 0 || reservedIRT <= 0 {
		t.Fatalf("reply references before root = froot %d, firt %d; want positive reserved phantom RIDs",
			reservedRoot, reservedIRT)
	}
	if reservedRoot != reservedIRT {
		t.Fatalf("reply references before root = froot %d, firt %d; want both references to the root phantom",
			reservedRoot, reservedIRT)
	}
	phantomRootRID, exists := blob.Exists(r.DB(), rootUUID)
	if !exists {
		t.Fatal("reply link did not reserve a phantom for its root UUID")
	}
	if reservedRoot != phantomRootRID {
		t.Fatalf("reply froot before root = %d, want root phantom RID %d", reservedRoot, phantomRootRID)
	}

	var rootRID libfossil.FslID
	if err := r.WithTx(func(tx *db.Tx) error {
		var ok bool
		rootRID, ok = blob.Exists(tx, rootUUID)
		if !ok {
			return fmt.Errorf("root phantom %s disappeared before arrival", rootUUID)
		}
		compressed, err := blob.EncodeForStorage(rootBytes, nil)
		if err != nil {
			return err
		}
		if _, err := tx.Exec("UPDATE blob SET size=?, content=?, rcvid=1 WHERE rid=?",
			len(rootBytes), compressed, rootRID); err != nil {
			return err
		}
		if _, err := tx.Exec("DELETE FROM phantom WHERE rid=?", rootRID); err != nil {
			return err
		}
		if _, err := tx.Exec("INSERT OR IGNORE INTO unclustered(rid) VALUES(?)", rootRID); err != nil {
			return err
		}
		return linker.LinkStored(ctx, tx, rootRID, rootUUID, rootBytes)
	}); err != nil {
		t.Fatalf("store root into phantom and link: %v", err)
	}
	if rootRID != phantomRootRID {
		t.Fatalf("root filled RID = %d, want reserved phantom RID %d", rootRID, phantomRootRID)
	}

	assertReplyThread := func(stage string) {
		t.Helper()
		var froot, firt libfossil.FslID
		if err := r.DB().QueryRow(
			"SELECT froot, firt FROM forumpost WHERE fpid=?", replyRID,
		).Scan(&froot, &firt); err != nil {
			t.Fatalf("%s: query reply forum references: %v", stage, err)
		}
		if froot != rootRID || firt != rootRID {
			t.Fatalf("%s: reply references = froot %d, firt %d; want root RID %d",
				stage, froot, firt, rootRID)
		}
	}
	assertReplyThread("after root arrival")

	if _, err := linker.Finalize(ctx); err != nil {
		t.Fatalf("ReceiveLinker.Finalize: %v", err)
	}
	assertReplyThread("after Finalize")

	if _, err := Crosslink(r); err != nil {
		t.Fatalf("Crosslink after ReceiveLinker.Finalize: %v", err)
	}
	assertReplyThread("after subsequent Crosslink")
}

// TestReceiveLinkerLinkStoredBoundsUncancelledCascadeAndFinalizesRemainder
// pins the receive-time hard budget independently of cancellation. One phantom
// source unblocks slightly more real checkin deltas than the budget permits:
// LinkStored must return after examining at most receiveCascadeLimit artifacts,
// with the accepted source included, and leave the durable remainder for
// Finalize rather than silently dropping it.
func TestReceiveLinkerLinkStoredBoundsUncancelledCascadeAndFinalizesRemainder(t *testing.T) {
	const fanout = receiveCascadeLimit + 1

	r := setupTestRepo(t)

	_, fileUUID, err := blob.Store(r.DB(), []byte("receive cascade budget file"))
	if err != nil {
		t.Fatalf("Store file blob: %v", err)
	}
	marshal := func(i int, parent string) []byte {
		d := &deck.Deck{
			Type: deck.Checkin,
			C:    fmt.Sprintf("receive cascade budget checkin %d", i),
			U:    deck.User("testuser"),
			D:    time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC).Add(time.Duration(i) * time.Minute),
			P:    nil,
			F:    []deck.FileCard{{Name: "receive-cascade-budget.txt", UUID: fileUUID}},
		}
		if parent != "" {
			d.P = []string{parent}
		}
		b, err := d.Marshal()
		if err != nil {
			t.Fatalf("Marshal checkin %d: %v", i, err)
		}
		return b
	}

	source := marshal(0, "")
	sourceUUID := repoHash(r, source)
	sourceRID, err := blob.StorePhantom(r.DB(), sourceUUID)
	if err != nil {
		t.Fatalf("StorePhantom source: %v", err)
	}

	children := make([]libfossil.FslID, fanout)
	for i := range children {
		child := marshal(i+1, sourceUUID)
		rid, err := blob.StoreDeltaRaw(
			r.DB(),
			repoHash(r, child),
			delta.Create(source, child),
			sourceRID,
			nil,
		)
		if err != nil {
			t.Fatalf("StoreDeltaRaw child %d: %v", i, err)
		}
		children[i] = rid
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	if err := r.WithTx(func(tx *db.Tx) error {
		rid, uuid, err := blob.Store(tx, source)
		if err != nil {
			return fmt.Errorf("fill phantom source: %w", err)
		}
		if rid != sourceRID {
			return fmt.Errorf("filled source rid = %d, want phantom rid %d", rid, sourceRID)
		}
		return linker.LinkStored(context.Background(), tx, rid, uuid, source)
	}); err != nil {
		t.Fatalf("fill and LinkStored source: %v", err)
	}

	if got := linker.Stats().Linked; got == 0 {
		t.Fatal("LinkStored did not link the accepted source")
	} else if got > receiveCascadeLimit {
		t.Fatalf("LinkStored linked %d artifacts, want at most hard receive cascade budget %d "+
			"including the accepted source (issue #214)", got, receiveCascadeLimit)
	}

	if _, err := linker.Finalize(context.Background()); err != nil {
		t.Fatalf("ReceiveLinker.Finalize: %v", err)
	}
	if got := countCrosslinked(t, r, children); got != len(children) {
		t.Fatalf("Finalize crosslinked %d of %d residual delta children, want all", got, len(children))
	}

	if got, err := Crosslink(r); err != nil {
		t.Fatalf("Crosslink after ReceiveLinker.Finalize: %v", err)
	} else if got != 0 {
		t.Fatalf("Crosslink after ReceiveLinker.Finalize linked %d artifacts, want no-op", got)
	}
}

func TestReceiveLinkerLinkStoredPreservesCrosslinkErrorAfterRollback(t *testing.T) {
	const forcedRollback = "forced crosslink rollback"

	r := setupTestRepo(t)
	eventTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	d := &deck.Deck{
		Type: deck.Event,
		E: &deck.EventCard{
			Date: eventTime,
			UUID: "fedcba9876543210fedcba9876543210fedcba98",
		},
		U: deck.User("testuser"),
		D: eventTime,
		W: []byte("forced receive linker rollback event body"),
		C: "forced receive linker rollback event",
	}
	eventBytes, err := d.Marshal()
	if err != nil {
		t.Fatalf("Marshal event: %v", err)
	}
	eventRID, eventUUID, err := blob.Store(r.DB(), eventBytes)
	if err != nil {
		t.Fatalf("Store event blob: %v", err)
	}
	if _, err := r.DB().Exec(`
		CREATE TRIGGER force_receive_linker_rollback
		BEFORE INSERT ON event
		BEGIN
			SELECT RAISE(ROLLBACK, 'forced crosslink rollback');
		END
	`); err != nil {
		t.Fatalf("create rollback trigger: %v", err)
	}

	linker, err := NewReceiveLinker(r)
	if err != nil {
		t.Fatalf("NewReceiveLinker: %v", err)
	}
	err = r.WithTx(func(tx *db.Tx) error {
		return linker.LinkStored(context.Background(), tx, eventRID, eventUUID, eventBytes)
	})
	if err == nil {
		t.Fatal("LinkStored error = nil, want crosslink rollback error")
	}
	if !strings.Contains(err.Error(), forcedRollback) {
		t.Errorf("LinkStored error = %q, want initiating crosslink error %q", err, forcedRollback)
	}
	if !strings.Contains(err.Error(), "manifest.ReceiveLinker rollback") {
		t.Errorf("LinkStored error = %q, want receive_artifact rollback context", err)
	}
}