Artifact [e297f656a0]

Artifact e297f656a053c91df962dc7eaa18202accbd9f0ab3512925a1e7c591153c4d61:

• File internal/seedcheck/seedcheck.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: 16494) [annotate] [blame] [check-ins using] [more...]

// Package seedcheck keeps an origin from linking or serving a horizon seed
// pushed to it (draft-fossil-horizon-00 §5.2.3).
//
// A pushed control artifact is a seed when it carries this implementation's
// seed label or a horizon-seed T card. It is judged when it first becomes
// readable. The push handler judges what it can read at once, and refuses
// the push. A delta whose basis it lacks is recorded pending, under the root
// of its basis chain (the row whose arrival makes it readable), and withheld
// from every reply until judged. Whatever later fills that root (a push or
// pull, a check-in, a tag) judges the rows pending under it, and only those
// (JudgeFilled); a delta received onto the root, by any path, judges them
// if it is readable and otherwise moves them to its own root (DeltaStored);
// the link step judges a pending row it reaches too (Judge).
// A seed found that way is refused by its hash: it stays stored, so a delta
// built on it stays readable, but it is never linked, announced or sent, and
// a re-push of it is refused on its bytes. Pulling the same hash from a peer
// stores nothing new: the row is held already, and stays refused.
//
// Both tables are store-local: no configuration group carries them, so no
// peer pulls them and a refusal can never make a clone drop the seed it
// holds. Each is created on first use; a store without them withholds
// nothing.
package seedcheck

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

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

// Label is the U card of the seeds go-libfossil mints: its own seed label,
// never a user's login, since a seed is a tag application no user made
// (§5.2.2).
const Label = "go-libfossil"

// Tag is the tag name of a seed's binding card; no pushed control artifact
// may carry it (§5.2.3).
const Tag = "horizon-seed"

const (
	pendingTable = "seedpending" // pushed deltas not yet readable, by rid, under their chain root
	refusedTable = "seedrefused" // hashes of refused seeds
)

const (
	pendingSchema = `CREATE TABLE IF NOT EXISTS ` + pendingTable + `(
  rid INTEGER PRIMARY KEY,
  root INTEGER NOT NULL
)`
	pendingIndex  = `CREATE INDEX IF NOT EXISTS ` + pendingTable + `_root ON ` + pendingTable + `(root)`
	refusedSchema = `CREATE TABLE IF NOT EXISTS ` + refusedTable + `(uuid TEXT PRIMARY KEY) WITHOUT ROWID`
)

// noRoot is the root of a row whose basis chain loops or runs past
// content.MaxDeltaChainDepth: no fill can make it readable, so it stays
// pending, withheld, and is never judged.
const noRoot libfossil.FslID = 0

// judgeCacheBytes bounds the expanded content one JudgeFilled call keeps,
// so a chain pending under one root is rebuilt once along its length.
const judgeCacheBytes = 16 << 20

// IsSeed reports whether d is a seed an origin refuses: a control artifact
// whose U card is Label, or one carrying a Tag T card under any label.
// Another implementation's label alone is ordinary, and so is a user's
// login unless it equals Label.
func IsSeed(d *deck.Deck) bool {
	if d == nil {
		panic("seedcheck.IsSeed: d must not be nil")
	}
	if d.Type != deck.Control {
		return false
	}
	if d.U != nil && *d.U == Label {
		return true
	}
	for _, t := range d.T {
		if t.Name == Tag {
			return true
		}
	}
	return false
}

// IsSeedBytes is IsSeed over an artifact's full content. Bytes that name
// neither the label nor the tag are rejected without parsing.
func IsSeedBytes(data []byte) bool {
	if data == nil {
		panic("seedcheck.IsSeedBytes: data must not be nil")
	}
	if !bytes.Contains(data, []byte("\nU "+Label+"\n")) && !bytes.Contains(data, []byte(Tag)) {
		return false
	}
	d, err := deck.Parse(data)
	return err == nil && IsSeed(d)
}

// MarkPending records rid, a pushed delta not yet readable, under the root
// of its basis chain, for judging when that root is filled.
func MarkPending(q db.Querier, rid libfossil.FslID) error {
	if q == nil || rid <= 0 {
		panic("seedcheck.MarkPending: q must not be nil and rid must be positive")
	}
	root, err := chainRoot(q, rid)
	if err != nil {
		return err
	}
	return markPending(q, rid, root)
}

func markPending(q db.Querier, rid, root libfossil.FslID) error {
	for _, stmt := range []string{pendingSchema, pendingIndex} {
		if _, err := q.Exec(stmt); err != nil {
			return fmt.Errorf("seedcheck.MarkPending: %w", err)
		}
	}
	if _, err := q.Exec("INSERT OR REPLACE INTO "+pendingTable+"(rid, root) VALUES(?, ?)", rid, root); err != nil {
		return fmt.Errorf("seedcheck.MarkPending: %w", err)
	}
	// It leaves the unclustered table too, so neither an inventory nor a
	// cluster names it until it is judged not a seed (decide).
	if _, err := q.Exec("DELETE FROM unclustered WHERE rid=?", rid); err != nil {
		return fmt.Errorf("seedcheck.MarkPending: %w", err)
	}
	return nil
}

// chainRoot follows delta.srcid from rid to the row with no delta, whose
// arrival makes rid readable, or reports noRoot for a chain that loops or
// runs past content.MaxDeltaChainDepth. The walk counts nodes as content's
// does, so a chain it roots is one content can expand, and a planted chain
// cannot hold the write lock for a walk of every row it holds.
func chainRoot(q db.Querier, rid libfossil.FslID) (libfossil.FslID, error) {
	cur := rid
	for nodes := 1; ; nodes++ {
		if nodes > content.MaxDeltaChainDepth {
			return noRoot, nil
		}
		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("seedcheck: basis chain of rid=%d: %w", rid, err)
		}
		cur = next
	}
}

// DeltaStored is told of every received delta stored at rid, on any path
// (push, pull, clone): the one event besides a full-text fill that can
// change what the rows pending under rid wait on. If rid is readable now,
// those rows are too, and are judged (JudgeFilled). If it is not, they wait
// on what rid waits on, and move to rid's chain root, so the fill that
// finally makes the chain readable judges them, whatever path it takes.
func DeltaStored(q db.Querier, rid libfossil.FslID) error {
	if q == nil || rid <= 0 {
		panic("seedcheck.DeltaStored: q must not be nil and rid must be positive")
	}
	ok, err := tableExists(q, pendingTable)
	if err != nil || !ok {
		return err
	}
	var waiting int
	if err := q.QueryRow("SELECT count(*) FROM "+pendingTable+" WHERE root=?", rid).Scan(&waiting); err != nil {
		return fmt.Errorf("seedcheck.DeltaStored: %w", err)
	}
	if waiting == 0 {
		return nil
	}
	if content.IsAvailable(q, rid) {
		_, err := JudgeFilled(q, []libfossil.FslID{rid})
		return err
	}
	root, err := chainRoot(q, rid)
	if err != nil {
		return err
	}
	if _, err := q.Exec("UPDATE "+pendingTable+" SET root=? WHERE root=?", root, rid); err != nil {
		return fmt.Errorf("seedcheck.DeltaStored: %w", err)
	}
	return nil
}

// JudgeFilled judges the rows pending under each of filled, rids whose full
// content was just stored, and reports how many rows it examined. Each fill
// costs the rows that waited on it, not the whole pending set. Each row is
// readable now and is judged: a seed's hash is refused, and either way the
// row goes. A delta pulled onto a pending row's basis and itself unreadable
// is not marked pending, so the rows waiting on it stay pending, withheld,
// until the link step reaches them (Judge); a plain file among them stays
// withheld.
//
// A row that does not expand, or whose bytes do not hash to its name, is
// not the artifact it names; it stays pending, withheld, rather than be
// judged on bytes that are not its own, and the batch goes on.
func JudgeFilled(q db.Querier, filled []libfossil.FslID) (int, error) {
	if q == nil {
		panic("seedcheck.JudgeFilled: q must not be nil")
	}
	ok, err := tableExists(q, pendingTable)
	if err != nil || !ok || len(filled) == 0 {
		return 0, err
	}
	cache := content.NewCache(judgeCacheBytes)
	examined := 0
	for _, root := range filled {
		if root <= 0 {
			panic("seedcheck.JudgeFilled: filled rids must be positive")
		}
		waiting, err := rows(q, "SELECT p.rid, b.uuid FROM "+pendingTable+" p JOIN blob b ON b.rid=p.rid"+
			" WHERE p.root=? ORDER BY p.rid", root)
		if err != nil {
			return examined, err
		}
		if len(waiting) == 0 {
			continue
		}
		for _, w := range waiting {
			examined++
			// A row whose own bytes cannot be read (its delta does not apply,
			// or they are not the artifact its name names) stays pending,
			// withheld, and does not stop the other fills in the batch from
			// being judged. Any other failure, the chain to the filled root
			// included, is a fault in the store, and is reported.
			data, err := cache.Expand(q, w.rid)
			if unreadable(err) {
				continue
			}
			if err != nil {
				return examined, fmt.Errorf("seedcheck.JudgeFilled: rid=%d: %w", w.rid, err)
			}
			if _, err := decide(q, w.rid, w.uuid, IsSeedBytes(data)); err != nil {
				return examined, err
			}
		}
	}
	return examined, nil
}

// unreadable reports whether err from expanding a row says the row cannot be
// read, whatever basis arrives: its chain loops or runs past the depth bound
// (a pusher can plant either under a root an honest push later supplies), a
// delta that does not apply, or bytes that are not the artifact its name
// names.
func unreadable(err error) bool {
	return errors.Is(err, content.ErrHashMismatch) || errors.Is(err, content.ErrChainUnresolvable) ||
		errors.Is(err, delta.ErrInvalidDelta) || errors.Is(err, delta.ErrChecksum)
}

// Tables is which of the two tables a store has. A caller that links many
// artifacts reads it once per transaction (Present) and hands it to Judge.
type Tables struct {
	Pending, Refused bool
}

// Present reads which tables exist.
func Present(q db.Querier) (Tables, error) {
	if q == nil {
		panic("seedcheck.Present: q must not be nil")
	}
	var t Tables
	var err error
	if t.Pending, err = tableExists(q, pendingTable); err != nil {
		return Tables{}, err
	}
	t.Refused, err = tableExists(q, refusedTable)
	return t, err
}

// Judge decides rid, about to be linked with d its parsed content, given
// the tables the store has, and reports whether it must not be linked: its
// hash is refused, or it was pending and is a seed, which refuses its hash.
// Any other pending row is cleared. This is the derive-time check every
// link path reaches, behind JudgeFilled.
func Judge(q db.Querier, t Tables, rid libfossil.FslID, d *deck.Deck) (bool, error) {
	if q == nil || rid <= 0 || d == nil {
		panic("seedcheck.Judge: q and d must not be nil and rid must be positive")
	}
	if !t.Pending && !t.Refused {
		return false, nil
	}
	var uuid string
	if err := q.QueryRow("SELECT uuid FROM blob WHERE rid=?", rid).Scan(&uuid); err != nil {
		return false, fmt.Errorf("seedcheck.Judge: rid=%d: %w", rid, err)
	}
	if t.Refused {
		var n int
		if err := q.QueryRow("SELECT count(*) FROM "+refusedTable+" WHERE uuid=?", uuid).Scan(&n); err != nil {
			return false, fmt.Errorf("seedcheck.Judge: %w", err)
		}
		if n > 0 {
			return true, nil
		}
	}
	if !t.Pending {
		return false, nil
	}
	var one int
	err := q.QueryRow("SELECT 1 FROM "+pendingTable+" WHERE rid=?", rid).Scan(&one)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("seedcheck.Judge: pending rid=%d: %w", rid, err)
	}
	return decide(q, rid, uuid, IsSeed(d))
}

// decide records the verdict on pending rid, which MarkPending took out of
// the unclustered table: a seed's hash is refused and it stays out, so no
// inventory or cluster names it; anything else goes back in. Either way the
// pending row goes.
func decide(q db.Querier, rid libfossil.FslID, uuid string, seed bool) (bool, error) {
	if seed {
		if err := Refuse(q, uuid); err != nil {
			return false, err
		}
	} else if _, err := q.Exec("INSERT OR IGNORE INTO unclustered(rid) VALUES(?)", rid); err != nil {
		return false, fmt.Errorf("seedcheck: judge rid=%d: %w", rid, err)
	}
	if _, err := q.Exec("DELETE FROM "+pendingTable+" WHERE rid=?", rid); err != nil {
		return false, fmt.Errorf("seedcheck: judge rid=%d: %w", rid, err)
	}
	return seed, nil
}

// Refuse records uuid as a refused seed. The row stored under it is kept,
// so deltas on it stay readable, but it is withheld and never linked.
func Refuse(q db.Querier, uuid string) error {
	if q == nil || uuid == "" {
		panic("seedcheck.Refuse: q must not be nil and uuid must not be empty")
	}
	if _, err := q.Exec(refusedSchema); err != nil {
		return fmt.Errorf("seedcheck.Refuse: %w", err)
	}
	if _, err := q.Exec("INSERT OR IGNORE INTO "+refusedTable+"(uuid) VALUES(?)", uuid); err != nil {
		return fmt.Errorf("seedcheck.Refuse: %w", err)
	}
	return nil
}

// IsRefusedUUID reports whether uuid was refused as a seed.
func IsRefusedUUID(q db.Querier, uuid string) (bool, error) {
	if q == nil {
		panic("seedcheck.IsRefusedUUID: q must not be nil")
	}
	ok, err := tableExists(q, refusedTable)
	if err != nil || !ok {
		return false, err
	}
	var n int
	if err := q.QueryRow("SELECT count(*) FROM "+refusedTable+" WHERE uuid=?", uuid).Scan(&n); err != nil {
		return false, fmt.Errorf("seedcheck.IsRefusedUUID: %w", err)
	}
	return n > 0, nil
}

// ForgetDeleted drops the pending rows whose blob rows are gone. shun.Purge,
// the one deleter of blob rows, runs it in the same transaction: rids are
// reused, so a later artifact stored under one must not be withheld.
func ForgetDeleted(q db.Querier) error {
	if q == nil {
		panic("seedcheck.ForgetDeleted: q must not be nil")
	}
	ok, err := tableExists(q, pendingTable)
	if err != nil || !ok {
		return err
	}
	const gone = " FROM " + pendingTable + " WHERE NOT EXISTS (SELECT 1 FROM blob WHERE rid=" + pendingTable + ".rid)"
	var n int
	if err := q.QueryRow("SELECT count(*)" + gone).Scan(&n); err != nil || n == 0 {
		return err
	}
	if _, err := q.Exec("DELETE" + gone); err != nil {
		return fmt.Errorf("seedcheck.ForgetDeleted: %w", err)
	}
	return nil
}

// IsWithheld reports whether rid is pending, or stored under a refused
// hash. No reply announces or sends it.
func IsWithheld(q db.Querier, rid libfossil.FslID) (bool, error) {
	if q == nil {
		panic("seedcheck.IsWithheld: q must not be nil")
	}
	t, err := Present(q)
	if err != nil {
		return false, err
	}
	var n int
	if t.Pending {
		if err := q.QueryRow("SELECT count(*) FROM "+pendingTable+" WHERE rid=?", rid).Scan(&n); err != nil {
			return false, fmt.Errorf("seedcheck.IsWithheld: %w", err)
		}
	}
	if n == 0 && t.Refused {
		if err := q.QueryRow("SELECT count(*) FROM blob b JOIN "+refusedTable+" r ON r.uuid=b.uuid WHERE b.rid=?",
			rid).Scan(&n); err != nil {
			return false, fmt.Errorf("seedcheck.IsWithheld: %w", err)
		}
	}
	return n > 0, nil
}

// Withheld returns every rid IsWithheld reports. The set is small: it holds
// only pushed deltas not yet readable, and refused seeds.
func Withheld(q db.Querier) (map[libfossil.FslID]bool, error) {
	if q == nil {
		panic("seedcheck.Withheld: q must not be nil")
	}
	t, err := Present(q)
	if err != nil {
		return nil, err
	}
	out := map[libfossil.FslID]bool{}
	var queries []string
	if t.Pending {
		queries = append(queries, "SELECT rid, '' FROM "+pendingTable)
	}
	if t.Refused {
		queries = append(queries, "SELECT b.rid, '' FROM blob b JOIN "+refusedTable+" r ON r.uuid=b.uuid")
	}
	for _, query := range queries {
		found, err := rows(q, query)
		if err != nil {
			return nil, err
		}
		for _, r := range found {
			out[r.rid] = true
		}
	}
	return out, nil
}

func tableExists(q db.Querier, name string) (bool, error) {
	var n int
	if err := q.QueryRow("SELECT count(*) FROM sqlite_master WHERE type='table' AND name=?", name).
		Scan(&n); err != nil {
		return false, fmt.Errorf("seedcheck: %w", err)
	}
	return n > 0, nil
}

// row is a rid and, where the query selects one, its hash.
type row struct {
	rid  libfossil.FslID
	uuid string
}

func rows(q db.Querier, query string, args ...any) ([]row, error) {
	rs, err := q.Query(query, args...)
	if err != nil {
		return nil, fmt.Errorf("seedcheck: %w", err)
	}
	defer rs.Close()
	var out []row
	for rs.Next() {
		var r row
		if err := rs.Scan(&r.rid, &r.uuid); err != nil {
			return nil, fmt.Errorf("seedcheck: %w", err)
		}
		out = append(out, r)
	}
	return out, rs.Err()
}