Skip to content
go-libfossil Artifact [2d4aac6810]
Login

Artifact [2d4aac6810]

Check-ins Using Download Hex

Artifact 2d4aac681079c6b3beec19cd5bde19585e43bb6ee18b8e1723b913ef7473f150:

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().
//
// It returns how many artifacts the cascade linked. Each batch the cascade
// commits runs its own repair gate first: it propagates tags from the parents
// its plink rows name and applies the leaf rechecks it scheduled (issue #432),
// so the session's closing crosslink owes the cascade no repair (issue #414).
func AfterDephantomize(ctx context.Context, r *repo.Repo, rid libfossil.FslID) int {
	if r == nil {
		panic("manifest.AfterDephantomize: r must not be nil")
	}
	if rid <= 0 {
		return 0
	}
	return afterDephantomize(ctx, r, rid, true).linked
}

// 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. It returns how many artifacts it linked.
func AfterDephantomizeBatch(ctx context.Context, r *repo.Repo, rids []libfossil.FslID) int {
	if r == nil {
		panic("manifest.AfterDephantomizeBatch: r must not be nil")
	}
	if len(rids) == 0 {
		return 0
	}
	cl := newCascadeLinker(r)
	for _, rid := range rids {
		if rid > 0 && !cl.walk(ctx, rid, true) {
			return cl.stats.linked
		}
	}
	cl.flush(ctx)
	logDeferredCheckinSummary("AfterDephantomizeBatch", cl.guard, cl.stats.linked)
	return 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 + cache.DerivedHits)
	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. Each of its link transactions runs the leaf/tag
// repair gate before it commits (linkTx).
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
	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.
	// Positives stand for the whole receive, which only adds content: a clone
	// whose round fails stops receiving. Walking every F card's chain afresh
	// for each blob cost a dense clone of the Fossil repository a sixth of its
	// CPU.
	rl.state.guard.avail.ForgetUnavailable()
	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)
}

// WithTx runs fn in a transaction on r, as r.WithTx does, for a caller whose
// fn calls LinkStored. If the transaction fails, the leaf rechecks and the
// parents LinkStored scheduled in it are dropped with it: a rolled-back blob's
// rid can be handed to an unrelated blob later (rid is not AUTOINCREMENT), and
// a recheck left behind would then insert a leaf row for a blob that is no
// check-in.
func (rl *ReceiveLinker) WithTx(r *repo.Repo, fn func(tx *db.Tx) error) error {
	if rl == nil {
		panic("manifest.ReceiveLinker.WithTx: linker must not be nil")
	}
	if r == nil {
		panic("manifest.ReceiveLinker.WithTx: r must not be nil")
	}
	return linkTx(r, rl.state.sched, fn)
}

// Finalize retries the bounded deferred set and sweeps only residual
// candidates with the same content cache, each in a transaction that runs its
// own repair gate. 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 {
		return 0, repairCrosslinkState(rl.r, rl.state.sched, err)
	}
	if err := rl.retryAllWaiters(ctx); err != nil {
		// A cancellation or savepoint-control failure stops new derivation.
		// What already committed was repaired in its own transaction, unless a
		// caller linked outside WithTx; repairCrosslinkState covers that.
		return 0, repairCrosslinkState(rl.r, rl.state.sched, err)
	}
	linked, sweepErr := crosslinkSweepWithState(ctx, rl.r, rl.state)
	return linked, repairCrosslinkState(rl.r, rl.state.sched, sweepErr)
}

// 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 _, waiting := rl.waitingOn[rid]; waiting {
		// A waiter is grounded and parsed already; only the arrival of what it
		// waits on (retryWaitersFor) or Finalize can change its verdict, so a
		// delta source arriving under it is no reason to test its files again.
		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 {
		if err := noteTreeReferences(tx, rid, d); err != nil {
			return err
		}
	}
	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 {
		deferred, err := rl.deferCheckin(tx, rid, d)
		if err != nil || deferred {
			return err
		}
	}

	handled, err := rl.linkInSavepoint(tx, rid, d)
	if err != nil {
		return err
	}
	if handled {
		rl.state.linked++
		rl.settle(rid)
	}
	return nil
}

// linkInSavepoint links one parsed artifact inside a savepoint, so a link that
// fails partway leaves neither rows nor scheduled leaf rechecks behind; the
// failure is logged and the blob stays a candidate for Finalize. Only a
// savepoint-control failure is returned.
func (rl *ReceiveLinker) linkInSavepoint(
	tx *db.Tx, rid libfossil.FslID, d *deck.Deck,
) (handled bool, err error) {
	if _, err := tx.Exec("SAVEPOINT " + receiveSavepoint); err != nil {
		return false, fmt.Errorf("manifest.ReceiveLinker savepoint rid=%d: %w", rid, err)
	}
	mark := rl.state.sched.mark()
	handled, linkErr := linkArtifact(tx, rid, d, rl.state.cache, rl.state.sched)
	if linkErr != nil {
		// The artifact's rows roll back, and so does what they scheduled.
		rl.state.sched.rollback(mark)
		if _, err := tx.Exec("ROLLBACK TO " + receiveSavepoint); err != nil {
			return false, 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 false, 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 false, nil
	}
	if _, err := tx.Exec("RELEASE " + receiveSavepoint); err != nil {
		return false, fmt.Errorf("manifest.ReceiveLinker release rid=%d: %w", rid, err)
	}
	return handled, nil
}

// deferCheckin reports whether checkin rid (deck d) must wait for a blob still
// in flight -- its baseline or a file (missingCheckinRefs) -- recording its
// baseline gap and queuing it as a waiter on each missing blob when so.
//
// A clone into a sparse store waits on no file: the origin's walk sends every
// file the store keeps in the same session, nothing reads the store before
// the clone ends, and most files never arrive. Waiting on them cost a content
// horizon clone of fossil-scm.org's repository its first round, past 20
// minutes. A pull still waits (missingCheckinRefs): its files come only when
// asked for, and a check-in linked first leaves them unrequested.
func (rl *ReceiveLinker) deferCheckin(tx *db.Tx, rid libfossil.FslID, d *deck.Deck) (bool, error) {
	storeCone, err := content.StoreCone(tx)
	if err != nil {
		return false, err
	}
	var missing []string
	if storeCone != nil {
		missing = missingFileSet(tx, d, rl.state.guard)
	} else {
		missing = missingCheckinRefs(tx, d, rl.state.guard)
	}
	if len(missing) == 0 {
		return false, nil
	}
	if err := rl.state.guard.recordDeferred(tx, "ReceiveLinker", rid, d, missing); err != nil {
		return true, err
	}
	rl.enqueueWaiter(rid, missing)
	return true, 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) {
	if d.Type == deck.Checkin {
		var incomplete int
		if err := tx.QueryRow("SELECT EXISTS(SELECT 1 FROM orphan WHERE rid=?)", rid).Scan(&incomplete); err != nil {
			return false, err
		}
		if incomplete != 0 {
			return false, nil
		}
	}
	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
			}
		}
		if rl.strikeWait(rid, uuid) {
			continue
		}
		rl.removeWaiter(rid)
		if err := rl.linkStored(ctx, tx, rid, nil, remaining); err != nil {
			return err
		}
	}
	return nil
}

// strikeWait records that uuid, one of the references waiter rid waits on,
// has arrived, and reports whether rid still waits on another. A check-in is
// examined again only once its whole wait set has arrived: testing every
// F card again at each arrival made a check-in cost one availability query
// per file per arrival, and a sparse store, whose out-of-cone files never
// arrive, left a clone of fossil-scm.org's repository in its first round
// after 20 minutes. A wait set the waiter limit cut short empties early,
// which costs a premature examination, never a missed one.
func (rl *ReceiveLinker) strikeWait(rid libfossil.FslID, uuid string) bool {
	byUUID := rl.waitingOn[rid]
	if _, ok := byUUID[uuid]; !ok {
		panic("manifest.ReceiveLinker.strikeWait: rid does not wait on uuid")
	}
	if len(byUUID) == 1 {
		return false
	}
	delete(byUUID, uuid)
	byRID := rl.waiters[uuid]
	delete(byRID, rid)
	if len(byRID) == 0 {
		delete(rl.waiters, uuid)
	}
	rl.waiterSize--
	return true
}

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 := linkTx(rl.r, rl.state.sched, 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;
	// a parsed form kept beside the cached bytes counts as a hit too.
	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 baseline or files are not yet locally
	// available (see checkinDeferralGuard and missingCheckinRefs in
	// crosslink.go). A deferred checkin gets at most its baseline gap record
	// (the baseline's phantom and an orphan row) and 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. 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

	// sched is the schedule of the batch being flushed. Each flush is its own
	// transaction and runs its own repair gate before it commits: it
	// propagates tags from the parents its artifacts' plink rows name, as
	// fossil's manifest_crosslink does as it links each one, and applies the
	// leaf rechecks its branch rows and that propagation schedule, as fossil
	// does at the end of each transaction (src/db.c, leaf_do_pending_checks).
	// A batch's parents are known only to the batch, so the gate cannot wait
	// for the session's (issue #432).
	sched *batchSchedule

	stats cascadeStats
}

func newCascadeLinker(r *repo.Repo) *cascadeLinker {
	guard := newCheckinDeferralGuard()
	// The cascade runs mid-session, with more blobs on the way, so it waits
	// on files as well as the baseline; the session's closing sweep links
	// whatever is still waiting (see missingCheckinRefs).
	guard.awaitFiles = true
	cl := &cascadeLinker{
		r:     r,
		cache: content.NewCache(crosslinkCacheBytes),
		guard: guard,
		sched: newBatchSchedule(),
	}
	guard.cache = cl.cache
	return cl
}

// 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. A deferral does write
				// 'orphan', but only under a baseline that is still absent, so
				// never under current, which has just arrived: 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.
		// A shunned child is never linked (candidateQuerySQL), nor walked past.
		childRows, err := cl.r.DB().Query(
			`SELECT rid FROM delta WHERE srcid=? AND NOT EXISTS (SELECT 1 FROM mlink WHERE mid=delta.rid)
			AND NOT EXISTS (SELECT 1 FROM shun JOIN blob ON blob.uuid=shun.uuid WHERE blob.rid=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 repairOnTx(tx, cl.sched)
	})
	// Committed or rolled back, the batch's schedule is spent: a rolled-back
	// batch's rows are gone, and the sweep that relinks its artifacts
	// schedules them again.
	cl.sched.reset()

	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 err := noteTreeReferences(tx, rid, d); err != nil {
		return fmt.Errorf("manifest.AfterDephantomize rid=%d: %w", rid, err)
	}
	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 whose baseline or files have not arrived yet, as
	// the whole-repository sweep's linkBatch holds one for its baseline
	// before ever calling linkArtifact. Checked before the savepoint opens,
	// not inside it: a deferred checkin leaves only its gap record in the
	// batch transaction, matching the sweep's defer skipping every
	// derived-row write. A failed gap record is a database fault, not an
	// artifact's, so it fails the batch. See cascadeLinker.guard and
	// checkinDeferralGuard.
	if d.Type == deck.Checkin {
		deferred, err := cl.guard.shouldDefer(tx, "AfterDephantomize", rid, d)
		if err != nil || deferred {
			return err
		}
	}

	if _, err := tx.Exec("SAVEPOINT " + cascadeSavepoint); err != nil {
		return fmt.Errorf("manifest.AfterDephantomize savepoint rid=%d: %w", rid, err)
	}
	mark := cl.sched.mark()
	handled, linkErr := linkArtifact(tx, rid, d, cl.cache, cl.sched)
	if linkErr != nil {
		// The artifact's rows roll back, and so does what they scheduled.
		cl.sched.rollback(mark)
		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)
}