Artifact [4109910717]

Artifact 4109910717d9691bf660d9837b08ca56e8bd88bb4589af1c59ee537c79d88cb9:

• File internal/sync/clone.go — part of check-in [18bfbc166c] at 2026-07-27 23:31:22 on branch trunk — fix(sync): derive the project code, and write artifact visibility inside the storing transaction (#204)

  * fix(sync): derive the project code from the repo instead of panicking (#199)

  Supplying User/Password made SyncOpts.ProjectCode mandatory in practice: buildLoginCard passed the caller's raw value straight to computeLogin, which asserts it is non-empty, so an authenticated Sync on a repo whose project code the caller had not repeated took the process down. Nothing in the public API signalled that a password made the field required, and a server embedding the library died with it.

  buildRequest already fell back to the repo's stored project-code for the push/pull cards; that fallback was local to one function, which is exactly how the login path drifted out of step with it. Resolve the code once when the session is built — caller value first, the repo's own config second — and let buildRequest and buildLoginCard both read it. A code that genuinely cannot be determined now returns a descriptive error from the site that needs it rather than panicking.

  Sync's deferred result-not-nil assertion fired during panic unwinding, where err and result are both nil, replacing the original cause with its own message. The invariant only holds on a normal return, so it no longer asserts while a panic is in flight.

  * fix(sync): write received artifacts' visibility inside the storing transaction (#200)

  Receiving an artifact cost three write transactions, not one. The blob was stored in a transaction, its dephantomize cascade committed another, and then applyPrivateStatus re-derived the rid on the pool and issued a bare DELETE FROM private -- an implicit transaction of its own, with its own BEGIN IMMEDIATE and fsync, for a row that a transaction microseconds earlier had guaranteed did not exist. Every igot card announcing an artifact already held cost one more. Measured against a real fossil peer: 1509 statements executed outside a transaction to receive 1504 artifacts.

  That is minutes of pure commit overhead on a large repository -- roughly 200,000 write transactions for a 67k-artifact pull -- and it is fragile as well as slow. Each of those writes contests the database's single write lock independently, so N artifacts means N chances to lose that contest and fail outright rather than one chance to wait at a single BEGIN.

  Visibility now travels with the content it describes: storeReceivedFile takes the visibility a private card announced and records it inside the transaction that stores the blob, using the rid that transaction already has. The status is now atomic with the content rather than a separate write that could be interrupted between the two. The igot path buffers its updates and writes them once per round. The same three-transaction shape on the server's receive path goes with it.

  The same real-fossil pull now executes 2 statements outside a transaction for those 1504 artifacts, down from 1509. db.DB.PoolExecs makes that budget assertable, which is how the regression test pins O(rounds) rather than O(artifacts) without needing a large corpus.

  On the failure originally reported: the "database is locked" error came from a harness that copied a live SQLite database out from under a running server without its WAL, so that specific lock holder is not a product defect and this does not claim to fix it. The transaction storm above is a real performance and robustness defect on its own evidence, and is what this changes.

  * test(sync): pin the transaction budget, not just writes outside a transaction

  The budget tests measured PoolExecs, which counts statements issued outside a transaction. That pins one half of the defect and misses the other: moving a per-artifact write from the pool into its own WithTx leaves PoolExecs unchanged while every artifact still pays a BEGIN IMMEDIATE and an fsync, and still contests the single write lock on its own. That is the cost issue #200 actually measured -- roughly three write transactions per received file, about 203,000 for one clone-sized pull.

  Verified by mutation: reverting the batched visibility flush to one transaction per update left both budget tests green. They now count transactions too, via a new db.DB.WriteTxns counter, and the same mutation fails with "began 51 write transactions for 50 igot cards over 1 rounds; want <= 4".

  Also stop the contention test from starving the pull it is meant to observe. The rival connection re-acquired the write lock immediately on release, so on a loaded machine it could hold the pull's single BEGIN past busy_timeout and fail the test even when the batching under test is perfect -- which is what happened on CI. It now yields between bursts, modelling another process doing occasional writes rather than testing SQLite's fairness. (user: [email protected], size: 17501) [annotate] [blame] [check-ins using]

package sync

import (
	"bytes"
	"context"
	"fmt"
	"os"
	"sort"
	"strings"

	"github.com/danmestas/go-libfossil/internal/content"
	"github.com/danmestas/go-libfossil/internal/manifest"
	"github.com/danmestas/go-libfossil/internal/repo"
	"github.com/danmestas/go-libfossil/internal/xfer"
	"github.com/danmestas/go-libfossil/simio"
)

// phantomStallSampleSize caps how many missing UUIDs PhantomStallError
// carries, so a repository with thousands of phantoms doesn't produce an
// unreadable error message.
const phantomStallSampleSize = 10

// processResponseCancelCheckStride is how many cards processResponse handles
// between context-cancellation checks. One clone batch is processed card by
// card with no network wait in between, so without an inner check a cancelled
// context would not be observed until the next round begins; the check is
// batched to avoid a channel read per card.
const processResponseCancelCheckStride = 256

// PhantomStallError reports a clone that stopped making progress while
// phantom blobs — content referenced but never received — were still
// outstanding. Terminating a stalled clone is correct; reporting it as
// success is not, so this is always returned instead of nil in that case.
// Count is programmatically reachable so a caller can act on it without
// parsing the message.
type PhantomStallError struct {
	Count  int      // phantom UUIDs still outstanding when the clone stopped
	Sample []string // up to phantomStallSampleSize lexicographically-smallest of those UUIDs, for diagnosis
}

func (e *PhantomStallError) Error() string {
	return fmt.Sprintf("sync.Clone: stalled with %d phantom artifact(s) outstanding, e.g. %s",
		e.Count, strings.Join(e.Sample, ", "))
}

// newPhantomStallError builds a PhantomStallError from the session's current
// phantom set. Called only once that set is known to be non-empty.
func newPhantomStallError(phantoms map[string]bool) *PhantomStallError {
	if len(phantoms) == 0 {
		panic("sync.newPhantomStallError: phantoms must not be empty")
	}
	all := make([]string, 0, len(phantoms))
	for uuid := range phantoms {
		all = append(all, uuid)
	}
	sort.Strings(all)
	if len(all) > phantomStallSampleSize {
		all = all[:phantomStallSampleSize]
	}
	return &PhantomStallError{Count: len(phantoms), Sample: all}
}

// cloneSession holds the mutable state of a running clone.
type cloneSession struct {
	repo        *repo.Repo
	env         *simio.Env
	opts        CloneOpts
	result      CloneResult
	phantoms    map[string]bool
	seqno       int
	projectCode string
	serverCode  string
	cookie      string // server-issued snapshot bound; echoed every round so the
	// server can scope its blob query to the rid range that existed when the
	// clone session opened. See cloneCapPrefix in handler.go.
	obs Observer
}

// Clone performs a full repository clone from a remote Fossil server.
// It creates a new repository at path, runs the clone protocol until
// convergence, and returns the opened repo and a result summary.
// On error, the partially-created repo file is removed.
func Clone(ctx context.Context, path string, t Transport, opts CloneOpts) (r *repo.Repo, result *CloneResult, err error) {
	if path == "" {
		panic("sync.Clone: path must not be empty")
	}
	if t == nil {
		panic("sync.Clone: t must not be nil")
	}

	env := opts.Env
	if env == nil {
		env = simio.RealEnv()
	}
	storage := env.Storage
	if storage == nil {
		storage = simio.OSStorage{}
	}

	// Path must not already exist.
	if _, statErr := storage.Stat(path); statErr == nil {
		return nil, nil, fmt.Errorf("sync.Clone: file already exists: %s", path)
	}

	user := opts.User
	if user == "" {
		user = "setup"
	}

	// Create the repository.
	r, err = repo.CreateWithEnv(path, user, env, "")
	if err != nil {
		return nil, nil, fmt.Errorf("sync.Clone: create repo: %w", err)
	}

	// Cleanup on error: close repo and remove the file.
	defer func() {
		if err != nil {
			r.Close()
			if rmErr := storage.Remove(path); rmErr != nil {
				fmt.Fprintf(os.Stderr, "sync.Clone: cleanup failed: %v\n", rmErr)
			}
			r = nil
		}
	}()

	// Clear project-code — the server will provide its own.
	if _, execErr := r.DB().Exec("DELETE FROM config WHERE name='project-code'"); execErr != nil {
		err = fmt.Errorf("sync.Clone: clear project-code: %w", execErr)
		return
	}

	cs := &cloneSession{
		repo:     r,
		env:      env,
		opts:     opts,
		seqno:    1,
		phantoms: make(map[string]bool),
	}
	cs.obs = resolveObserver(opts.Observer)

	cloneResult, cloneErr := cs.run(ctx, t)
	if cloneErr != nil {
		// cloneResult is still meaningful on a stall or round-limit error —
		// callers inspecting the returned result (e.g. BlobsRecvd) alongside
		// a *PhantomStallError shouldn't see it silently dropped to nil.
		result = cloneResult
		err = cloneErr
		return
	}

	// Crosslink: parse received manifests into event/plink/leaf/mlink tables.
	// Use the ctx-aware variant so a clone deadline can interrupt this
	// whole-repository sweep, which runs after the round loop and so has no
	// round boundary of its own to observe cancellation at.
	linked, xlinkErr := manifest.CrosslinkContext(ctx, r)
	if xlinkErr != nil {
		result = cloneResult
		err = fmt.Errorf("sync.Clone: crosslink: %w", xlinkErr)
		return
	}
	cloneResult.ArtifactsLinked = linked

	return r, cloneResult, nil
}

// run executes the clone loop.
func (cs *cloneSession) run(ctx context.Context, t Transport) (*CloneResult, error) {
	ctx = cs.obs.Started(ctx, SessionStart{
		Operation: "clone",
		Pull:      true,
	})

	prevPhantomCount := -1

	for cycle := 0; ; cycle++ {
		select {
		case <-ctx.Done():
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), ctx.Err())
			return &cs.result, ctx.Err()
		default:
		}
		if cycle >= MaxRounds {
			err := fmt.Errorf("sync.Clone: exceeded %d rounds", MaxRounds)
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
			return &cs.result, err
		}

		roundCtx := cs.obs.RoundStarted(ctx, cycle)

		req, err := cs.buildRequest(cycle)
		if err != nil {
			cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{})
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
			return &cs.result, fmt.Errorf("sync.Clone: buildRequest round %d: %w", cycle, err)
		}

		resp, err := exchangeWithRetry(ctx, t, req, cs.env.Clock, cs.obs)
		if err != nil {
			cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{})
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
			return &cs.result, fmt.Errorf("sync.Clone: exchange round %d: %w", cycle, err)
		}

		recvdBefore := cs.result.BlobsRecvd

		done, err := cs.processResponse(ctx, resp)
		if err != nil {
			cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{FilesReceived: cs.result.BlobsRecvd - recvdBefore})
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
			return &cs.result, fmt.Errorf("sync.Clone: process round %d: %w", cycle, err)
		}

		cs.result.Rounds = cycle + 1
		cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{FilesReceived: cs.result.BlobsRecvd - recvdBefore})

		stop, stopErr := cs.checkStop(cycle, done, &prevPhantomCount)
		if stopErr != nil {
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), stopErr)
			return &cs.result, stopErr
		}
		if stop {
			break
		}
	}

	cs.result.ProjectCode = cs.projectCode
	cs.result.ServerCode = cs.serverCode
	cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), nil)
	return &cs.result, nil
}

// checkStop decides whether the clone loop should stop after processing a
// round. Convergence needs at least two rounds (cycle >= 1), and only once a
// round delivers no new file content (roundDone). *prevPhantomCount tracks
// the phantom count observed at the end of the prior round and is updated
// in place.
//
// A non-nil err means the loop is stopping because it stalled with phantoms
// still outstanding — the phantom count held steady or grew across a round
// that delivered nothing new. That is a failed clone, not a successful one,
// even though continuing would only spin forever (see PhantomStallError).
func (cs *cloneSession) checkStop(cycle int, roundDone bool, prevPhantomCount *int) (stop bool, err error) {
	if prevPhantomCount == nil {
		panic("cloneSession.checkStop: prevPhantomCount must not be nil")
	}

	if cycle < 1 {
		return false, nil
	}
	if !roundDone {
		*prevPhantomCount = len(cs.phantoms)
		return false, nil
	}

	phantomCount := len(cs.phantoms)
	if cs.seqno <= 0 && phantomCount == 0 {
		return true, nil
	}
	if cs.seqno <= 0 && phantomCount > 0 && phantomCount >= *prevPhantomCount {
		return true, newPhantomStallError(cs.phantoms)
	}
	*prevPhantomCount = phantomCount
	return false, nil
}

// sessionEndFromClone builds a SessionEnd from a CloneResult.
func sessionEndFromClone(r *CloneResult) SessionEnd {
	return SessionEnd{
		Operation:   "clone",
		Rounds:      r.Rounds,
		FilesRecvd:  r.BlobsRecvd,
		ProjectCode: r.ProjectCode,
		Errors:      r.Messages,
	}
}

// buildRequest assembles one outbound xfer message for a clone round.
func (cs *cloneSession) buildRequest(cycle int) (*xfer.Message, error) {
	var cards []xfer.Card

	// Pragma: client-version (every round)
	cards = append(cards, &xfer.PragmaCard{
		Name:   "client-version",
		Values: []string{"22800", "20260315", "120000"},
	})

	// Cookie carries the server-captured clone snapshot bound. Echoing it back
	// every round lets the stateless handler enforce a stable upper rid bound
	// on its batch query, so a hub that is being written to during a clone
	// session can't keep extending the queue and prevent convergence.
	if cs.cookie != "" {
		cards = append(cards, &xfer.CookieCard{Value: cs.cookie})
	}

	// Clone card — only when seqno > 0 (sequential delivery in progress).
	// When seqno reaches 0, the server has sent all blobs and the client
	// switches to gimme-based phantom resolution (matching Fossil xfer.c:2706).
	if cs.seqno > 0 {
		version := cs.opts.Version
		if version <= 0 {
			version = 3
		}
		// The clone card carries the pagination cursor itself — `clone
		// VERSION SEQNO`, canonical xfer.c:1553. A clone_seqno card must
		// never go out from here: it is server-to-client only, canonical's
		// page_xfer() has no parser for it, and sending one lands in the
		// server's unknown-card branch as `bad command: clone_seqno N`
		// (issue #74).
		cards = append(cards, &xfer.CloneCard{
			Version:        version,
			SeqNo:          cs.seqno,
			SeqNoIsDecimal: true,
		})
	} else {
		// Pull mode for phantom resolution after sequential delivery completes.
		if cs.projectCode != "" && cs.serverCode != "" {
			cards = append(cards, &xfer.PullCard{
				ServerCode:  cs.serverCode,
				ProjectCode: cs.projectCode,
			})
		}
	}

	// Gimme cards for phantoms — only when seqno <= 1 (main transfer done or finishing).
	if cs.seqno <= 1 {
		gimmes := make([]string, 0, len(cs.phantoms))
		for uuid := range cs.phantoms {
			gimmes = append(gimmes, uuid)
		}
		// BUGGIFY: 5% chance drop the last gimme card.
		if len(gimmes) > 1 && cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.buildRequest.dropGimme", 0.05) {
			gimmes = gimmes[:len(gimmes)-1]
		}
		for _, uuid := range gimmes {
			cards = append(cards, &xfer.GimmeCard{UUID: uuid})
		}
	}

	// Login card: skip round 0. On round 1+, only if User is set AND projectCode received.
	if cycle > 0 && cs.opts.User != "" && cs.projectCode != "" {
		loginCard, err := cs.buildLoginCard(cards)
		if err != nil {
			return nil, fmt.Errorf("clone buildLoginCard: %w", err)
		}
		cards = append([]xfer.Card{loginCard}, cards...)
	}

	return &xfer.Message{Cards: cards}, nil
}

// buildLoginCard encodes the non-login cards, appends a random comment,
// then computes the login card.
func (cs *cloneSession) buildLoginCard(cards []xfer.Card) (*xfer.LoginCard, error) {
	var buf bytes.Buffer
	for _, c := range cards {
		if err := xfer.EncodeCard(&buf, c); err != nil {
			return nil, err
		}
	}
	payload := appendRandomComment(buf.Bytes(), cs.env.Rand)
	login := computeLogin(cs.opts.User, cs.opts.Password, cs.projectCode, payload)
	// BUGGIFY: 5% chance corrupt login nonce to test auth failure recovery.
	if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.buildRequest.badLogin", 0.05) {
		login.Nonce = "corrupted-nonce"
	}
	return login, nil
}

// processResponse handles all cards in a server response for a clone round.
// Returns true when the round produced no new file content.
func (cs *cloneSession) processResponse(ctx context.Context, msg *xfer.Message) (bool, error) {
	if msg == nil {
		panic("sync.Clone.processResponse: msg must not be nil")
	}

	filesRecvd := 0

	for i, card := range msg.Cards {
		if i%processResponseCancelCheckStride == 0 {
			select {
			case <-ctx.Done():
				cs.result.BlobsRecvd += filesRecvd
				return false, ctx.Err()
			default:
			}
		}
		switch c := card.(type) {
		case *xfer.PushCard:
			// Server sends push card with project-code and server-code.
			if c.ProjectCode != "" && cs.projectCode == "" {
				cs.projectCode = c.ProjectCode
				if _, err := cs.repo.DB().Exec(
					"REPLACE INTO config(name, value) VALUES('project-code', ?)",
					c.ProjectCode,
				); err != nil {
					return false, fmt.Errorf("sync.Clone: store project-code: %w", err)
				}
			}
			if c.ServerCode != "" && cs.serverCode == "" {
				cs.serverCode = c.ServerCode
				if _, err := cs.repo.DB().Exec(
					"REPLACE INTO config(name, value) VALUES('server-code', ?)",
					c.ServerCode,
				); err != nil {
					return false, fmt.Errorf("sync.Clone: store server-code: %w", err)
				}
			}

		case *xfer.FileCard:
			content := c.Content
			// BUGGIFY: 2% chance corrupt file content to test hash verification.
			// Relies on blob.Store verify-before-commit to catch corruption.
			if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.corruptHash", 0.02) {
				corrupted := make([]byte, len(content))
				copy(corrupted, content)
				if len(corrupted) > 0 {
					corrupted[0] ^= 0xff
				}
				content = corrupted
			}
			// BUGGIFY: 5% chance skip storing a received file, creating a phantom.
			if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.dropFile", 0.05) {
				filesRecvd++
				continue
			}
			if err := cs.handleFile(ctx, c.UUID, c.DeltaSrc, content, nil); err != nil {
				return false, err
			}
			filesRecvd++

		case *xfer.CFileCard:
			content := c.Content
			storedBlob := c.StoredBlob
			// BUGGIFY: 2% chance corrupt file content to test hash verification.
			// Relies on blob.Store verify-before-commit to catch corruption.
			// storedBlob must be dropped alongside content: it decodes to
			// the *original* bytes, so keeping it would let verbatim
			// storage quietly mask the corruption this path exists to
			// exercise, instead of the corrupted content being what gets
			// hash-checked and stored.
			if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.corruptHash", 0.02) {
				corrupted := make([]byte, len(content))
				copy(corrupted, content)
				if len(corrupted) > 0 {
					corrupted[0] ^= 0xff
				}
				content = corrupted
				storedBlob = nil
			}
			// BUGGIFY: 5% chance skip storing a received file, creating a phantom.
			if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.dropFile", 0.05) {
				filesRecvd++
				continue
			}
			if err := cs.handleFile(ctx, c.UUID, c.DeltaSrc, content, storedBlob); err != nil {
				return false, err
			}
			filesRecvd++

		case *xfer.CloneSeqNoCard:
			// BUGGIFY: 5% chance ignore completion signal, forcing extra round.
			if c.SeqNo == 0 && cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.dropSeqNo", 0.05) {
				continue
			}
			// The server owns this cursor; the client only reads it. A
			// non-decimal NEXT never reaches here — the decoder withholds it
			// per §8.2 so the recorded sequence stays unchanged — and §8.2
			// bars the client from validating that a positive value advances.
			cs.seqno = c.SeqNo

		case *xfer.ErrorCard:
			return false, fmt.Errorf("sync.Clone: server error: %s", c.Message)

		case *xfer.CookieCard:
			cs.cookie = c.Value

		case *xfer.MessageCard:
			cs.result.Messages = append(cs.result.Messages, c.Message)
		}
	}

	cs.result.BlobsRecvd += filesRecvd
	return filesRecvd == 0, nil
}

// handleFile stores a received file. storeReceivedFile now persists a
// delta whose base hasn't arrived yet rather than discarding it (see
// storeDeltaAgainstPhantomBase), so uuid itself is never re-requested here
// once stored — only a still-missing base might need another round.
//
// storedBlob, when non-nil, is payload already re-expressed in Fossil's
// on-disk blob format (see xfer.CFileCard.StoredBlob) and is persisted
// verbatim rather than recompressed — this is what makes a clone
// byte-identical to its source's blob table, not merely content-identical.
func (cs *cloneSession) handleFile(ctx context.Context, uuid, deltaSrc string, payload []byte, storedBlob []byte) error {
	if err := storeReceivedFile(ctx, cs.repo, uuid, deltaSrc, payload, storedBlob, visibilityPublic); err != nil {
		return fmt.Errorf("sync.Clone: handleFile %s: %w", uuid, err)
	}

	delete(cs.phantoms, uuid)
	if deltaSrc != "" {
		// Availability, not existence: a phantom row for deltaSrc still
		// needs requesting, and so does a delta whose own base is a
		// phantom.
		if _, available := content.AvailableByUUID(cs.repo.DB(), deltaSrc); !available {
			cs.phantoms[deltaSrc] = true
		}
	}
	return nil
}