Artifact ea97281516855ae4c82e59c64bcc80655a7370b64273cd973aea62ec678f4ee8:
- File
internal/sync/clone.go
— part of check-in
[62683d1d0f]
at
2026-10-12 23:06:19
on branch main
— fix(sync): end client sessions by section 3.7 and keep igot phantoms (#301) (#357)
fossil's client_sync continues a round when stored content or a new phantom leaves the phantom table non-empty, or when files are queued for the next request (section 3.7). The client instead continued on any received file and on any phantom until it had aged three rounds, kept a gimme it could not answer queued forever, and kept an igot's phantom only in the session's map. After a clone's first round it aborted on a server error, where the draft ends the reply-card loop and goes on.
converged now applies 3.7 (a) and (b). A gimme is queued only for content the client holds and may send. An igot's phantom is written to the phantom table, batched once per reply as content_new would write it, and the after-dephantomize cascades of a reply run as one batch (manifest.AfterDephantomizeBatch), so a pull no longer begins a crosslink transaction per artifact. A clone records a later round's server error, stops reading that reply, continues by 3.7, and fails at the end naming the error. Four unit tests that encoded the old continuation now expect fossil's. (user: [email protected], size: 24600) [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/shun" "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 serverErr string // a server error after the first round, which fails the clone 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 := cs.roundCapError() 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 cs.serverErr != "" && (nextErr != nil || !next) { nextErr = fmt.Errorf("sync.Clone: server error: %s", cs.serverErr) } 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 } // roundCapError is the error a clone that ran MaxRounds rounds fails with: // the server's own error when it sent one, since a server that errs every // round is the cause, not the round cap. func (cs *cloneSession) roundCapError() error { if cs.serverErr != "" { return fmt.Errorf("sync.Clone: server error: %s", cs.serverErr) } return fmt.Errorf("sync.Clone: exceeded %d rounds", MaxRounds) } // 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 }() cardLoop: 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: stored, err := cs.receiveCFile(ctx, c) if err != nil { return err } if stored { filesRecvd++ } case *xfer.CloneSeqNoCard: cs.recordSeqNo(c) case *xfer.ErrorCard: if cs.serverError(cycle, c) { break cardLoop } 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 } // receiveCFile stores a cfile and reports whether it did. fossil drops a // shunned cfile even in a clone, though not a shunned file card (xfer.c:162, // !cloneFlag), so a shunned one is skipped and not counted. func (cs *cloneSession) receiveCFile(ctx context.Context, c *xfer.CFileCard) (bool, error) { shunned, err := shun.IsShunned(cs.repo.DB(), c.UUID) if err != nil || shunned { return false, err } err = cs.receiveContent(ctx, c.UUID, c.DeltaSrc, c.Content, c.StoredBlob) return err == nil, err } // recordSeqNo records the clone sequence a clone_seqno card carries. 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. func (cs *cloneSession) recordSeqNo(c *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) { return } cs.seqno = c.SeqNo } // serverError handles an error card and reports whether it ends the // reply-card loop. 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. // Canonical does the same: `(syncFlags & SYNC_CLONE)==0 || nCycle>0` guards // the abort in xfer.c client_sync. // // After round 0 an error ends the reply-card loop (§10.1): the cards after it // are not read, and the session continues by §3.7 on what was stored before // it. Clone still fails once the session ends, naming the error, so a clone // cut short by the server is never reported complete. func (cs *cloneSession) serverError(cycle int, c *xfer.ErrorCard) bool { if c == nil { panic("cloneSession.serverError: nil card") } cs.result.Messages = append(cs.result.Messages, c.Message) if cycle == 0 { return false } if cs.serverErr == "" { cs.serverErr = c.Message } return true } // 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 { // request_phantoms never asks for a shunned name. shunned, err := shun.IsShunned(cs.repo.DB(), deltaSrc) if err != nil { return fmt.Errorf("sync.Clone: basis of %s: %w", uuid, err) } if !shunned { cs.phantoms[deltaSrc] = true } } else if len(cs.availableDeltaSources) < cloneAvailableDeltaSourceLimit { cs.availableDeltaSources[deltaSrc] = struct{}{} } } } return nil }