Artifact [aef2f350ea]

Artifact aef2f350ea3b23c2629137504a2478fad5988287b6155ced6f7710c919b72d9a:

• File internal/sync/receive_pass.go — part of check-in [eac16cf7ed] at 2026-10-24 13:08:15 on branch main — fix(sync): bound basis-chain walks at the read depth

  Any pusher could hold the write lock about 30s per push with a two-delta cycle. The receive pass's basis-chain walks, and the horizon seed check's pending-row walk, ran to 1<<24 steps before giving up.

  All of them now stop at content.MaxDeltaChainDepth, the depth at which content.Expand stops, so no walk outlasts what a read could ever use. What crossing the bound means at each site:

  - chainEnd (rootOf, for a dangling delta being stored) keeps a visited set. A chain that loops is refused, as it was, but at the revisit rather than after 1<<24 steps. A chain longer than the bound returns errChainTooDeep. storeDeltaContent then stores the delta untracked: the push succeeds, as a push onto a chain between 16k and 16M deep always did, and since no basis can make the delta readable there is nothing to check against its name later. Storing matches the current behaviour, and an honest transfer never fails for a deep chain it did not build. - verifyFrom's cascade carries each row's depth from the basis that arrived, its own chain included. It does not check, or descend past, a row deeper than the bound: no read can produce those bytes, so the pass owes them no check. A delta that does not hash to its name past the bound no longer aborts the pass. - pathRows loses its step counter. Its visited marks already stop it at a row on a path, so it visits each row once, and the cascade's depth check bounds what it verifies. - seedcheck's chainRoot gives a chain that loops or runs past the bound no root, consistent with ErrChainUnresolvable: the row stays pending, withheld, and is never judged.

  Tests: - Closing a two-delta cycle is refused in under 5s (it took 27s). - A push onto a chain one past the bound is stored. - chainEnd stops exactly at its limit. - A dangling delta with a wrong name one node past the bound does not abort the push, whether the basis that arrives is full text or a delta deep on a held chain, while one within the bound is still checked and readable. - MarkPending gives an over-deep chain no root, so filling its far end examines nothing, and returns at once on a cycle. - TestHorizonOverDeepPendingChainDoesNotBlockFill still passes. - Removing each bound, the cycle check, the untracked store or the root depth fails a test. - chainDepth's own early stop is an equivalent mutant: the cascade's root has just been expanded, so its chain is within the bound unless the store is damaged. (user: [email protected], size: 18939) [annotate] [blame] [check-ins using] [more...]

package sync

import (
	"database/sql"
	"errors"
	"fmt"

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

// receivePass is one verification pass of received content: one reply on
// the client, one request on the server, one round of a clone. Each runs in
// one write transaction (see runPass), as fossil's page_xfer and client_sync
// run each request and each round.
//
// Section 7.7 (SP-7.7-4) has the repository recompute, before the pass
// commits, every artifact stored in it whose complete bytes are available by
// then, and abort the commit on a mismatch. fossil queues each stored rid
// with verify_before_commit and checks the queue in its commit hook. Here the
// pass remembers the deltas it stored dangling, and the store that makes
// one's bytes available checks it at once. A mismatch is a late
// hashMismatchError, which aborts the pass: its transaction rolls back
// everything the pass stored, as the failed commit hook does. A delta stored
// against a held basis, and full content, are verified before storage and
// never enter the pass.
type receivePass struct {
	dangling map[libfossil.FslID]danglingDelta
	// byRoot indexes dangling by the rid whose arrival makes each delta's
	// bytes available: the unavailable end of its basis chain. The chain may
	// run through rows stored in an earlier pass, so the direct basis is not
	// the key. A store checks only the deltas its artifact grounds, not the
	// whole set: a clone round can leave thousands dangling.
	byRoot map[libfossil.FslID][]libfossil.FslID
	// basis holds the full bytes of artifacts this pass stored, and of the
	// chain nodes checkDelta rebuilt, unchecked against their names as
	// checkDelta's basis is (content.NewUncheckedCache), so checking a delta
	// against a basis the pass already holds starts from those bytes. A
	// file's history arrives as one chain, each delta against the one
	// before: rebuilding every basis from the chain's root made a clone
	// quadratic in its depth. Only bytes of rows that outlive their card
	// enter it (see remember).
	basis *content.Cache
}

// receiveBasisBytes bounds a pass's basis cache, as the receive linker's
// expanded-content cache is bounded.
const receiveBasisBytes = 64 << 20

// danglingDelta is a delta this pass stored before its basis was available:
// the name it must hash to, and the rid that grounds it.
type danglingDelta struct {
	uuid string
	root libfossil.FslID
}

func newReceivePass() *receivePass {
	return newReceivePassOver(content.NewUncheckedCache(receiveBasisBytes))
}

// newReceivePassOver is newReceivePass with a basis cache the caller keeps
// across passes. It holds only rows a committed pass stored or read, so the
// caller must clear it when a pass rolls back.
func newReceivePassOver(basis *content.Cache) *receivePass {
	if basis == nil {
		panic("sync.newReceivePassOver: basis must not be nil")
	}
	return &receivePass{
		dangling: make(map[libfossil.FslID]danglingDelta),
		byRoot:   make(map[libfossil.FslID][]libfossil.FslID),
		basis:    basis,
	}
}

// remember keeps rid's full bytes, which hash to its name, for checking the
// deltas the pass receives against it. It is called only once the card that
// stored rid has committed its savepoint: a rolled-back row's rid can be
// handed to the next row stored.
func (p *receivePass) remember(rid libfossil.FslID, full []byte) {
	if rid <= 0 {
		panic("sync.receivePass.remember: rid must be positive")
	}
	if full == nil {
		panic("sync.receivePass.remember: full must not be nil")
	}
	p.basis.Remember(rid, full)
}

// heldReal reports whether uuid already has a real (non-phantom) row, which
// storage leaves as it is.
func heldReal(q db.Querier, uuid string) (bool, error) {
	var size int64
	err := q.QueryRow("SELECT size FROM blob WHERE uuid = ?", uuid).Scan(&size)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("state of %s: %w", uuid, err)
	}
	return size >= 0, nil
}

// rootOf finds the rid whose arrival makes a delta on srcRid available: the
// end of srcRid's basis chain. A basis that is itself dangling in this pass
// shares that delta's root. The delta adds a node to the chain, so the walk
// allows one node fewer than content.MaxDeltaChainDepth.
func (p *receivePass) rootOf(q db.Querier, srcRid libfossil.FslID) (libfossil.FslID, error) {
	return chainEnd(q, srcRid, p.dangling, content.MaxDeltaChainDepth-1)
}

// errChainTooDeep is a basis chain longer than a delta stored on it may be
// and still be read: content.Expand stops at content.MaxDeltaChainDepth.
var errChainTooDeep = errors.New("basis chain too deep to read")

// chainEnd follows delta.srcid from start to the row with no delta, which is
// where the chain's bytes begin, visiting at most limit nodes. A rid in memo
// ends the walk at that delta's recorded root. A chain that loops is an
// error; one longer than limit is errChainTooDeep, since nothing could read
// a delta stored on it. Both stop early, so a planted chain cannot hold the
// write lock for a walk of every row it holds.
func chainEnd(q db.Querier, start libfossil.FslID,
	memo map[libfossil.FslID]danglingDelta, limit int) (libfossil.FslID, error) {
	if limit <= 0 {
		panic("sync.chainEnd: limit must be positive")
	}
	seen := make(map[libfossil.FslID]bool)
	cur := start
	for nodes := 1; ; nodes++ {
		if d, ok := memo[cur]; ok {
			return d.root, nil
		}
		seen[cur] = true
		var next libfossil.FslID
		err := q.QueryRow("SELECT srcid FROM delta WHERE rid=?", cur).Scan(&next)
		if errors.Is(err, sql.ErrNoRows) {
			return cur, nil
		}
		if err != nil {
			return 0, fmt.Errorf("basis chain of rid=%d: %w", start, err)
		}
		if seen[next] {
			return 0, fmt.Errorf("basis chain of rid=%d loops at rid=%d", start, next)
		}
		if nodes >= limit {
			return 0, errChainTooDeep
		}
		cur = next
	}
}

// track records a delta stored dangling, which must hash to uuid, whose bytes
// become available when root arrives. Deltas waiting on rid itself now wait
// on root, since rid is no more available than its own basis.
func (p *receivePass) track(rid, root libfossil.FslID, uuid string) {
	if rid <= 0 || root <= 0 {
		panic("sync.receivePass.track: rid and root must be positive")
	}
	if uuid == "" {
		panic("sync.receivePass.track: uuid must not be empty")
	}
	p.dangling[rid] = danglingDelta{uuid: uuid, root: root}
	waiting := append(p.byRoot[rid], rid)
	delete(p.byRoot, rid)
	for _, w := range waiting {
		d := p.dangling[w]
		d.root = root
		p.dangling[w] = d
	}
	p.byRoot[root] = append(p.byRoot[root], waiting...)
}

// cascadeItem is a row the cascade visits, with its own bytes (nil when
// they cannot be rebuilt).
type cascadeItem struct {
	rid   libfossil.FslID
	bytes []byte
	depth int // nodes in rid's basis chain, rid and its root included
}

// verifyFrom checks every dangling delta that rid, just made available, now
// grounds. The first that does not give its HASH is returned as a late
// hashMismatchError, which aborts the pass: nothing is undone here, since
// the pass's transaction rolls it all back.
//
// The walk goes down the delta table from rid, applying each row's delta to
// its parent's bytes as it goes, so a chain of n deltas costs n applies, not
// the n(n+1)/2 that rebuilding each from rid would. fossil gets the same
// bound from content_get's cache of every 8th step. It passes through rows
// this pass did not store, which a dangling delta's chain may run through.
func (p *receivePass) verifyFrom(q db.Querier, rid libfossil.FslID) error {
	if q == nil {
		panic("sync.receivePass.verifyFrom: q must not be nil")
	}
	if rid <= 0 {
		panic("sync.receivePass.verifyFrom: rid must be positive")
	}
	deps := p.byRoot[rid]
	if len(deps) == 0 {
		return nil
	}
	delete(p.byRoot, rid)
	onPath, err := pathRows(q, rid, deps)
	if err != nil {
		return err
	}
	// A basis that cannot be rebuilt is applied to nothing: its dependents
	// get no HASH, as when fossil's blob_delta_apply fails on what
	// content_get gave it.
	rootBytes, _ := content.ExpandUnchecked(q, rid)
	rootDepth, err := chainDepth(q, rid)
	if err != nil {
		return err
	}
	stack := []cascadeItem{{rid, rootBytes, rootDepth}}
	for len(stack) > 0 {
		it := stack[len(stack)-1]
		stack = stack[:len(stack)-1]
		kids, err := childRows(q, it.rid, onPath)
		if err != nil {
			return err
		}
		for _, kid := range kids {
			// A row deeper than content.MaxDeltaChainDepth can never be
			// read, so it, and every row on it, is not checked against its
			// name: the pass owes nothing for bytes no read can produce.
			if it.depth+1 > content.MaxDeltaChainDepth {
				continue
			}
			kidBytes, ok, err := p.visit(q, kid, it.bytes)
			if err != nil {
				return err
			}
			if !ok {
				return &hashMismatchError{uuid: p.dangling[kid].uuid, late: true}
			}
			stack = append(stack, cascadeItem{kid, kidBytes, it.depth + 1})
		}
	}
	return nil
}

// visit applies row kid's delta to its parent's bytes. For a dangling delta
// it reports whether the result is its HASH, and drops it from the pass when
// it is; any other row always passes.
func (p *receivePass) visit(q db.Querier, kid libfossil.FslID,
	parent []byte) ([]byte, bool, error) {
	var kidBytes []byte
	if parent != nil {
		payload, err := blob.Load(q, kid)
		if err != nil {
			return nil, false, fmt.Errorf("verify received delta rid=%d: %w", kid, err)
		}
		if out, err := delta.Apply(parent, payload); err == nil {
			kidBytes = out
		}
	}
	d, dangling := p.dangling[kid]
	if !dangling {
		return kidBytes, true, nil
	}
	if kidBytes == nil || !hashMatches(d.uuid, kidBytes) {
		return nil, false, nil
	}
	delete(p.dangling, kid)
	return kidBytes, true, nil
}

// pathRows returns the rows on the chains from each of deps up to root, the
// only rows the cascade needs to visit.
func pathRows(q db.Querier, root libfossil.FslID,
	deps []libfossil.FslID) (map[libfossil.FslID]bool, error) {
	on := make(map[libfossil.FslID]bool, len(deps))
	for _, d := range deps {
		cur := d
		// The walk stops at a row already on a path, so it visits each row
		// once however the chains share or loop; and the cascade does not go
		// past content.MaxDeltaChainDepth whatever this marks.
		for cur != root && !on[cur] {
			on[cur] = true
			err := q.QueryRow("SELECT srcid FROM delta WHERE rid=?", cur).Scan(&cur)
			if errors.Is(err, sql.ErrNoRows) {
				break
			}
			if err != nil {
				return nil, fmt.Errorf("basis chain of rid=%d: %w", d, err)
			}
		}
	}
	return on, nil
}

// chainDepth counts the nodes of rid's basis chain, rid and its root
// included, stopping one past content.MaxDeltaChainDepth: rid is readable,
// so its chain ends within the bound unless the store is damaged.
func chainDepth(q db.Querier, rid libfossil.FslID) (int, error) {
	cur := rid
	for nodes := 1; ; nodes++ {
		if nodes > content.MaxDeltaChainDepth {
			return nodes, nil
		}
		var next libfossil.FslID
		err := q.QueryRow("SELECT srcid FROM delta WHERE rid=?", cur).Scan(&next)
		if errors.Is(err, sql.ErrNoRows) {
			return nodes, nil
		}
		if err != nil {
			return 0, fmt.Errorf("basis chain of rid=%d: %w", rid, err)
		}
		cur = next
	}
}

// childRows lists the rows whose delta is against parent and that lie on a
// path to a dangling delta, in rid order; with a nil onPath, every one.
func childRows(q db.Querier, parent libfossil.FslID,
	onPath map[libfossil.FslID]bool) ([]libfossil.FslID, error) {
	rows, err := q.Query("SELECT rid FROM delta WHERE srcid=? ORDER BY rid", parent)
	if err != nil {
		return nil, fmt.Errorf("deltas on rid=%d: %w", parent, err)
	}
	defer rows.Close()
	var kids []libfossil.FslID
	for rows.Next() {
		var kid libfossil.FslID
		if err := rows.Scan(&kid); err != nil {
			return nil, fmt.Errorf("deltas on rid=%d: %w", parent, err)
		}
		if onPath == nil || onPath[kid] {
			kids = append(kids, kid)
		}
	}
	return kids, rows.Err()
}

// available reports whether the artifact named uuid has its bytes, as
// content.AvailableByUUID does, without walking its chain when the pass
// already knows: bytes the pass holds are available, and a delta dangling in
// the pass is not. A clone asks this of every delta's basis, and the walk is
// as deep as the chain.
func (p *receivePass) available(q db.Querier, uuid string) bool {
	if q == nil {
		panic("sync.receivePass.available: q must not be nil")
	}
	rid, held := blob.Exists(q, uuid)
	if !held {
		return false
	}
	if p.basis.Holds(rid) {
		return true
	}
	if _, dangling := p.dangling[rid]; dangling {
		return false
	}
	return content.IsAvailable(q, rid)
}

// deltaVerdict is what applying a received delta to its basis showed.
type deltaVerdict int

const (
	// deltaUnverifiable: the basis's bytes are not available in this pass.
	deltaUnverifiable deltaVerdict = iota
	// deltaMatch: the delta gives its HASH.
	deltaMatch
	// deltaMismatch: it does not, or does not apply.
	deltaMismatch
)

// checkDelta applies a received delta to its basis and compares the result
// with uuid. Section 7.7 verifies a reconstructable file before storage, as
// xfer_accept_file applies the delta to the source content_get gives it. A
// basis that is not available leaves the delta unverifiable, to be stored
// dangling, as fossil stores it.
//
// The basis is reconstructed without checking it against its own name, as
// content_get does: integrity is the result's hash, not the basis's, so a
// held basis whose bytes are not its name still has the delta checked
// against what it gives. A basis reported available whose bytes cannot be
// rebuilt (bad zlib, or a delta in its chain that does not apply) is a
// mismatch too: fossil 2.29's content_of_blob and content_get keep going on
// the bytes they have, so xfer_accept_file applies the delta to those and
// hname_verify_hash refuses the result.
//
// A delta that does not apply is a mismatch: fossil verifies what
// blob_delta_apply left, which is not the artifact either.
//
// The basis is rebuilt through the pass's cache (basis) from the deepest node
// of its chain the cache holds, so a chain received basis first costs one
// apply per delta. A basis the cache holds is available; a delta dangling in
// this pass is not; any other basis has its chain walked for availability
// first, as before.
func (p *receivePass) checkDelta(q db.Querier, uuid, deltaSrc string,
	payload []byte) (deltaVerdict, []byte, error) {
	if q == nil {
		panic("sync.checkDelta: q must not be nil")
	}
	if uuid == "" || deltaSrc == "" {
		panic("sync.checkDelta: uuid and deltaSrc must not be empty")
	}
	var srcRid libfossil.FslID
	err := q.QueryRow("SELECT rid FROM blob WHERE uuid=?", deltaSrc).Scan(&srcRid)
	if errors.Is(err, sql.ErrNoRows) {
		return deltaUnverifiable, nil, nil
	}
	if err != nil {
		return deltaUnverifiable, nil, fmt.Errorf("look up %s: %w", deltaSrc, err)
	}
	if _, dangling := p.dangling[srcRid]; dangling {
		return deltaUnverifiable, nil, nil
	}
	if !p.basis.Holds(srcRid) {
		ok, err := content.CheckAvailable(q, srcRid)
		if err != nil {
			return deltaUnverifiable, nil, err
		}
		if !ok {
			return deltaUnverifiable, nil, nil
		}
	}
	src, err := p.basis.ExpandUnchecked(q, srcRid)
	if err != nil {
		return deltaMismatch, nil, nil
	}
	target, err := delta.Apply(src, payload)
	if err != nil || !hashMatches(uuid, target) {
		return deltaMismatch, nil, nil
	}
	return deltaMatch, target, nil
}

// hashMismatchError is a received artifact whose content is not its name.
// uuid names the artifact at fault, which for a delta caught late is not the
// card being stored when it is found. late marks that case: the commit check
// of Section 7.7 failed, which aborts the whole pass on either end. A
// mismatch found before storage refuses only its own card on the server; a
// client fails the round on either kind (failsRound, Section 10.1).
type hashMismatchError struct {
	uuid string
	late bool
}

func (e *hashMismatchError) Error() string {
	return "received file " + e.uuid + ": " + errHashMismatch.Error()
}

func (e *hashMismatchError) Unwrap() error { return errHashMismatch }

// abortsPass reports whether err ends a pass by rolling back everything it
// stored (Section 7.7's failed commit check), rather than ending only the
// card that returned it.
func abortsPass(err error) bool {
	var mismatch *hashMismatchError
	return errors.As(err, &mismatch) && mismatch.late
}

// failsRound reports whether err is a received file or cfile whose content
// does not match its name, found before storage or late. Section 10.1 makes
// either an explicit fail-fast path for a client: the round that received it
// commits nothing, its processed prefix included, and no further round
// starts. A server still treats the early case as a card error (abortsPass).
func failsRound(err error) bool {
	var mismatch *hashMismatchError
	return errors.As(err, &mismatch)
}

// errPassAborted is what a pass's callback returns to roll its transaction
// back once it has recorded why: the server's late mismatch and §8.1 fatal,
// whose reply is already built.
var errPassAborted = errors.New("sync: pass aborted")

// runPass runs fn in the one write transaction of a verification pass, with
// the repository bound to it: every store, cascade and crosslink fn makes
// through that repository commits together or not at all. fn reports two
// errors. A card error that only ends the card loop (cardErr) leaves the
// prefix processed before it to commit, as Section 10.1 has a handler error
// do; runPass returns it as cardErr with a nil err. Any other error, a late
// hash mismatch included, rolls the pass back and is returned as err, except
// errPassAborted, which rolls back and returns nil: its caller already holds
// the reason. A non-nil err therefore means nothing fn did was kept.
func runPass(r *repo.Repo,
	fn func(bound *repo.Repo) (cardErr, err error)) (cardErr, err error) {
	if r == nil {
		panic("sync.runPass: r must not be nil")
	}
	return runPassIn(r.WithBoundTx, fn)
}

// runPassIn is runPass in the transaction begin opens: r.WithBoundTx, or
// r.WithBoundReadTx for a pass that stores nothing.
func runPassIn(begin func(func(bound *repo.Repo) error) error,
	fn func(bound *repo.Repo) (cardErr, err error)) (cardErr, err error) {
	if begin == nil {
		panic("sync.runPassIn: begin must not be nil")
	}
	if fn == nil {
		panic("sync.runPassIn: fn must not be nil")
	}
	err = begin(func(bound *repo.Repo) error {
		var ferr error
		cardErr, ferr = fn(bound)
		if ferr == nil && abortsPass(cardErr) {
			ferr, cardErr = cardErr, nil
		}
		return ferr
	})
	if errors.Is(err, errPassAborted) {
		err = nil
	}
	if err != nil {
		cardErr = nil
	}
	return cardErr, err
}