Artifact [adefcef5e6]

Artifact adefcef5e6286d1a56d44b644dd04fe717ecf07c31e4004811ccfbe7161fb18d:

• File internal/sync/clone.go — part of check-in [ddc4b5396d] at 2026-10-12 18:01:26 on branch main — fix(xfer): follow the draft's card grammar, not whole-message failure (#299) (#340)

  The card decoder refused a card by failing the entire message: the server answered HTTP 400 where §5.2 has it emit "bad command" and parse on, and the client dropped a whole reply including its valid prefix.

  The decoder now yields a bad-command card for a line no rule accepts (an unknown operator, a bare pragma, a gimme with a surplus token, a config with a signed size) and a bad-file card for a file/cfile/uvfile line it cannot frame. The server resets its reply, emits the matching error (bad command, or the handler's "malformed file line" and the like) and either parses on (bad command) or stops (bad file), as fossil's page_xfer does; the client ends the reply-card loop, keeping the cards decoded before the bad one; a clone treats it as a non-retryable protocol error.

  Tokens are split on ASCII whitespace only, within fossil's six-token window (blob_tokenize over aToken[6]): a no-break space no longer separates tokens, and a seventh token is ignored rather than rejected, so igot and private tolerate a surplus token. A full cfile's USIZE is carried through as metadata, not checked against the inflated size (the hash binds the content instead), and the byte after a config payload is consumed whatever it is, both matching fossil's xfer_accept_compressed_file and page_xfer.

  This also resolves the bad-command cases tracked in #300 (sync.5.2.bad-command, .bad-command-resets, .unknown-prefix, sync.10.1.bad-command-keeps-mutations), whose marks are removed here. (user: [email protected], size: 22821) [annotate] [blame] [check-ins using]

package sync

import (
	"context"
	"errors"
	"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

// cloneAvailableDeltaSourceLimit bounds the positive-only availability memo
// used while processing received delta cards. Once full, correctness falls
// back to querying the repository for subsequent sources.
const cloneAvailableDeltaSourceLimit = 4096

// 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
	linker                *manifest.ReceiveLinker
	phantoms              map[string]bool
	availableDeltaSources map[string]struct{}
	seqno                 int
	projectCode           string
	serverCode            string
	roundRecvd            int    // file/cfile cards the last reply delivered
	roundPush             bool   // the last reply carried a push card
	cleanup               bool   // the next request is an operation-less cleanup
	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, storage := cloneEnv(opts)

	// 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
		}
	}()

	if err = clearInheritedConfig(r); err != nil {
		return
	}

	linker, linkerErr := manifest.NewReceiveLinker(r)
	if linkerErr != nil {
		err = fmt.Errorf("sync.Clone: create receive linker: %w", linkerErr)
		return
	}

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

	result, err = cs.complete(ctx, t)
	if err != nil {
		return
	}
	return r, result, nil
}

// cloneEnv is the environment a clone runs in and the storage its file
// lives in.
func cloneEnv(opts CloneOpts) (*simio.Env, simio.Storage) {
	env := opts.Env
	if env == nil {
		env = simio.RealEnv()
	}
	storage := env.Storage
	if storage == nil {
		storage = simio.OSStorage{}
	}
	return env, storage
}

// complete runs the clone session, links what it received and fetches any
// artifact it left missing. The result is meaningful on error too: callers
// inspecting it (e.g. BlobsRecvd) alongside a *PhantomStallError shouldn't
// see it dropped to nil.
func (cs *cloneSession) complete(ctx context.Context, t Transport) (*CloneResult, error) {
	cloneResult, err := cs.run(ctx, t)
	if err != nil {
		return cloneResult, err
	}

	// The receive linker has already handled artifacts as their verified
	// content arrived. Finalize retries deferred candidates and runs the
	// bounded residual sweep with that same cache.
	_, finalizeErr := cs.linker.Finalize(ctx)
	cloneResult.ArtifactsLinked = cs.linker.Stats().Linked
	if finalizeErr != nil {
		return cloneResult, fmt.Errorf(
			"sync.Clone: finalize receive-linker crosslink: %w", finalizeErr)
	}
	return cloneResult, cs.fillPhantoms(ctx, t)
}

// clearInheritedConfig drops the config rows a fresh repository is seeded
// with that a clone must take from its source instead.
func clearInheritedConfig(r *repo.Repo) error {
	if r == nil {
		panic("sync.clearInheritedConfig: nil repo")
	}
	// Clear project-code — the server will provide its own.
	if _, execErr := r.DB().Exec("DELETE FROM config WHERE name='project-code'"); execErr != nil {
		return fmt.Errorf("sync.Clone: clear project-code: %w", execErr)
	}

	// Clear hash-policy for the same reason: a clone adopts its source's
	// policy, not the sha3 a fresh repo is seeded with, so a clone of a SHA1
	// repo must not start naming artifacts SHA3. The protocol carries no
	// config, so dropping the row is what conveys it — an absent row reads
	// as "auto", which derives the algorithm from the artifacts that arrive.
	if _, execErr := r.DB().Exec("DELETE FROM config WHERE name='hash-policy'"); execErr != nil {
		return fmt.Errorf("sync.Clone: clear hash-policy: %w", execErr)
	}

	return nil
}

// fillPhantoms fetches artifacts the clone left missing (phantoms) with an
// ordinary pull session. The clone session cannot ask for them: clone_seqno
// 0 bars pull from it, and a gimme without pull fetches nothing (draft
// §8.2). A clone of a consistent repository leaves none, and then no pull
// is made. Any still missing after the pull end the clone with a
// *PhantomStallError, as an incomplete repository is not a clone.
func (cs *cloneSession) fillPhantoms(ctx context.Context, t Transport) error {
	missing, err := phantomUUIDs(cs.repo)
	if err != nil {
		return err
	}
	if len(missing) == 0 {
		return nil
	}
	res, err := Sync(ctx, cs.repo, t, SyncOpts{
		Pull:        true,
		ProjectCode: cs.projectCode,
		ServerCode:  cs.serverCode,
		User:        cs.opts.User,
		Password:    cs.opts.Password,
		Env:         cs.env,
		Buggify:     cs.opts.Buggify,
		Observer:    cs.opts.Observer,
	})
	if res != nil {
		cs.result.Rounds += res.Rounds
		cs.result.BlobsRecvd += res.FilesRecvd
	}
	if err != nil {
		return fmt.Errorf("sync.Clone: pull missing artifacts: %w", err)
	}
	if missing, err = phantomUUIDs(cs.repo); err != nil {
		return err
	}
	if len(missing) > 0 {
		return newPhantomStallError(missing)
	}
	return nil
}

// phantomUUIDs returns the artifacts the repository knows of but lacks.
func phantomUUIDs(r *repo.Repo) (map[string]bool, error) {
	rows, err := r.DB().Query(
		"SELECT b.uuid FROM phantom p JOIN blob b ON p.rid = b.rid WHERE b.size < 0")
	if err != nil {
		return nil, fmt.Errorf("sync.Clone: query phantoms: %w", err)
	}
	missing := make(map[string]bool)
	for rows.Next() {
		var uuid string
		if err := rows.Scan(&uuid); err != nil {
			return nil, errors.Join(fmt.Errorf("sync.Clone: scan phantom: %w", err), rows.Close())
		}
		missing[uuid] = true
	}
	if err := errors.Join(rows.Err(), rows.Close()); err != nil {
		return nil, fmt.Errorf("sync.Clone: read phantoms: %w", err)
	}
	return missing, 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,
	})

	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

		err = cs.processResponse(ctx, cycle, 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})

		next, nextErr := cs.continueAfter(cycle)
		if nextErr != nil {
			cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), nextErr)
			return &cs.result, nextErr
		}
		if !next {
			break
		}
	}

	if cs.projectCode == "" {
		err := fmt.Errorf("sync.Clone: server sent no project code")
		cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
		return &cs.result, err
	}
	cs.result.ProjectCode = cs.projectCode
	cs.result.ServerCode = cs.serverCode
	cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), nil)
	return &cs.result, nil
}

// continueAfter decides, after a reply is processed, whether the clone
// session sends another request and of which kind: draft §3.7, which for a
// clone continues when the round was the first, or stored content (limbs (a)
// and (e); this client sends no private or unversioned cards in a clone, so
// (b) through (d) never hold). The next request reissues `clone 3 NEXT` only
// when the reply carried a push card and the recorded sequence is positive;
// otherwise it is the operation-less cleanup of §8.2, so nothing follows
// clone_seqno 0 but at most one cleanup.
//
// A session that stops while the sequence is still positive did not get the
// whole repository; the draft simply stops, but Clone reports it.
func (cs *cloneSession) continueAfter(cycle int) (bool, error) {
	if cycle < 0 {
		panic("cloneSession.continueAfter: negative cycle")
	}
	if cycle > 0 && cs.roundRecvd == 0 {
		if cs.seqno > 0 {
			return false, fmt.Errorf(
				"sync.Clone: server stopped sending at clone sequence %d", cs.seqno)
		}
		return false, nil
	}
	cs.cleanup = !cs.roundPush || cs.seqno <= 0
	return true, 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})
	}

	// 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). A cleanup request
	// carries no operation card at all (draft §8.2).
	if !cs.cleanup {
		if cs.seqno <= 0 {
			panic("cloneSession.buildRequest: clone reissue after clone_seqno 0")
		}
		version := cs.opts.Version
		if version <= 0 {
			version = 3
		}
		cards = append(cards, &xfer.CloneCard{
			Version:        version,
			SeqNo:          cs.seqno,
			SeqNoIsDecimal: true,
		})
	}

	// 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, signed, err := cs.buildLoginCard(cards)
		if err != nil {
			return nil, fmt.Errorf("clone buildLoginCard: %w", err)
		}
		cards = append([]xfer.Card{loginCard}, signed...)
	} else {
		// Every request ends with randomness (draft §3); a login card's
		// signature already covers the one buildLoginCard appends.
		cards = append(cards, randomComment(cs.env.Rand))
	}

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

// buildLoginCard appends a random comment to cards and computes the login card
// over the result. It returns the login card and the card list it signed, which
// the caller must send verbatim behind that login card — the server recomputes
// the nonce from those bytes.
func (cs *cloneSession) buildLoginCard(cards []xfer.Card) (*xfer.LoginCard, []xfer.Card, error) {
	signed := append(cards[:len(cards):len(cards)], randomComment(cs.env.Rand))
	payload, err := encodeCards(signed)
	if err != nil {
		return nil, nil, err
	}
	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, signed, nil
}

// processResponse handles all cards in a server response for a clone round.
// cycle is that round's 0-based index; round 0 is the one sent before the
// project code — and so the login card — is known.
func (cs *cloneSession) processResponse(ctx context.Context, cycle int, msg *xfer.Message) error {
	if msg == nil {
		panic("sync.Clone.processResponse: msg must not be nil")
	}

	filesRecvd := 0
	cs.roundPush = false
	defer func() { cs.roundRecvd = filesRecvd }()

	for i, card := range msg.Cards {
		if i%processResponseCancelCheckStride == 0 {
			select {
			case <-ctx.Done():
				cs.result.BlobsRecvd += filesRecvd
				return ctx.Err()
			default:
			}
		}
		switch c := card.(type) {
		// §5.2: a reply card the decoder could not read ends the reply-card
		// loop. On a clone it is a protocol defect, not a transport fault, so
		// it aborts the clone without a retry (a retry would only hide it).
		case *xfer.BadCommandCard:
			return fmt.Errorf("sync.Clone: bad command in reply: %q", strings.TrimRight(c.Line, "\n"))
		case *xfer.BadFileCard:
			return fmt.Errorf("sync.Clone: %s", c.Message)

		case *xfer.PushCard:
			cs.roundPush = true
			if err := cs.recordCodes(c); err != nil {
				return err
			}

		case *xfer.FileCard:
			if err := cs.receiveContent(ctx, c.UUID, c.DeltaSrc, c.Content, nil); err != nil {
				return err
			}
			filesRecvd++

		case *xfer.CFileCard:
			err := cs.receiveContent(ctx, c.UUID, c.DeltaSrc, c.Content, c.StoredBlob)
			if err != nil {
				return 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:
			if err := cs.serverError(cycle, c); err != nil {
				return err
			}

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

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

	cs.result.BlobsRecvd += filesRecvd
	return nil
}

// serverError handles an error card. A clone's first round carries no
// login card: the signature is salted with the project code, and the project
// code is exactly what this round exists to learn. A server that requires
// authentication therefore answers round 0 with its push card — which
// supplies that code — followed by "not authorized to clone". Treating that
// as fatal ends the clone before the credentials are ever offered, so round 0
// records the message and carries on; round 1 repeats the request with a
// login card, and an error there (a genuinely wrong password, say) is fatal.
// Canonical does the same: `(syncFlags & SYNC_CLONE)==0 || nCycle>0` guards
// the abort in xfer.c client_sync.
func (cs *cloneSession) serverError(cycle int, c *xfer.ErrorCard) error {
	if c == nil {
		panic("cloneSession.serverError: nil card")
	}
	if cycle == 0 {
		cs.result.Messages = append(cs.result.Messages, c.Message)
		return nil
	}
	return fmt.Errorf("sync.Clone: server error: %s", c.Message)
}

// recordCodes stores the project and server codes a push card carries, the
// first time each arrives.
func (cs *cloneSession) recordCodes(c *xfer.PushCard) error {
	if c == nil {
		panic("cloneSession.recordCodes: nil card")
	}
	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 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 fmt.Errorf("sync.Clone: store server-code: %w", err)
		}
	}
	return nil
}

// receiveContent stores the content a file or cfile card delivered.
// storedBlob is the cfile's payload in fossil's on-disk format, or nil.
func (cs *cloneSession) receiveContent(
	ctx context.Context, uuid, deltaSrc string, content, storedBlob []byte,
) error {
	// 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) {
		return nil
	}
	return cs.handleFile(ctx, uuid, deltaSrc, content, storedBlob)
}

// 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, cs.linker); err != nil {
		return fmt.Errorf("sync.Clone: handleFile %s: %w", uuid, err)
	}

	delete(cs.phantoms, uuid)
	if deltaSrc != "" {
		if _, cached := cs.availableDeltaSources[deltaSrc]; !cached {
			// Availability, not existence: a phantom row for deltaSrc still
			// needs requesting, and so does a delta whose own base is a
			// phantom. Only a confirmed available source is memoized: an
			// unavailable source can arrive later in this same clone.
			if _, available := content.AvailableByUUID(cs.repo.DB(), deltaSrc); !available {
				cs.phantoms[deltaSrc] = true
			} else if len(cs.availableDeltaSources) < cloneAvailableDeltaSourceLimit {
				cs.availableDeltaSources[deltaSrc] = struct{}{}
			}
		}
	}
	return nil
}