Skip to content
go-libfossil Artifact [583e4e484a]
Login

Artifact [583e4e484a]

Check-ins Using Download Hex

Artifact 583e4e484af730aa6dfa4a55e49af3e45d2139564f7ec01b0868f6922aead152:

package manifest

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"

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

// AfterDephantomize crosslinks a formerly-phantom blob and any dependents.
// Matches Fossil's after_dephantomize (content.c:389-456).
//
// ctx bounds the graph walk: this cascade runs synchronously mid-round as a
// clone fills phantoms (see sync.storeReceivedFile), and on a large delta/orphan
// graph can run for many multiples of a clone's deadline. A cancelled ctx aborts
// the walk within crosslinkCancelCheckStride steps (see afterDephantomize),
// mirroring the whole-repository sweep's CrosslinkContext contract. Callers with
// no deadline pass context.Background().
func AfterDephantomize(ctx context.Context, r *repo.Repo, rid libfossil.FslID) {
	if r == nil {
		panic("manifest.AfterDephantomize: r must not be nil")
	}
	if rid <= 0 {
		return
	}
	afterDephantomize(ctx, r, rid, true)
}

// AfterDephantomizeBatch runs AfterDephantomize's cascade for every rid in
// one linker, so a sync round that fills many phantoms commits its crosslinks
// together rather than in one transaction per artifact. A cancelled ctx stops
// it as it stops AfterDephantomize.
func AfterDephantomizeBatch(ctx context.Context, r *repo.Repo, rids []libfossil.FslID) {
	if r == nil {
		panic("manifest.AfterDephantomizeBatch: r must not be nil")
	}
	if len(rids) == 0 {
		return
	}
	cl := newCascadeLinker(r)
	for _, rid := range rids {
		if rid > 0 && !cl.walk(ctx, rid, true) {
			return
		}
	}
	cl.flush(ctx)
	logDeferredCheckinSummary("AfterDephantomizeBatch", cl.guard, cl.stats.linked)
}

// afterDephantomize walks the orphan/delta-child graph reachable from rid and
// crosslinks every artifact it finds, returning how that run executed.
//
// The stats are the batching and cache-sharing guarantees made assertable: they
// exist so a test can pin "one commit, not one per artifact" and "the shared
// cache absorbed the repeated parent-manifest expansions" without a package
// hook. AfterDephantomize discards them.
func afterDephantomize(ctx context.Context, r *repo.Repo, rid libfossil.FslID, linkFlag bool) cascadeStats {
	cl := newCascadeLinker(r)
	if cl.walk(ctx, rid, linkFlag) {
		cl.flush(ctx)
	}
	cache := cl.cache.Stats()
	cl.stats.expansions = int(cache.Misses)
	cl.stats.cacheHits = int(cache.Hits)
	logDeferredCheckinSummary("AfterDephantomize", cl.guard, cl.stats.linked)
	return cl.stats
}

// ReceiveLinkStats reports the receive-scoped content work. ExpansionMisses
// counts only durable-delta expansions; supplied full bytes are linked directly
// and therefore do not take a cache miss.
type ReceiveLinkStats struct {
	ExpansionMisses int
	ExpansionHits   int
	Linked          int
}

const (
	receiveSavepoint    = "receive_artifact"
	receiveWaiterLimit  = 8192
	receiveCascadeLimit = 2000
	receiveCacheBytes   = 64 << 20
	receiveSettledLimit = 100_000
)

// ReceiveLinker incrementally derives manifest state while a clone receives
// blobs. It owns one bounded expanded-content cache, one bounded reverse waiter
// index, and one fixed-cap settled RID set through Finalize, when it runs the
// residual candidate sweep and the single combined leaf/tag repair gate.
type ReceiveLinker struct {
	r     *repo.Repo
	state *linkState

	// waiters maps a newly available UUID to the deferred checkins that need it.
	// waitingOn is the reverse index needed to replace a checkin's wait set when
	// another retry discovers a different missing reference.
	waiters     map[string]map[libfossil.FslID]struct{}
	waitingOn   map[libfossil.FslID]map[string]struct{}
	waiterSize  int
	waiterFull  bool
	needsRepair bool
	inFlight    map[libfossil.FslID]struct{}

	// settled keeps terminal receive-session verdicts without consulting
	// derived tables on every delta child. Its fixed cap is a performance
	// optimization, not a correctness boundary: once full, parsed artifacts
	// consult type-appropriate durable replay evidence before derivation.
	settled map[libfossil.FslID]struct{}
}

// NewReceiveLinker creates the receive-scoped linker and its on-demand
// crosslink tables. It is not safe for concurrent use; one clone owns one.
func NewReceiveLinker(r *repo.Repo) (*ReceiveLinker, error) {
	if r == nil {
		panic("manifest.NewReceiveLinker: r must not be nil")
	}
	return newReceiveLinker(r, receiveCacheBytes)
}

// newReceiveLinker creates a linker with an explicit expanded-content budget.
// CrosslinkContext uses the rebuild budget while incremental receive sessions
// use receiveCacheBytes.
func newReceiveLinker(r *repo.Repo, cacheBytes int64) (*ReceiveLinker, error) {
	if r == nil {
		panic("manifest.newReceiveLinker: r must not be nil")
	}
	if cacheBytes <= 0 {
		panic("manifest.newReceiveLinker: cacheBytes must be > 0")
	}
	if err := prepareCrosslinkTables(r.DB()); err != nil {
		return nil, err
	}
	return &ReceiveLinker{
		r:         r,
		state:     newLinkState(cacheBytes),
		waiters:   make(map[string]map[libfossil.FslID]struct{}),
		waitingOn: make(map[libfossil.FslID]map[string]struct{}),
		inFlight:  make(map[libfossil.FslID]struct{}),
		settled:   make(map[libfossil.FslID]struct{}),
	}, nil
}

// LinkStored examines and links one blob inside the transaction that stored it.
// verifiedFullBytes is supplied only for already hash-verified full content;
// nil means the durable delta must be expanded through the receive cache.
//
// One accepted blob may make a large delta or deferred-checkin graph eligible,
// so its synchronous receive-time work is bounded. Residual work is durable:
// Finalize retries the waiters and sweeps every still-unlinked candidate.
func (rl *ReceiveLinker) LinkStored(ctx context.Context, tx *db.Tx, rid libfossil.FslID, uuid string, verifiedFullBytes []byte) error {
	if rl == nil {
		panic("manifest.ReceiveLinker.LinkStored: linker must not be nil")
	}
	if tx == nil {
		panic("manifest.ReceiveLinker.LinkStored: tx must not be nil")
	}
	if rid <= 0 {
		panic("manifest.ReceiveLinker.LinkStored: rid must be positive")
	}
	if err := ctx.Err(); err != nil {
		return err
	}

	// A blob arrived since the previous checkin examination. Availability
	// negatives are deliberately scoped to one accepted blob so a formerly
	// missing UUID cannot keep a waiter deferred after this transaction fills it.
	rl.state.guard.avail = content.NewAvailabilityCache()
	remaining := receiveCascadeLimit
	if err := rl.linkStored(ctx, tx, rid, verifiedFullBytes, &remaining); err != nil {
		return err
	}
	if uuid == "" || remaining == 0 {
		return nil
	}
	return rl.retryWaitersFor(ctx, tx, uuid, &remaining)
}

// Finalize retries the bounded deferred set, sweeps only residual candidates
// with the same content cache, and repairs leaf/tag state exactly once. Its
// return count is only the final-sweep work, not receive-time links.
func (rl *ReceiveLinker) Finalize(ctx context.Context) (int, error) {
	if rl == nil {
		panic("manifest.ReceiveLinker.Finalize: linker must not be nil")
	}
	if err := ctx.Err(); err != nil {
		_, repairErr := repairCrosslinkState(rl.r, rl.state.linked, err)
		return 0, repairErr
	}
	if err := rl.retryAllWaiters(ctx); err != nil {
		// A cancellation or savepoint-control failure stops new derivation, but
		// receive-time batches may already have committed plink/tagxref rows.
		// They still get the same single repair gate before the cause returns.
		_, repairErr := repairCrosslinkState(rl.r, rl.state.linked, err)
		return 0, repairErr
	}
	linked, sweepErr := crosslinkSweepWithState(ctx, rl.r, rl.state)
	if linked > 0 {
		rl.needsRepair = true
	}
	if !rl.needsRepair {
		return linked, sweepErr
	}
	_, repairErr := repairCrosslinkState(rl.r, 1, sweepErr)
	if repairErr == nil {
		rl.needsRepair = false
	}
	return linked, repairErr
}

// Stats returns a snapshot of receive-time linking and bounded-cache work.
func (rl *ReceiveLinker) Stats() ReceiveLinkStats {
	if rl == nil {
		return ReceiveLinkStats{}
	}
	cache := rl.state.cache.Stats()
	return ReceiveLinkStats{
		ExpansionMisses: int(cache.Misses),
		ExpansionHits:   int(cache.Hits),
		Linked:          rl.state.linked,
	}
}

// linkStored examines one durable blob, using supplied full bytes before it
// touches the cache. Link failures roll back only this artifact and remain
// candidates for Finalize; savepoint-control failures return to abort the
// caller's outer storage transaction.
//
// remaining is shared by a top-level receive call and counts only candidates
// that were neither already settled nor already on the active recursion path.
func (rl *ReceiveLinker) linkStored(ctx context.Context, tx *db.Tx, rid libfossil.FslID, verifiedFullBytes []byte, remaining *int) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	if _, settled := rl.settled[rid]; settled {
		return nil
	}
	if _, busy := rl.inFlight[rid]; busy {
		return nil
	}
	if *remaining == 0 {
		return nil
	}
	(*remaining)--
	rl.inFlight[rid] = struct{}{}
	defer delete(rl.inFlight, rid)

	data := verifiedFullBytes
	direct := data != nil
	available := false
	defer func() {
		if direct {
			// Parsing/linking consumes these exact verified bytes first. An
			// over-budget entry is only eligible for eviction afterwards.
			rl.state.cache.Remember(rid, data)
		}
		if err != nil || !available {
			return
		}
		if ctxErr := ctx.Err(); ctxErr != nil {
			err = ctxErr
			return
		}
		if *remaining != 0 {
			err = rl.retryDeltaChildren(ctx, tx, rid, remaining)
		}
	}()
	if !direct {
		data, err = rl.state.cache.Expand(tx, rid)
		if err != nil {
			return nil // not grounded yet; a later source arrival retries it
		}
	}
	available = true

	d, err := deck.Parse(data)
	if err != nil || !isArtifact(d) {
		err = recordNonArtifacts(tx, []libfossil.FslID{rid})
		if err == nil {
			rl.settle(rid)
		}
		return err
	}
	if len(rl.settled) >= receiveSettledLimit {
		replayed, replayErr := parsedArtifactReplayed(tx, rid, d)
		if replayErr != nil {
			return replayErr
		}
		if replayed {
			return nil
		}
	}

	if d.Type == deck.Checkin {
		missing := missingCheckinRefs(tx, d, rl.state.guard.avail)
		if len(missing) != 0 {
			rl.state.guard.recordDeferred(tx, "ReceiveLinker", rid, missing)
			rl.enqueueWaiter(rid, missing)
			return nil
		}
	}

	if _, err := tx.Exec("SAVEPOINT " + receiveSavepoint); err != nil {
		return fmt.Errorf("manifest.ReceiveLinker savepoint rid=%d: %w", rid, err)
	}
	handled, linkErr := linkArtifact(tx, rid, d, rl.state.cache)
	if linkErr != nil {
		if _, err := tx.Exec("ROLLBACK TO " + receiveSavepoint); err != nil {
			return errors.Join(
				fmt.Errorf("manifest.ReceiveLinker crosslink rid=%d: %w", rid, linkErr),
				fmt.Errorf("manifest.ReceiveLinker rollback rid=%d: %w", rid, err),
			)
		}
		if _, err := tx.Exec("RELEASE " + receiveSavepoint); err != nil {
			return errors.Join(
				fmt.Errorf("manifest.ReceiveLinker crosslink rid=%d: %w", rid, linkErr),
				fmt.Errorf("manifest.ReceiveLinker release rid=%d: %w", rid, err),
			)
		}
		slog.Warn("manifest.ReceiveLinker: crosslink failed, blob left un-crosslinked",
			"rid", rid, "type", d.Type, "error", linkErr)
		return nil
	}
	if _, err := tx.Exec("RELEASE " + receiveSavepoint); err != nil {
		return fmt.Errorf("manifest.ReceiveLinker release rid=%d: %w", rid, err)
	}
	if handled {
		rl.state.linked++
		rl.needsRepair = true
		rl.settle(rid)
	}
	return nil
}

func (rl *ReceiveLinker) settle(rid libfossil.FslID) {
	if len(rl.settled) >= receiveSettledLimit {
		return
	}
	rl.settled[rid] = struct{}{}
}

// parsedArtifactReplayed reports whether a parsed artifact has the durable
// output that proves a prior receive attempt committed. Event and Cluster
// artifacts use their tagxref source rows as replay evidence; every other
// handled artifact type uses its event row.
func parsedArtifactReplayed(tx *db.Tx, rid libfossil.FslID, d *deck.Deck) (bool, error) {
	table, column := "event", "objid"
	if d.Type == deck.Event || d.Type == deck.Cluster {
		table, column = "tagxref", "srcid"
	}
	var replayed int
	if err := tx.QueryRow(
		"SELECT EXISTS(SELECT 1 FROM "+table+" WHERE "+column+"=?)",
		rid,
	).Scan(&replayed); err != nil {
		return false, fmt.Errorf("manifest.ReceiveLinker replay rid=%d: %w", rid, err)
	}
	return replayed != 0, nil
}

// retryDeltaChildren follows the durable delta graph from an accepted source.
// Its query is deliberately just the indexed delta source lookup joined to real
// blobs: receive-session settled RIDs avoid repeated work without materializing
// or scanning any derived table for each source. inFlight still breaks cycles,
// and Finalize's candidate sweep remains the durable correctness backstop.
func (rl *ReceiveLinker) retryDeltaChildren(ctx context.Context, tx *db.Tx, srcid libfossil.FslID, remaining *int) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if *remaining == 0 {
		return nil
	}
	rows, err := tx.Query(`
		SELECT d.rid
		  FROM delta d JOIN blob b ON b.rid=d.rid
		 WHERE d.srcid=?
		   AND b.size>=0
		 ORDER BY d.rid
		 LIMIT ?`, srcid, *remaining)
	if err != nil {
		return fmt.Errorf("manifest.ReceiveLinker delta children rid=%d: %w", srcid, err)
	}
	defer rows.Close()
	children := make([]libfossil.FslID, 0, *remaining)
	for i := 0; rows.Next(); i++ {
		if i%crosslinkCancelCheckStride == 0 {
			if err := ctx.Err(); err != nil {
				return err
			}
		}
		var rid int64
		if err := rows.Scan(&rid); err != nil {
			return fmt.Errorf("manifest.ReceiveLinker delta child rid=%d: %w", srcid, err)
		}
		children = append(children, libfossil.FslID(rid))
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("manifest.ReceiveLinker delta children rid=%d: %w", srcid, err)
	}
	for i, rid := range children {
		if *remaining == 0 {
			return nil
		}
		if i%crosslinkCancelCheckStride == 0 {
			if err := ctx.Err(); err != nil {
				return err
			}
		}
		if err := rl.linkStored(ctx, tx, rid, nil, remaining); err != nil {
			return err
		}
	}
	return nil
}

func (rl *ReceiveLinker) enqueueWaiter(rid libfossil.FslID, missing []string) {
	rl.removeWaiter(rid)
	if rl.waiterFull {
		return
	}
	for _, uuid := range missing {
		if rl.waiterSize == receiveWaiterLimit {
			rl.waiterFull = true
			return
		}
		byRID := rl.waiters[uuid]
		if byRID == nil {
			byRID = make(map[libfossil.FslID]struct{})
			rl.waiters[uuid] = byRID
		}
		if _, exists := byRID[rid]; exists {
			continue
		}
		byRID[rid] = struct{}{}
		byUUID := rl.waitingOn[rid]
		if byUUID == nil {
			byUUID = make(map[string]struct{})
			rl.waitingOn[rid] = byUUID
		}
		byUUID[uuid] = struct{}{}
		rl.waiterSize++
	}
}

func (rl *ReceiveLinker) removeWaiter(rid libfossil.FslID) {
	for uuid := range rl.waitingOn[rid] {
		byRID := rl.waiters[uuid]
		delete(byRID, rid)
		rl.waiterSize--
		if len(byRID) == 0 {
			delete(rl.waiters, uuid)
		}
	}
	delete(rl.waitingOn, rid)
}

func (rl *ReceiveLinker) retryWaitersFor(ctx context.Context, tx *db.Tx, uuid string, remaining *int) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if *remaining == 0 {
		return nil
	}
	byRID := rl.waiters[uuid]
	if len(byRID) == 0 {
		return nil
	}
	rids := make([]libfossil.FslID, 0, len(byRID))
	for rid := range byRID {
		rids = append(rids, rid)
	}
	sort.Slice(rids, func(i, j int) bool { return rids[i] < rids[j] })
	for i, rid := range rids {
		if *remaining == 0 {
			return nil
		}
		if i%crosslinkCancelCheckStride == 0 {
			if err := ctx.Err(); err != nil {
				return err
			}
		}
		rl.removeWaiter(rid)
		if err := rl.linkStored(ctx, tx, rid, nil, remaining); err != nil {
			return err
		}
	}
	return nil
}

func (rl *ReceiveLinker) retryAllWaiters(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if len(rl.waitingOn) == 0 {
		return nil
	}
	rids := make([]libfossil.FslID, 0, len(rl.waitingOn))
	for rid := range rl.waitingOn {
		rids = append(rids, rid)
	}
	sort.Slice(rids, func(i, j int) bool { return rids[i] < rids[j] })
	for i, rid := range rids {
		if i%crosslinkCancelCheckStride == 0 {
			if err := ctx.Err(); err != nil {
				return err
			}
		}
		rl.removeWaiter(rid)
		if err := rl.r.WithTx(func(tx *db.Tx) error {
			remaining := receiveCascadeLimit
			return rl.linkStored(ctx, tx, rid, nil, &remaining)
		}); err != nil {
			return err
		}
	}
	return nil
}

// cascadeStats reports how one cascade executed: how much work it did, how many
// transactions that cost, and how much content expansion the shared cache
// absorbed.
type cascadeStats struct {
	// attempts counts artifacts handed to the linker, duplicates included: an
	// artifact reachable both as an orphan and as a delta child is crosslinked
	// once per visit, as it always was.
	attempts int
	// linked counts artifacts linkArtifact wrote rows for.
	linked int
	// commits counts write transactions this cascade committed. It is the
	// number the batching exists to hold down (issue #172).
	commits int
	// expansions counts content expansions actually materialized, and cacheHits
	// those the cascade's shared cache served instead. A delta base or parent
	// manifest several artifacts need shows up as one expansion and many hits.
	expansions int
	cacheHits  int
}

// cascadeItem is one artifact the walk discovered. baseline is the orphan row's
// baseline rid when the artifact was found as an orphan, and 0 otherwise; it is
// carried only so a failure is logged against the same context it used to be.
type cascadeItem struct {
	rid      libfossil.FslID
	baseline libfossil.FslID
}

// cascadeSavepoint isolates one artifact's writes inside a batch transaction.
// Never nested: the batch loop releases or rolls back before the next artifact.
const cascadeSavepoint = "cascade_artifact"

// cascadeLinker crosslinks the artifacts one phantom-fill cascade discovers,
// with the two savings the whole-repository sweep already takes: a content cache
// shared by every artifact in the cascade, and one commit per crosslinkBatchSize
// artifacts rather than one per artifact. A single phantom fill can unblock
// hundreds or thousands of related artifacts that share delta bases and parent
// manifests, so the per-artifact shape this replaces paid a transaction commit
// and a from-scratch delta-chain expansion for work a sibling had just done
// (issue #172).
//
// It is the cascade's whole crosslink path; the per-artifact helper it replaces
// had no callers outside this file, so nothing else has to keep working the old
// way. Artifact linking itself routes through linkArtifact, the same type switch
// the sweep uses, rather than a third copy of it.
type cascadeLinker struct {
	r     *repo.Repo
	cache *content.Cache

	// guard holds back a Checkin whose referenced blobs are not yet locally
	// available, exactly as the whole-repository sweep's linkState.guard does
	// (see checkinDeferralGuard in crosslink.go). A deferred checkin gets no
	// event/leaf/plink/mlink/tagxref rows from this cascade at all -- that is
	// what lets the sweep's own candidate query, which only reselects rids
	// with no event row, rediscover and complete it once the missing blob
	// arrives. The cascade has no candidate query or round boundary of its
	// own to retry from, so it depends entirely on falling through to the
	// sweep's recovery path; writing so much as the event row here would
	// strand the checkin permanently (issue #180).
	guard *checkinDeferralGuard

	// pending holds discovered artifacts whose writes have not been committed.
	// The walk discovers work as it goes, so there is no candidate slice to cut
	// into batches the way linkCandidatesInOrder does: the buffer flushes when it
	// fills, and once more when the walk finishes.
	pending []cascadeItem

	stats cascadeStats
}

func newCascadeLinker(r *repo.Repo) *cascadeLinker {
	return &cascadeLinker{
		r:     r,
		cache: content.NewCache(crosslinkCacheBytes),
		guard: newCheckinDeferralGuard(),
	}
}

// walk is the work-stack traversal of the orphan/delta-child graph reachable
// from rid. It reports whether it ran to completion; false means ctx was
// cancelled, in which case the buffered batch is deliberately dropped rather
// than flushed. Dropping costs nothing -- the buffer holds discovered rids, not
// finished work -- and flushing it would spend up to a full batch of crosslinks
// after the deadline already fired, which is the overshoot issue #166 closed.
// Committed batches stand; the artifacts still buffered are simply not
// crosslinked, and the next sync's Crosslink sweep re-selects them because they
// have no event row.
func (cl *cascadeLinker) walk(ctx context.Context, rid libfossil.FslID, linkFlag bool) bool {
	// Work stack replaces recursion. Bounded by total blob count in repo.
	type workItem struct {
		rid      libfossil.FslID
		linkFlag bool
	}
	stack := []workItem{{rid: rid, linkFlag: linkFlag}}

	const maxIterations = 1_000_000 // Guard against pathological delta chains.
	iterations := 0

	for len(stack) > 0 {
		// Observe the deadline every crosslinkCancelCheckStride pops, mirroring
		// linkBatch's stride-batched poll in the whole-repository sweep. This
		// cascade runs synchronously mid-round as a clone fills phantoms (see
		// sync.storeReceivedFile), so without this a cancelled clone context
		// could not interrupt it until the walk finished or hit maxIterations --
		// the deadline-blind overshoot of issue #166. On cancellation we return
		// what has been crosslinked so far, exactly as the batch sweep abandons
		// its remaining candidates; nothing pretends the whole cascade completed.
		if iterations%crosslinkCancelCheckStride == 0 {
			select {
			case <-ctx.Done():
				return false
			default:
			}
		}
		iterations++
		if iterations > maxIterations {
			return true // Safety bound exceeded; commit what is buffered.
		}

		// Pop from stack.
		item := stack[len(stack)-1]
		stack = stack[:len(stack)-1]
		current := item.rid

		if current <= 0 {
			continue
		}

		if item.linkFlag {
			if !cl.add(ctx, cascadeItem{rid: current}) {
				return false
			}
		}

		// Process orphaned delta manifests whose baseline is this rid.
		orphanRows, err := cl.r.DB().Query("SELECT rid FROM orphan WHERE baseline=?", current)
		if err == nil {
			var orphans []libfossil.FslID
			for orphanRows.Next() {
				var orid int64
				if orphanRows.Scan(&orid) == nil {
					orphans = append(orphans, libfossil.FslID(orid))
				}
			}
			orphanRows.Close()
			for _, orid := range orphans {
				if !cl.add(ctx, cascadeItem{rid: orid, baseline: current}) {
					return false
				}
			}
			if len(orphans) > 0 {
				// Orphans are the one discovery the delta-child query below can
				// observe: an orphan is reached by its baseline, not by the delta
				// graph, so the same rid can also be a delta child of current, and
				// the query filters exactly on the mlink rows crosslinking it
				// writes. Every other buffered artifact was itself discovered by
				// that query -- delta.rid is a primary key, so an artifact has one
				// delta base and is queried for once -- and so cannot come back.
				// Committing here restores the read-after-write the per-artifact
				// transactions gave the query for free, at one commit per
				// orphan-bearing node rather than one per artifact. It happens
				// before the DELETE below, keeping the original order: orphans
				// crosslinked, orphan rows cleared, children queried.
				if !cl.flush(ctx) {
					return false
				}
				// Runs on r.DB() in autocommit, as it always did: no batch
				// transaction is open between flushes, and no crosslink path
				// writes 'orphan', so batching cannot reorder this against the
				// query above.
				if _, err := cl.r.DB().Exec("DELETE FROM orphan WHERE baseline=?", current); err != nil {
					continue
				}
			}
		}

		// Find delta children not yet crosslinked. The mlink filter reads
		// committed rows, so buffering defers when a just-linked artifact starts
		// filtering itself out. That only matters for a cycle in the delta base
		// relation, which fossil does not produce and content.Expand would not
		// survive: such a cycle now spins until the buffer flushes at
		// crosslinkBatchSize rather than dying after two pops. Bounded,
		// idempotent, and still capped by maxIterations.
		childRows, err := cl.r.DB().Query(
			`SELECT rid FROM delta WHERE srcid=? AND NOT EXISTS (SELECT 1 FROM mlink WHERE mid=delta.rid)`, current)
		if err != nil {
			continue
		}
		var children []libfossil.FslID
		for childRows.Next() {
			var crid int64
			if childRows.Scan(&crid) == nil {
				children = append(children, libfossil.FslID(crid))
			}
		}
		childRows.Close()

		// Push all children onto work stack (reverse order for LIFO processing).
		for i := len(children) - 1; i >= 0; i-- {
			stack = append(stack, workItem{rid: children[i], linkFlag: true})
		}
	}
	return true
}

// add buffers one artifact, flushing once the buffer holds a full batch. It
// reports whether the cascade may continue; false means ctx was cancelled
// during the flush.
func (cl *cascadeLinker) add(ctx context.Context, item cascadeItem) bool {
	cl.pending = append(cl.pending, item)
	if len(cl.pending) < crosslinkBatchSize {
		return true
	}
	return cl.flush(ctx)
}

// flush crosslinks the buffered artifacts in one transaction and commits it,
// reporting whether the cascade may continue.
//
// Failure handling matches the per-artifact transactions this replaces. An
// artifact that cannot be expanded, parsed, or linked is logged and skipped
// while the rest of the batch still commits, so one bad blob no longer costs a
// batch of unrelated successful work; and each artifact's writes are wrapped in
// a savepoint, so a link that fails partway leaves no partial
// event/plink/mlink/tagxref behind -- the atomicity its own transaction used to
// give it. A commit that fails loses the batch instead of one artifact, which is
// the one place batching widens the blast radius: crosslink output is derived,
// the blobs stay durable, and the next sync's Crosslink sweep re-selects every
// artifact that still has no event row, exactly as it did for an artifact whose
// own transaction failed.
func (cl *cascadeLinker) flush(ctx context.Context) bool {
	if len(cl.pending) == 0 {
		return true
	}
	batch := cl.pending
	cl.pending = nil

	err := cl.r.WithTx(func(tx *db.Tx) error {
		for i, item := range batch {
			// Poll on the same stride the sweep's linkBatch uses, so a deadline
			// interrupts a batch rather than only the walk between batches. The
			// batch rolls back on cancellation, as linkBatch's does.
			if i%crosslinkCancelCheckStride == 0 {
				select {
				case <-ctx.Done():
					return ctx.Err()
				default:
				}
			}
			if err := cl.linkOne(tx, item); err != nil {
				return err
			}
		}
		return nil
	})

	if err != nil {
		if ctx.Err() != nil {
			return false
		}
		slog.Warn("manifest.AfterDephantomize: crosslink batch failed, blobs left un-crosslinked",
			"artifacts", len(batch), "error", err)
		return true
	}
	cl.stats.commits++
	return true
}

// linkOne crosslinks one artifact on the batch transaction, isolated by a
// savepoint so a mid-artifact failure discards only its own writes.
func (cl *cascadeLinker) linkOne(tx *db.Tx, item cascadeItem) error {
	cl.stats.attempts++
	rid := item.rid

	data, err := cl.cache.Expand(tx, rid)
	if err != nil {
		cl.warn(item, fmt.Errorf("expand rid=%d: %w", rid, err))
		return nil
	}
	d, err := deck.Parse(data)
	if err != nil {
		// Not a manifest, so there is nothing to crosslink -- the ordinary
		// outcome for a file blob, which is most of what a clone delivers.
		// The whole-repository sweep has always treated this as a skip (see
		// linkBatch); warning here instead made a clone log one line per file
		// blob it filled, ~39k of them on fossil's own repository, and buried
		// the failures that do matter (issue #186). A blob that parses but
		// cannot be linked is still a real fault and still warns, below.
		return nil
	}
	if !isArtifact(d) {
		// Satisfies the card grammar but not the artifact grammar -- a file
		// whose leading bytes read as cards, or an artifact missing a card its
		// type requires. Fossil refuses both, so linking either would write
		// rows `fossil rebuild` removes. Same skip the sweep applies in
		// linkBatch; see isArtifact.
		return nil
	}

	// Hold back a Checkin referencing a blob that has not arrived yet, same
	// as the whole-repository sweep's linkBatch does before ever calling
	// linkArtifact. Checked before the savepoint opens, not inside it: a
	// deferred checkin must leave nothing in the open batch transaction at
	// all, matching the sweep's defer skipping every row write. See
	// cascadeLinker.guard and checkinDeferralGuard.
	if d.Type == deck.Checkin && cl.guard.shouldDefer(tx, "AfterDephantomize", rid, d) {
		return nil
	}

	if _, err := tx.Exec("SAVEPOINT " + cascadeSavepoint); err != nil {
		return fmt.Errorf("manifest.AfterDephantomize savepoint rid=%d: %w", rid, err)
	}
	handled, linkErr := linkArtifact(tx, rid, d, cl.cache)
	if linkErr != nil {
		if _, err := tx.Exec("ROLLBACK TO " + cascadeSavepoint); err != nil {
			return fmt.Errorf("manifest.AfterDephantomize rollback rid=%d: %w", rid, err)
		}
		if _, err := tx.Exec("RELEASE " + cascadeSavepoint); err != nil {
			return fmt.Errorf("manifest.AfterDephantomize release rid=%d: %w", rid, err)
		}
		cl.warn(item, fmt.Errorf("link rid=%d type=%d: %w", rid, d.Type, linkErr))
		return nil
	}
	if _, err := tx.Exec("RELEASE " + cascadeSavepoint); err != nil {
		return fmt.Errorf("manifest.AfterDephantomize release rid=%d: %w", rid, err)
	}
	if handled {
		cl.stats.linked++
	}
	return nil
}

// warn logs an artifact-level failure, keeping the orphan and non-orphan
// wordings the per-artifact call sites used so existing log consumers still see
// which side of the walk lost the blob.
func (cl *cascadeLinker) warn(item cascadeItem, err error) {
	if item.baseline != 0 {
		slog.Warn("manifest.AfterDephantomize: crosslink of orphan failed, blob left un-crosslinked",
			"rid", item.rid, "baseline", item.baseline, "error", err)
		return
	}
	slog.Warn("manifest.AfterDephantomize: crosslink failed, blob left un-crosslinked",
		"rid", item.rid, "error", err)
}