Artifact fb03c8ac57fd63e912abff3c9f071ca6c6ec1440efe689c81e920e02d1f99b67:
- File
internal/sync/clone.go
— part of check-in
[6963b0a9dc]
at
2026-10-12 23:10:13
on branch main
— fix(sync): exchange server-version and gate SHA3 content by peer release (#303) (#358)
fossil's page_xfer answers a client-version that carries DATE and TIME with pragma server-version, and both sides record the peer's release; send_file, send_compressed_file and send_unversioned_file then append "error Fossil version 2.0 or later required." in place of SHA3-named content for a peer older than 2.0, with a server treating a client that announced no release as one (sections 7.6, 9.1). The server sent no server-version and neither side kept or gated on the peer's release.
The server gates gimme, clone and uvgimme; the client gates its answer to a gimme. Because a server-version from a current release turns fossil's client to logging in through the x-f-l-c cookie (section 6.5), the HTTP server now reads that cookie: xfer.DecodeRequest puts its login card first with the nonce bound to the whole body, as page_xfer does. Without it a real fossil client's later rounds arrived as nobody. That moves sync.6.5.server-cookie-login, marked for #304, here. (user: [email protected], size: 24532) [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, versionPragma("client-version")) // 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 }