Artifact 51dae6e7df562df632a2c0f36851a93358f20c240aec62acba2ab08c64d33438:
- File
internal/sync/handler.go
— part of check-in
[479df656d3]
at
2026-10-13 04:56:41
on branch fix/sync-server-order
— fix(sync): handle server cards in request order, phantom gimme after push, clone bad cards
Tracked in #385.
The server verified the login card first and handled every control card before any data card. A file card ahead of its push card was stored, and cards ahead of a successful login got the login's capabilities. Cards are now handled in one in-order loop, as fossil's page_xfer reads them. A request holding an invalid clone sequence is first walked without side effects, which stands in for fossil's whole-request rollback.
A refused push without an earlier pull now resets and answers "not authorized to write"; a refused pull resets, answers "not authorized to read" and ends card processing (sync-protocol-01 Section 6.6).
An accepted push now asks for the server's phantoms after the card loop, at most 500, skipping what the client asked for, shunned names and private ones outside a private sync, as request_phantoms does. A pushed igot makes a phantom instead of drawing a gimme on the spot.
A clone reply card the client cannot read ends the reply-card loop instead of aborting: the session goes on, the second clone round is still made, and the clone then fails naming the card.
The conformance matchers that accepted "push denied" and "pull denied" are narrowed to the draft's text, and new cases put output ahead of a refusal and cards ahead of the push or login they depend on. (user: [email protected], size: 64049) [annotate] [blame] [check-ins using] [more...]
package sync import ( "bytes" "context" "database/sql" "errors" "fmt" "strconv" "strings" "github.com/danmestas/go-libfossil/db" "github.com/danmestas/go-libfossil/internal/auth" "github.com/danmestas/go-libfossil/internal/blob" "github.com/danmestas/go-libfossil/internal/content" "github.com/danmestas/go-libfossil/internal/hash" "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" libfossil "github.com/danmestas/go-libfossil/internal/fsltype" ) // DefaultCloneBatchBytes bounds the wire bytes a single clone round emits. // // A clone batch is bounded by output size, never by artifact count. A count // cannot adapt to artifact size: 200 tiny control artifacts and 200 large // blobs produce wildly different messages, and neither corresponds to what // the transport can carry. §8.2 leaves the bound entirely to the server // ("clients MUST NOT infer count or size from range"), so the only real // constraint is that the response stays a size a server can hold and that a // repository drains in rounds proportional to its bytes, not its artifacts. // // emitCloneBatch now retransmits stored content per §7.2 (issue #98) rather // than the fully expanded artifacts it used to send, so bytesSent tracks the // same quantity canonical's 5,000,000 max-download bounds: repository bytes, // not the 80x-220x-larger expanded form. This constant was chosen before that // fix, against the expanded-bytes regime documented in issue #88 where a // smaller value pushed fossil- and sqlite-sized corpora past MaxRounds -- // that constraint is gone now that wire bytes and repository bytes are the // same quantity. Retuning this value toward canonical's constant is issue // #109's territory (server pagination/budget accounting), not this fix's; it // is left at its prior value here to avoid colliding with that work in // flight. 16,000,000 stays a safe, if generous, upper bound either way: it // only widens how much stored content one round may carry. // // This bounds an ordinary round at the budget plus the artifact that crosses // it, and emitCloneBatch caps that sum at xfer.MaxDecompressedBytes so the // client can always decode a round (issue #109). The one exception is a single // artifact larger than that bound: it is sent alone, in a round of its own, to // make progress -- so a repository holding one 2 GB artifact still emits a 2 GB // round, but such an artifact is unclonable regardless. Measured worst case is // ~25 MB across the three repositories above, since the largest single // artifacts are 9.4 MB (libfossil) and 17.0 MB (sqlite). Against the count // bound's unbounded 44.9 MB that is a large reduction, and concurrent clones // multiply it. const DefaultCloneBatchBytes = 16_000_000 // A round this server emits must stay inside the bound the client applies when // it decompresses one, or a clone between two libfossil peers fails on a // message this same code produced (issue #104). emitCloneBatch enforces that at // runtime: it flushes a round rather than let an artifact carry it past // xfer.MaxDecompressedBytes, and sends an over-bound artifact alone (issue // #109). This compile-time guard makes that enforcement well-conditioned: with // the bound at least twice the budget, filler accumulates up to one budget and // any artifact up to a second budget still fits in the same round, so ordinary // artifacts ride along and only genuinely large ones (above bound minus budget) // take a round of their own. Raising DefaultCloneBatchBytes past half of // xfer.MaxDecompressedBytes underflows this unsigned expression and fails the // build rather than the clone. const _ = uint(xfer.MaxDecompressedBytes - 2*DefaultCloneBatchBytes) // cloneCardOverheadBytes over-approximates the non-payload bytes of an // encoded cfile card -- keyword, hash, delta-source hash, two decimal // lengths, and newline come to somewhat more than the plain "file" card's // ~58 -- so that a repository of zero-length artifacts still paginates // instead of emitting the whole blob table into a single response. The slack // also absorbs the PrivateCard that may precede a card, which is not charged // separately. This is charged against data before encodeCFile's zlib // compression runs, so it is already conservative in the same direction as // the framing overhead it approximates. Over-charging is the safe direction // and cannot be defeated: a corpus of nothing but zero-length artifacts caps // at ~167,000 cards per round, about 9.7 MB of actual output, still inside // the budget. const cloneCardOverheadBytes = 96 // cloneCapPrefix tags a CookieCard value that carries the upper rid bound // of a clone session. The handler captures max(rid) on the first clone round // and echoes it via CookieCard; the client returns it on every subsequent // round so the (otherwise stateless) server can scope its blob query and // converge even if the hub is being written to during the clone session. const cloneCapPrefix = "lf-clone-cap:" // parseCloneCap returns the snapshot rid encoded in a CookieCard value, // or (0, false) if the value isn't a clone-cap cookie. Non-matching // cookies (e.g. arbitrary sync session cookies) leave the bound at zero, // which preserves pre-snapshot behavior for forward/back compat. func parseCloneCap(v string) (int, bool) { if !strings.HasPrefix(v, cloneCapPrefix) { return 0, false } rid, err := strconv.Atoi(v[len(cloneCapPrefix):]) if err != nil || rid < 0 { return 0, false } return rid, true } // HandleFunc is the server-side sync handler signature. // Transport listeners call this with decoded requests and write back the response. type HandleFunc func(ctx context.Context, r *repo.Repo, req *xfer.Message) (*xfer.Message, error) // HandleOpts configures optional behavior for HandleSync. type HandleOpts struct { Buggify BuggifyChecker // nil in production. Observer Observer // nil defaults to no-op. ContentCache *content.Cache // nil = no caching. Clock simio.Clock // time for the reply's timestamp comment; nil = real time } // HandleSync processes an incoming xfer request and produces a response. // Stateless per-round — the client drives convergence. func HandleSync(ctx context.Context, r *repo.Repo, req *xfer.Message) (*xfer.Message, error) { return HandleSyncWithOpts(ctx, r, req, HandleOpts{}) } // HandleSyncWithOpts processes an incoming xfer request with optional // fault injection. Used by DST harness; production callers use HandleSync. func HandleSyncWithOpts(ctx context.Context, r *repo.Repo, req *xfer.Message, opts HandleOpts) (*xfer.Message, error) { if r == nil { panic("sync.HandleSync: r must not be nil") } if req == nil { panic("sync.HandleSync: req must not be nil") } obs := resolveObserver(opts.Observer) ctx = obs.HandleStarted(ctx, HandleStart{ Operation: detectOperation(req), }) h := &handler{repo: r, buggify: opts.Buggify, cache: opts.ContentCache, clock: opts.Clock} if h.clock == nil { h.clock = simio.RealClock{} } resp, err := h.process(ctx, req) if err == nil && resp == nil { panic("sync.HandleSync: resp must not be nil on success") } obs.HandleCompleted(ctx, HandleEnd{ CardsProcessed: len(req.Cards), FilesSent: h.filesSent, FilesReceived: h.filesRecvd, Err: err, }) return resp, err } // detectOperation checks request cards to determine if this is a clone or sync. func detectOperation(req *xfer.Message) string { for _, c := range req.Cards { if _, ok := c.(*xfer.CloneCard); ok { return "clone" } } return "sync" } // remoteHasEntry records that the client announced a blob via igot. type remoteHasEntry struct { isPrivate bool // IsPrivate flag from the client's igot card } // handler holds per-request state while processing cards. type handler struct { repo *repo.Repo buggify BuggifyChecker resp []xfer.Card pushOK bool // client sent a valid push card pullOK bool // client sent a valid pull card cloneMode bool // client sent a clone card fatal bool // a card rule ended the request; resp holds only its error broke bool // card loop broken; the post-loop still runs errsReset int // counted errors a reset discarded (§5.3 errors N) projectCode string // the repository's project-code serverCode string // the repository's server-code uncompressed bool // reply goes out uncompressed (clone v3) cloneDone bool // clone card has no usable cursor: send nothing peerVersion int // client's release, from client-version peerDate int // client's check-in date, from client-version cloneSeq int // rid of the next blob to send, from the client's clone card cloneSnapMax int // upper rid bound for this clone session (from cookie); 0 = capture fresh in emitCloneBatch uvCatalogSent bool // true after sending UV catalog pendingPhantoms []pendingPhantom // igot names awaiting flushPhantoms reqClusters bool // client sent pragma req-clusters filesSent int // files sent in response (for observer) filesRecvd int // files received from client (for observer) syncPrivate bool // true if pragma send-private was accepted nextIsPrivate bool // true if a private card precedes the next file/cfile syncedTables map[string]*SyncedTable // cached table definitions xrowsSent int // table sync rows sent xrowsRecvd int // table sync rows received cache *content.Cache // nil = passthrough to content.Expand clock simio.Clock // time for the reply's timestamp comment remoteHas map[string]remoteHasEntry // UUIDs the client announced via igot (mirrors Fossil's onremote table) remoteLacks map[string]bool // UUIDs the client asked for (fossil's unk) logins int // login cards read so far; only one is allowed // Auth state user string // verified username ("nobody" if no login card) caps string // capability string from user table authed bool // whether login card was verified // cloneRead is Read derived from Clone before any login, as page_xfer // derives it from nobody's capabilities ahead of its card loop. cloneRead bool } func (h *handler) initAuth() { h.user = "nobody" h.caps = "" h.authed = false var caps string err := h.repo.DB().QueryRow("SELECT cap FROM user WHERE login='nobody'").Scan(&caps) if err == nil { h.caps = caps } h.cloneRead = auth.CanClone(h.caps) } // canRead reports the requester's Read permission for this request, the one // owner of that decision for every read gate the handler applies: the // unversioned catalog, uvigot and uvgimme answers, xtable-hash and // read-gated reqconfig. // // Clone implies Read for the capabilities a request starts with, nobody's, // as fossil's page_xfer sets it (`if( g.perm.Clone ) g.perm.Read = 1;`, // xfer.c:1301) before its card loop and the §7.3 erratum states. A Clone // that a login card grants does not become Read: check_login only adds the // user's capabilities, after that point. // // Pull is deliberately not one of these gates. fossil trunk's derived Read // also lets a Clone-only nobody pull; here a pull still needs o // (auth.CanPull), by the owner's decision. func (h *handler) canRead() bool { return auth.CanRead(h.caps) || h.cloneRead } // handleLoginCard verifies a login card and adds the user's capabilities to // nobody's, as fossil's check_login does through login_set_capabilities: a // login only grants. It reports the refusal's error text, "" when the login // is accepted. fossil does not go on serving the request as nobody after a // refusal, and nor does this: the caller ends card processing there. func (h *handler) handleLoginCard(c *xfer.LoginCard) string { // A login's shared secret is salted with the project code; a repository // without one can verify no login, and fossil's page_xfer stops there // too ("missing project code"). Refuse it rather than ask VerifyLogin, // which takes an empty project code as a programmer error. if h.projectCode == "" { return "login failed" } u, err := auth.VerifyLogin(h.repo.DB(), h.projectCode, c) if err != nil { return "login failed" } h.user = u.Login h.caps += u.Cap h.authed = true return "" } // handleLogin answers a login card at its own position in the request, as // page_xfer's check_login runs where the card is read: the capabilities it // grants apply to the cards after it, never to those before. Only one is // allowed (§6.1, src/xfer.c "error multiple\slogin\scards"). A refusal is // page_xfer's reset, error and break, so the cards before it keep their // effect and the end-of-request steps still run (§6.4). func (h *handler) handleLogin(c *xfer.LoginCard) { if c == nil { panic("handler.handleLogin: c must not be nil") } h.logins++ if h.logins > 1 { h.breakLoop("multiple login cards") return } if msg := h.handleLoginCard(c); msg != "" { h.breakLoop(msg) } } // fail ends the request with a single error card, discarding the reply built // so far, as fossil's cgi_reset_content before an error does. func (h *handler) fail(msg string) { h.resetContent() h.resp = []xfer.Card{&xfer.ErrorCard{Message: msg}} h.fatal = true } // resetContent discards the reply built so far, as fossil's cgi_reset_content // does before a card-level error. Only the reply is cleared; any blob already // stored this round stays (fossil keeps its mutations too, §10.1). The errors // it discards are still counted in the closing comment (§5.3). func (h *handler) resetContent() { h.errsReset += countedErrors(h.resp) h.resp = nil } // badCommandPrefix begins the error that answers an unknown or malformed card. const badCommandPrefix = "bad command: " // badCommand answers a line no card rule accepts. §5.2, src/xfer.c page_xfer's // final "else": the reply so far is reset, an "error bad\scommand:\s<LINE>" // card is emitted, and parsing goes on -- so a later card still takes effect. // line keeps its terminating LF, which the error quotes verbatim. func (h *handler) badCommand(line string) { h.resetContent() h.resp = append(h.resp, &xfer.ErrorCard{Message: badCommandPrefix + line}) } // replyOnlyLine re-encodes a card the server should not have received (error, // message -- reply-only in fossil's page_xfer, which has no branch for them) // into the wire line it arrived as, so badCommand can quote it the way fossil // quotes blob_str(&xfer.line). The LF EncodeCard writes is the one the error // text carries. func replyOnlyLine(c xfer.Card) string { var buf bytes.Buffer if err := xfer.EncodeCard(&buf, c); err != nil { // error and message cards always encode; this is a programmer error. panic("handler.replyOnlyLine: " + err.Error()) } return buf.String() } // breakLoop answers a card with fossil's cgi_reset_content, error and break: // the reply so far is reset to one error card and no later request card is // processed, but the post-loop work and the timestamp comment still follow, // as they do after page_xfer's card loop. func (h *handler) breakLoop(msg string) { h.resetContent() h.resp = append(h.resp, &xfer.ErrorCard{Message: msg}) h.broke = true // The reset also discards the cfiles page_xfer's clone card already // sent, so no clone batch or clone push card follows the error, and the // cluster igots it answered req-clusters with in place. h.cloneMode = false h.reqClusters = false } func (h *handler) process(ctx context.Context, req *xfer.Message) (*xfer.Message, error) { // Initialize auth state from nobody user. h.initAuth() // A login's shared secret and every push and pull card are checked // against the project code. A repository without one leaves it empty, // which no artifact hash matches. if err := h.loadCodes(); err != nil { return nil, err } // Load synced tables. if err := h.loadSyncedTables(); err != nil { return nil, err } if h.probeCloneFatal(req.Cards) { return h.reply(), nil } if err := h.processCards(ctx, req.Cards); err != nil { return nil, err } if h.fatal { return h.reply(), nil } // The end-of-request steps run after a broken loop too (§6.4). if err := h.emitInventory(); err != nil { return nil, err } if h.cloneMode { if err := h.emitClonePush(); err != nil { return nil, err } } if err := h.crosslinkReceived(ctx); err != nil { return nil, err } h.resp = append(h.resp, h.timestampComment()) return h.reply(), nil } // processCards handles the request's cards in request order, as page_xfer's // card loop reads them, until one ends the request (h.fatal) or breaks the // loop. Each card sees only the state the cards before it established: a // file card before its push card is refused, and a card before a login card // has nobody's capabilities (§5.2, §6.4, §6.6). func (h *handler) processCards(ctx context.Context, cards []xfer.Card) error { for _, card := range cards { // Phantoms from a run of igot cards are written in one transaction, // as fossil writes them inside page_xfer's single write // transaction; any other card sees them already stored. if _, isIGot := card.(*xfer.IGotCard); !isIGot { if err := h.flushPhantoms(); err != nil { return err } } if err := h.processCard(ctx, card); err != nil { return err } if h.fatal || h.broke { break } } if h.fatal { // A fatal rolls the request back (§8.1): nothing pending is kept. h.pendingPhantoms = nil return nil } return h.flushPhantoms() } // processCard handles one request card. func (h *handler) processCard(ctx context.Context, card xfer.Card) error { if card == nil { panic("handler.processCard: card must not be nil") } switch c := card.(type) { case *xfer.LoginCard: h.handleLogin(c) case *xfer.PragmaCard, *xfer.PushCard, *xfer.PullCard, *xfer.CloneCard, *xfer.CloneSeqNoCard, *xfer.CookieCard, *xfer.SchemaCard: h.handleControlCard(card) case *xfer.BadFileCard: // A file/cfile/uvfile line the decoder could not frame: page_xfer // resets its reply, emits the handler's error and breaks (§5.2). h.breakLoop(c.Message) case *xfer.BadCommandCard: h.badCommand(c.Line) case *xfer.ErrorCard, *xfer.MessageCard: // error and message are reply-only: a client that sends one gets a // bad command, as fossil's page_xfer has no branch for them (§5.2). h.badCommand(replyOnlyLine(c)) default: return h.handleDataCard(ctx, card) } return nil } // probeCloneFatal answers a request that reaches §8.1's invalid clone // sequence fatal, and reports whether it did. page_xfer answers that fatal // by rolling back the whole request, so nothing the cards before it stored // survives. This handler stores as it goes, with no request transaction to // roll back, so a request holding such a clone card is first walked without // side effects: only the cards that decide whether the clone card is reached, // and with which capabilities, are acted on. When the walk ends in the // fatal, its reply is the answer and no card has written anything. // // The walk stores no file, so a file card that fails validation does not // break it as it breaks the real loop. A request that pushes a bad file // ahead of an invalid clone card therefore gets the clone error where // fossil sends the file's; no conforming client sends push and clone // together (§8.5). func (h *handler) probeCloneFatal(cards []xfer.Card) bool { if !hasInvalidCloneSeq(cards) { return false } probe := &handler{repo: h.repo, clock: h.clock, projectCode: h.projectCode, serverCode: h.serverCode} probe.initAuth() for _, card := range cards { probe.probeCard(card) if probe.fatal || probe.broke { break } } if !probe.fatal { return false } h.resp, h.fatal = probe.resp, true h.uncompressed = probe.uncompressed return true } // hasInvalidCloneSeq reports whether a clone card in cards would meet // §8.1's fatal if an authorized requester reached it. func hasInvalidCloneSeq(cards []xfer.Card) bool { for _, card := range cards { c, ok := card.(*xfer.CloneCard) if ok && c.SeqNoIsDecimal && c.Version >= 2 && c.SeqNo <= 0 { return true } } return false } // probeCard is processCard for probeCloneFatal: the cards that can end the // card loop or change the capabilities, none of which writes to the // repository. func (h *handler) probeCard(card xfer.Card) { if card == nil { panic("handler.probeCard: card must not be nil") } switch c := card.(type) { case *xfer.LoginCard: h.handleLogin(c) case *xfer.PushCard: h.handlePushCard(c) case *xfer.PullCard: h.handlePullCard(c) case *xfer.CloneCard: h.handleCloneCard(c) case *xfer.FileCard, *xfer.CFileCard: if !h.pushOK { h.breakLoop("not authorized to write") } case *xfer.BadFileCard: h.breakLoop(c.Message) case *xfer.ConfigCard: h.handleConfigPush() } } // crosslinkReceived crosslinks any newly-received manifests into the // relational tables (event/leaf/plink/mlink/tagxref). Without this the // receiving repo stores blobs durably but exposes nothing on the timeline, // fork detection sees stale leaf state, and the clone protocol cannot // traverse manifest parents (mlink is empty). // // It walks the scanner only when we accepted files this round; pure // pull/igot rounds add nothing relational to update. It is context-aware // (#120): the sweep walks the whole repository in one call, so on a large // repo it is where a request whose client has already gone away would // otherwise keep working uninterrupted. func (h *handler) crosslinkReceived(ctx context.Context) error { if h.filesRecvd == 0 { return nil } if _, err := manifest.CrosslinkContext(ctx, h.repo); err != nil { return fmt.Errorf("HandleSync: crosslink: %w", err) } return nil } // reply is the message the handler answers with. func (h *handler) reply() *xfer.Message { return &xfer.Message{Cards: h.resp, Uncompressed: h.uncompressed} } // loadCodes reads the repository's project and server codes. A login's // shared secret and every push and pull card are checked against the project // code; a clone's push card carries both. A repository without one leaves it // empty, which no artifact hash matches. func (h *handler) loadCodes() error { for _, c := range []struct { name string dst *string }{{"project-code", &h.projectCode}, {"server-code", &h.serverCode}} { err := h.repo.DB().QueryRow( "SELECT value FROM config WHERE name=?", c.name).Scan(c.dst) if err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("HandleSync: %s: %w", c.name, err) } } return nil } // emitClonePush emits the push card with project-code and server-code so // the clone client can identify the repo. Only in clone mode — sync clients // already have codes, and real Fossil treats server-sent "push" as unknown // during sync. // // This must trail the clone batch's clone_seqno card, matching canonical's // order (fossil-scm xfer.c emits the batch, then clone_seqno at :1571, then // push at :1577). A real fossil client re-issues `clone 3 SEQNO` every round // it sees a `push` card while its clone cursor is still positive (xfer.c:2706), // and it learns the cursor hit zero only from the clone_seqno card. Emitting // push ahead of clone_seqno makes the client queue one more clone request // before it sees the terminal clone_seqno 0, so the server serves the whole // repository a second time — the 2.06x end-to-end blow-up of issue #138. func (h *handler) emitClonePush() error { if !h.cloneMode { panic("handler.emitClonePush: not a clone") } if h.projectCode != "" { h.resp = append(h.resp, &xfer.PushCard{ ProjectCode: h.projectCode, ServerCode: h.serverCode, }) } return nil } // timestampComment is the comment that ends a completed reply, as fossil's // page_xfer writes it: the server's UTC time and the request's error count // (§5.3). A fail-fast reply goes without it. func (h *handler) timestampComment() *xfer.CommentCard { if h.clock == nil { panic("handler.timestampComment: nil clock") } if h.errsReset < 0 { panic("handler.timestampComment: negative reset error count") } errs := h.errsReset + countedErrors(h.resp) now := h.clock.Now().UTC().Format("2006-01-02T15:04:05") return &xfer.CommentCard{Text: fmt.Sprintf("timestamp %s errors %d", now, errs)} } // privateRefusal is the error a requester without the private capability // gets for pragma send-private or a private card. const privateRefusal = "not authorized to sync private content" // countedErrors counts the error cards that §5.3's errors N counts. N is the // errors raised, as page_xfer's nErr is, and three of them fossil never adds // to nErr: the unknown-operator error, the private-content refusal and the // SHA3 version refusal (erratum sync-protocol-01-s5.3-errors-count). The // count is made from card text because those three texts are fixed by the // draft and no other error carries them. func countedErrors(cards []xfer.Card) int { n := 0 for _, c := range cards { e, ok := c.(*xfer.ErrorCard) if !ok { continue } switch { case strings.HasPrefix(e.Message, badCommandPrefix), e.Message == privateRefusal, e.Message == sha3Refusal().Message: continue } n++ } return n } // emitInventory sends what follows the card loop: the cluster igots a // req-clusters asked for, the igot inventory of a pull, and a clone's batch. func (h *handler) emitInventory() error { // pragma req-clusters is answered with the cluster igots, ahead of the // inventory as page_xfer answers it in card order. Like fossil, any // requester gets them: they name public artifacts and carry no content, // which a gimme still needs Read to fetch. if h.reqClusters { if err := h.sendAllClusters(); err != nil { return err } } // An accepted push asks for the server's phantoms, ahead of the pull's // inventory as page_xfer's request_phantoms runs ahead of // send_unclustered (§6.4, §8.5). if h.pushOK { if err := h.requestPhantoms(); err != nil { return err } } // If pull was requested, emit igot for all non-phantom blobs. if h.pullOK { if err := h.emitIGots(); err != nil { return err } if err := h.emitXIGots(); err != nil { return err } } // If clone, emit paginated file cards, after the unversioned catalog // page_xfer sends with a clone when uv-sync is on (§8.4). if h.cloneMode { if err := h.sendCloneUVCatalog(); err != nil { return err } if h.cloneDone { h.resp = append(h.resp, &xfer.CloneSeqNoCard{SeqNo: 0}) return nil } return h.emitCloneBatch() } return nil } func (h *handler) handleControlCard(card xfer.Card) { switch c := card.(type) { case *xfer.PragmaCard: h.handlePragma(c) case *xfer.PushCard: h.handlePushCard(c) case *xfer.PullCard: h.handlePullCard(c) case *xfer.CloneCard: h.handleCloneCard(c) case *xfer.CloneSeqNoCard: // Deliberately ignored. clone_seqno is a reply-only card, so no // conforming client sends one; canonical answers `bad command`. We // stay tolerant rather than strict, which is why the model server in // TestCloneMultiBatchAgainstCanonicalServer rejects it and we do not. case *xfer.CookieCard: // Clone clients echo the cap cookie on every round. Sync sessions also // use cookies opaquely; non-matching values leave cloneSnapMax at zero. if rid, ok := parseCloneCap(c.Value); ok { h.cloneSnapMax = rid } case *xfer.SchemaCard: h.handleSchemaCard(c) } } // handlePushCard accepts a push from a requester with Write, after the // project check. func (h *handler) handlePushCard(c *xfer.PushCard) { if h.wrongProject(c.ProjectCode) { return } if auth.CanPush(h.caps) { h.pushOK = true return } if h.pullOK { // A requester who may read but not write is told to pull only, // not refused, as fossil's page_xfer answers a push card with // "message pull only - not authorized to push" once a pull card // has set isPull. The pull still proceeds (§10.2). h.resp = append(h.resp, &xfer.MessageCard{ Message: "pull only - not authorized to push", }) return } // Otherwise the reply so far is reset and the push refused, but card // processing goes on (§6.6): page_xfer has no break here. h.resetContent() h.resp = append(h.resp, &xfer.ErrorCard{Message: "not authorized to write"}) } // handlePullCard accepts a pull from a requester with Read, after the project // check. func (h *handler) handlePullCard(c *xfer.PullCard) { if h.wrongProject(c.ProjectCode) { return } if auth.CanPull(h.caps) { h.pullOK = true return } // page_xfer resets, refuses and breaks (§6.6, §10.1). h.breakLoop("not authorized to read") } // handleCloneCard starts a clone for a requester with Clone, at the rid the // card's SEQNO names. func (h *handler) handleCloneCard(c *xfer.CloneCard) { if !auth.CanClone(h.caps) { h.refuseClone() return } // §4.1, §8.1: page_xfer sends a version 3 clone's reply uncompressed, // and sets that before the sequence-number fatal below. h.uncompressed = c.Version >= 3 // §8.1: an otherwise authorized three-token clone with VERSION >= 2 // and a digit-only SEQNO of zero or less clears accumulated output, // rolls back, emits this error, and parses no later request card. // probeCloneFatal reaches this ahead of the card loop, before any card // has written, which is what stands in for the rollback. if c.SeqNoIsDecimal && c.Version >= 2 && c.SeqNo <= 0 { h.fail("invalid clone sequence number") return } h.cloneMode = true // The pagination cursor rides on the clone card and nowhere else // (§8.1; canonical xfer.c:1553 reads token 2 of `clone VERSION // SEQNO`). §8.1 withholds the fatal above from a negative SEQNO, which // fails digit-only recognition; page_xfer then has no cursor at all, so // no content is sent and the reply says the clone is done. A bare // `clone` starts from the first rid. if c.Version >= 2 && c.SeqNo < 0 { h.cloneDone = true return } h.cloneSeq = c.SeqNo if h.cloneSeq <= 0 { h.cloneSeq = 1 } } // refuseClone answers a clone from a requester without Clone as page_xfer // does (§8.2, §10.1): the reply so far is reset, the push card naming the // repository goes ahead of the error, and no later card is processed. func (h *handler) refuseClone() { h.resetContent() serverCode := h.serverCode if serverCode == "" { serverCode = "x" // db_get("server-code", "x") } h.resp = append(h.resp, &xfer.PushCard{ServerCode: serverCode, ProjectCode: h.projectCode}, &xfer.ErrorCard{Message: "not authorized to clone"}) h.broke = true } // wrongProject refuses a push or pull card naming another project, ahead of // any capability check, as page_xfer does with "error wrong\sproject" and a // break (§8.5). The servercode beside it is informational and not checked. // Cards are handled in request order, so a file card ahead of it has been // stored, as fossil stores it. func (h *handler) wrongProject(code string) bool { if code == h.projectCode { return false } h.breakLoop("wrong project") return true } // handlePragma acts on the pragmas the server understands and ignores the // rest, client-version included. // // Pragmas are handled in card order, so a pull card later in the request is // not yet known here. The catalogs are therefore gated on // the Read capability itself, as fossil's page_xfer gates uv-hash on // g.perm.Read, rather than on an accepted pull. func (h *handler) handlePragma(c *xfer.PragmaCard) { if c == nil { panic("handler.handlePragma: c must not be nil") } switch { case c.Name == "client-version" && len(c.Values) >= 1: // §9.1: the client's release gates SHA3 content (§7.6). A // client-version carrying DATE and TIME is answered in place, as // page_xfer answers it, so the answer precedes any content. h.peerVersion = peerVersion(c) h.peerDate = peerDate(c) if len(c.Values) >= 3 { h.resp = append(h.resp, versionPragma("server-version")) } case c.Name == "uv-hash" && len(c.Values) >= 1: if err := h.handlePragmaUVHash(c.Values[0]); err != nil { h.resp = append(h.resp, &xfer.ErrorCard{ Message: fmt.Sprintf("uv-hash: %v", err), }) } case c.Name == "xtable-hash" && len(c.Values) >= 2: // The inventory serves both directions of table sync: xgimme // pulls rows from it, and a pushing client queues the rows it // holds newer (client_tablesync.go handleXIGotResponse). So it // needs Read and a direction card, not xgimme's pull alone. A // Clone-only requester has Read by derivation (#312) but sends no // direction card, so it gets no inventory (#351). Table sync is a // go-libfossil extension with no fossil counterpart. if h.canRead() && (h.pullOK || h.pushOK) { h.handlePragmaXTableHash(c.Values[0], c.Values[1]) } case c.Name == "req-clusters": h.reqClusters = true case c.Name == "ci-lock" && len(c.Values) >= 2: // Fossil takes a lock for any requester. A lock only matters to // someone who can commit, and ci-lock-fail names the holder, so // this server takes and reports locks for writers alone. if !auth.CanPush(h.caps) { return } fail := processCkinLock(h.repo.DB(), c.Values[0], c.Values[1], h.user, DefaultCkinLockTimeout) if fail != nil { h.resp = append(h.resp, &xfer.PragmaCard{ Name: "ci-lock-fail", Values: []string{fail.HeldBy, fmt.Sprintf("%d", fail.Since.Unix())}, }) } case c.Name == "ci-unlock" && len(c.Values) == 1 && hash.IsValidHash(c.Values[0]): // fossil drops CLIENT-ID's check-in locks for any requester; as with // ci-lock, this server lets only writers touch locks. if auth.CanPush(h.caps) { releaseCkinLocks(h.repo.DB(), c.Values[0]) } case c.Name == "send-private": if auth.CanSyncPrivate(h.caps) { h.syncPrivate = true } else { h.resp = append(h.resp, &xfer.ErrorCard{Message: privateRefusal}) } } } func (h *handler) handleDataCard(ctx context.Context, card xfer.Card) error { switch c := card.(type) { case *xfer.IGotCard: return h.handleIGot(c) case *xfer.GimmeCard: return h.handleGimme(c) case *xfer.FileCard: return h.handleFile(ctx, c.UUID, c.DeltaSrc, c.Content, nil) case *xfer.CFileCard: return h.handleFile(ctx, c.UUID, c.DeltaSrc, c.Content, c.StoredBlob) case *xfer.PrivateCard: if !auth.CanSyncPrivate(h.caps) { h.resp = append(h.resp, &xfer.ErrorCard{Message: privateRefusal}) h.nextIsPrivate = false } else { h.nextIsPrivate = true } return nil case *xfer.ReqConfigCard: return h.handleReqConfig(c) case *xfer.ConfigCard: h.handleConfigPush() return nil case *xfer.UVIGotCard: return h.handleUVIGot(c) case *xfer.UVGimmeCard: return h.handleUVGimme(c) case *xfer.UVFileCard: return h.handleUVFile(c) case *xfer.XIGotCard: return h.handleXIGot(c) case *xfer.XGimmeCard: return h.handleXGimme(c) case *xfer.XRowCard: return h.handleXRow(c) case *xfer.XDeleteCard: return h.handleXDelete(c) } return nil } func (h *handler) handleIGot(c *xfer.IGotCard) error { if c == nil { panic("handler.handleIGot: c must not be nil") } // page_xfer reads igot only with an artifact name (blob_is_hname); any // other line falls through to its bad command (§5.2). IsValidHash takes // lowercase hex only where validate16 also takes A-F: an uppercase name // would break the blob table's CHECK constraint, so it is refused here. if !hash.IsValidHash(c.UUID) { h.badCommand(replyOnlyLine(c)) return nil } if _, exists := blob.Exists(h.repo.DB(), c.UUID); exists { // Record that the client has this blob so emitIGots can skip it. // Mirrors Fossil's remote_has() → onremote table (xfer.c:1471), // which fossil fills only under push; it is kept for a pull too, so // a pull-only client is not told about what it just announced. h.recordRemoteHas(c) return nil } // Only a pushing client can deliver what it announces. Fossil's // page_xfer wraps igot handling in `if( isPush )`; asking a pull-only // client for artifacts it will never send holds its pull open until // MaxRounds (#277). if !h.pushOK { return nil } // A private igot makes a phantom only for a requester who may push // private content, as page_xfer checks g.perm.Private. if c.IsPrivate && !auth.CanSyncPrivate(h.caps) { return nil } // Fossil's igot handling goes through rid_from_uuid → content_new, which // returns 0 for a shunned name: no phantom is made, so no gimme follows // (src/content.c, #349). shunned, err := shun.IsShunned(h.repo.DB(), c.UUID) if err != nil { return fmt.Errorf("handler: shun check for igot %s: %w", c.UUID, err) } if shunned { return nil } h.pendingPhantoms = append(h.pendingPhantoms, pendingPhantom{uuid: c.UUID, private: c.IsPrivate}) // page_xfer calls remote_has on the new phantom too, so a file card // later in the request that fills it is not announced back. h.recordRemoteHas(c) return nil } // recordRemoteHas notes that the client holds what an igot announced. func (h *handler) recordRemoteHas(c *xfer.IGotCard) { if c == nil { panic("handler.recordRemoteHas: c must not be nil") } if h.remoteHas == nil { h.remoteHas = make(map[string]remoteHasEntry) } h.remoteHas[c.UUID] = remoteHasEntry{isPrivate: c.IsPrivate} } // pendingPhantom is an artifact a pushing client announced that // flushPhantoms has yet to record. type pendingPhantom struct { uuid string private bool } // flushPhantoms records the pending igot names as phantoms in one // transaction, as page_xfer's rid_from_uuid(..., 1, isPriv) does: push // creates server phantoms from client igot (§3.3, §8.5). The gimme for them // is sent by requestPhantoms after the card loop, so a file card later in // the same request fills one without being asked for. One transaction for // the run avoids a WAL commit per announced artifact. func (h *handler) flushPhantoms() error { if len(h.pendingPhantoms) == 0 { return nil } pending := h.pendingPhantoms h.pendingPhantoms = nil return h.repo.WithTx(func(tx *db.Tx) error { for _, p := range pending { rid, err := blob.StorePhantom(tx, p.uuid) if err != nil { return fmt.Errorf("handler: phantom for igot %s: %w", p.uuid, err) } if !p.private { continue } if err := content.MakePrivate(tx, int64(rid)); err != nil { return err } } return nil }) } // maxPhantomGimme bounds the gimme cards one reply sends for the server's // phantoms: page_xfer's request_phantoms(&xfer, 500). const maxPhantomGimme = 500 // requestPhantoms asks a pushing client for the server's phantoms, as // page_xfer's request_phantoms does after the card loop: every phantom but // those the client asked for this round (it lacks them), shunned names, and // private ones outside a private sync (§3.3), at most maxPhantomGimme. func (h *handler) requestPhantoms() error { if !h.pushOK { panic("handler.requestPhantoms: push not accepted") } query := "SELECT uuid FROM phantom CROSS JOIN blob USING(rid)" + " WHERE NOT EXISTS(SELECT 1 FROM shun WHERE uuid=blob.uuid)" if !h.syncPrivate { query += " AND NOT EXISTS(SELECT 1 FROM private WHERE rid=blob.rid)" } rows, err := h.repo.DB().Query(query + " ORDER BY rid") if err != nil { return fmt.Errorf("handler: listing phantoms: %w", err) } defer rows.Close() sent := 0 for sent < maxPhantomGimme && rows.Next() { var uuid string if err := rows.Scan(&uuid); err != nil { return fmt.Errorf("handler: listing phantoms: %w", err) } if h.remoteLacks[uuid] { continue } h.resp = append(h.resp, &xfer.GimmeCard{UUID: uuid}) sent++ } return rows.Err() } func (h *handler) handleGimme(c *xfer.GimmeCard) error { if c == nil { panic("handler.handleGimme: c must not be nil") } // page_xfer reads gimme only with an artifact name (blob_is_hname); any // other line falls through to its bad command (§5.2). if !hash.IsValidHash(c.UUID) { h.badCommand(replyOnlyLine(c)) return nil } // The client lacks what it asks for, so the server does not ask for it // back (fossil's remote_unk, ahead of the isPull check). if h.remoteLacks == nil { h.remoteLacks = make(map[string]bool) } h.remoteLacks[c.UUID] = true // Fossil's page_xfer answers gimme only `if( isPull )`: a pull card the // requester's Read permission accepted, or a clone, which sets isPull // too. Answering any other gimme hands content to whoever names a hash. if !h.pullOK && !h.cloneMode { return nil } // BUGGIFY: 5% chance skip sending a file to test client retry. if h.buggify != nil && h.buggify.Check("handler.handleGimme.skip", 0.05) { return nil } rid, ok := content.AvailableByUUID(h.repo.DB(), c.UUID) if !ok { return nil // blob not found — not fatal, skip. } isPriv := content.IsPrivate(h.repo.DB(), int64(rid)) if isPriv && !h.syncPrivate { // §8.6: outside a private sync, a peer dated 2020-04-13 or later is // told the artifact exists and is private; an older or undated one // is told nothing, as fossil's send_file decides on remoteDate. if h.peerDate >= privateIGotDate { h.resp = append(h.resp, &xfer.IGotCard{UUID: c.UUID, IsPrivate: true}) } return nil } // §7.6: SHA3 content is withheld from a client older than SHA3 // support, or one that named no release; the refusal is appended, so // what the reply already holds stands (§10.1). if sha3Barred(c.UUID, h.peerVersion) { h.resp = append(h.resp, sha3Refusal()) return nil } data, err := h.cache.Expand(h.repo.DB(), rid) if err != nil { h.resp = append(h.resp, &xfer.ErrorCard{ Message: fmt.Sprintf("expand %s: %v", c.UUID, err), }) return nil } if isPriv { // BUGGIFY: 5% chance skip the PrivateCard prefix — client should // treat the file as public; next sync round corrects the status. if h.buggify == nil || !h.buggify.Check("handler.handleGimme.dropPrivateCard", 0.05) { h.resp = append(h.resp, &xfer.PrivateCard{}) } } h.resp = append(h.resp, &xfer.FileCard{UUID: c.UUID, Content: data}) h.filesSent++ return nil } func (h *handler) handleFile(ctx context.Context, uuid, deltaSrc string, payload []byte, storedBlob []byte) error { if uuid == "" { panic("handler.handleFile: uuid must not be empty") } // §6.6: without an accepted push, the first file or cfile card resets // the reply, is refused and ends card processing, as page_xfer's // "error not authorized to write" and break do. The end-of-request // steps of a pull accepted earlier still run. if !h.pushOK { h.breakLoop("not authorized to write") return nil } // A shunned artifact is dropped on receipt, with no error, as fossil's // xfer_accept_file drops it (§7.7); it is not received. shunned, err := shun.IsShunned(h.repo.DB(), uuid) if err != nil { return err } if shunned { h.nextIsPrivate = false return nil } // BUGGIFY: 3% chance reject a valid file to test client re-push. if h.buggify != nil && h.buggify.Check("handler.handleFile.reject", 0.03) { h.resp = append(h.resp, &xfer.ErrorCard{ Message: fmt.Sprintf("buggify: rejected file %s", uuid), }) return nil } vis := visibility(h.nextIsPrivate) h.nextIsPrivate = false if err := storeReceivedFile(ctx, h.repo, uuid, deltaSrc, payload, storedBlob, vis, nil); err != nil { h.resp = append(h.resp, &xfer.ErrorCard{ Message: fmt.Sprintf("storing %s: %v", uuid, err), }) return nil } _, ok := blob.Exists(h.repo.DB(), uuid) if h.buggify != nil && h.buggify.Check("handler.handleFile.missingAfterStore", 0.01) { ok = false } if !ok { h.resp = append(h.resp, &xfer.ErrorCard{ Message: fmt.Sprintf("blob %s missing after store", uuid), }) return nil } h.filesRecvd++ return nil } // handleReqConfig answers reqconfig NAME with the configuration groups NAME // selects and the requester may read (config_groups.go). Anything else — a // bare name, an unknown group, a group the requester lacks the capability // for — gets no cards and no error, as fossil answers it, so the rest of the // exchange goes on. func (h *handler) handleReqConfig(c *xfer.ReqConfigCard) error { if c == nil { panic("handler.handleReqConfig: c must not be nil") } mask := configRequestMask(c.Name, h.caps, h.canRead(), h.authed) if mask == 0 { return nil } cards, err := configGroupCards(h.repo.DB(), mask) if err != nil { return fmt.Errorf("handler: reqconfig %q: %w", c.Name, err) } h.resp = append(h.resp, cards...) return nil } func (h *handler) emitIGots() error { // Emit igot for all non-phantom blobs so the client can discover // everything the server has. Cluster generation is a client-side // optimization for push; the server always advertises all blobs. rows, err := h.repo.DB().Query(` SELECT uuid FROM blob WHERE size >= 0 AND NOT EXISTS(SELECT 1 FROM shun WHERE uuid=blob.uuid) AND NOT EXISTS(SELECT 1 FROM private WHERE rid=blob.rid)`, ) if err != nil { return fmt.Errorf("handler: listing blobs: %w", err) } defer rows.Close() var uuids []string for rows.Next() { var uuid string if err := rows.Scan(&uuid); err != nil { return err } // remoteHas is populated from client igot cards in handleIGot. // Skip if the client already has this blob as public (non-private). // If the client has it as private, we still emit the public igot so // the client can clear its private status (private→public transition). if e, ok := h.remoteHas[uuid]; ok && !e.isPrivate { continue } uuids = append(uuids, uuid) } if err := rows.Err(); err != nil { return err } // BUGGIFY: 10% chance truncate igot list to test multi-round convergence. if h.buggify != nil && h.buggify.Check("handler.emitIGots.truncate", 0.10) && len(uuids) > 1 { uuids = uuids[:len(uuids)/2] } for _, uuid := range uuids { h.resp = append(h.resp, &xfer.IGotCard{UUID: uuid}) } if h.syncPrivate { if err := h.emitPrivateIGots(); err != nil { return err } } return nil } // emitPrivateIGots emits igot cards with IsPrivate=true for all blobs in // the private table. Only called when the client sent pragma send-private // and has the 'x' capability. // // Shunned private artifacts are still announced: fossil's send_private // (src/xfer.c) selects from the private table with no shun filter, unlike // the public igot paths. #349 asked for the filter, but matching fossil // wins; the receiving side drops a shunned file on arrival either way. func (h *handler) emitPrivateIGots() error { rows, err := h.repo.DB().Query( "SELECT b.uuid FROM private p JOIN blob b ON p.rid=b.rid WHERE b.size >= 0", ) if err != nil { return fmt.Errorf("handler: listing private blobs: %w", err) } defer rows.Close() var uuids []string for rows.Next() { var uuid string if err := rows.Scan(&uuid); err != nil { return err } // Skip if the client already has this blob as private. // If the client has it as public, we still emit the private igot so // the client can update its private status (public→private transition). if e, ok := h.remoteHas[uuid]; ok && e.isPrivate { continue } uuids = append(uuids, uuid) } if err := rows.Err(); err != nil { return err } // BUGGIFY: 10% chance truncate private igot list to test multi-round convergence. if h.buggify != nil && h.buggify.Check("handler.emitPrivateIGots.truncate", 0.10) && len(uuids) > 1 { uuids = uuids[:len(uuids)/2] } for _, uuid := range uuids { h.resp = append(h.resp, &xfer.IGotCard{UUID: uuid, IsPrivate: true}) } return nil } // sendAllClusters emits igot cards for all cluster artifacts that are // not still in unclustered (i.e., already fully clustered themselves). func (h *handler) sendAllClusters() error { rows, err := h.repo.DB().Query(` SELECT b.uuid FROM tagxref tx JOIN blob b ON tx.rid = b.rid WHERE tx.tagid = 7 AND NOT EXISTS (SELECT 1 FROM unclustered WHERE rid = b.rid) AND NOT EXISTS (SELECT 1 FROM phantom WHERE rid = b.rid) AND NOT EXISTS (SELECT 1 FROM shun WHERE uuid = b.uuid) AND NOT EXISTS (SELECT 1 FROM private WHERE rid = b.rid) AND b.size >= 0 `) if err != nil { return fmt.Errorf("handler: listing clusters: %w", err) } defer rows.Close() bracketed := false for rows.Next() { var uuid string if err := rows.Scan(&uuid); err != nil { return err } // Clusters are always public (query excludes private table). if e, ok := h.remoteHas[uuid]; ok && !e.isPrivate { continue } if !bracketed { h.resp = append(h.resp, &xfer.CommentCard{Text: "sending-clusters"}) bracketed = true } h.resp = append(h.resp, &xfer.IGotCard{UUID: uuid}) } if err := rows.Err(); err != nil { return err } // fossil's send_all_clusters brackets the cluster igots with comments, // and sends neither when there is no cluster to announce (§5.3). if bracketed { h.resp = append(h.resp, &xfer.CommentCard{Text: "end-of-clusters"}) } return nil } // handleConfigPush answers a pushed config card. fossil's page_xfer refuses // one from a requester without Admin with a reset, this error and a break // (§9.3). It applies an Admin's through configure_receive, which this server // does not implement, so it refuses that too, naming why, rather than drop // the configuration while the push appears to succeed. func (h *handler) handleConfigPush() { if !auth.IsAdmin(h.caps) { h.breakLoop("not authorized to push configuration") return } h.breakLoop("configuration push is not supported by this server") } func (h *handler) emitCloneBatch() error { // Capture the snapshot bound on the first round of a clone session. // Without an upper bound the batch query keeps picking up blobs that other // writers commit between rounds, the completion sentinel // CloneSeqNoCard{SeqNo:0} never fires, and the client loops to MaxRounds // (issue #17). The bound rides back to the client via CookieCard and // gets echoed on every subsequent round. if h.cloneSnapMax == 0 { var maxRid int if err := h.repo.DB().QueryRow("SELECT COALESCE(MAX(rid), 0) FROM blob WHERE size >= 0").Scan(&maxRid); err != nil { return fmt.Errorf("handler: capture clone snapshot: %w", err) } h.cloneSnapMax = maxRid h.resp = append(h.resp, &xfer.CookieCard{Value: fmt.Sprintf("%s%d", cloneCapPrefix, maxRid)}) } budget := DefaultCloneBatchBytes // BUGGIFY: 10% chance reduce the budget to one byte to stress pagination. // The budget is consulted before each artifact and never suppresses the // first one, so this yields exactly one artifact per round. if h.buggify != nil && h.buggify.Check("handler.emitCloneBatch.smallBatch", 0.10) { budget = 1 } truncate := h.buggify != nil && h.buggify.Check("clone.emitCloneBatch.truncate", 0.10) // cloneSeq is the rid of the next blob to send and is inclusive, matching // §8.2's "iterate from supplied RID through current maximum" and // canonical's `while( seqno<=max )` loop in page_xfer(). handleControlCard // is the sole writer and floors it at 1, but that is enforced far from // here with nothing compile-time behind it, so this stays a check rather // than an assumption. cursorAtEntry := h.cloneSeq if cursorAtEntry <= 0 { return fmt.Errorf("handler: clone cursor must be positive, got %d", cursorAtEntry) } rows, err := h.repo.DB().Query( "SELECT rid, uuid, size FROM blob WHERE rid >= ? AND rid <= ? AND size >= 0 ORDER BY rid", h.cloneSeq, h.cloneSnapMax, ) if err != nil { return fmt.Errorf("handler: clone batch: %w", err) } defer rows.Close() count := 0 // sent records every rid emitted this round, including delta sources a // dependent's chain pulls in ahead of their ascending position (issue // #141). It caps re-emission at once per round the way canonical fossil's // per-request "sent" bag does; across rounds the ascending cursor resumes, // and a source pulled in early in a prior round may be re-sent, which the // receiver dedupes by hash. sent := map[int]bool{} // sentTop counts top-level artifacts sent this round. Each carries a whole // delta chain, so it is distinct from count (the card total) and is the // quantity the truncate liveness guard below reasons about. sentTop := 0 // bytesSent accumulates the wire cost of the cards emitted this round, // mirroring canonical's `while( mxSend > blob_size(pOut) )`: the budget is // tested before each artifact and the artifact (with its chain) that // crosses it is still sent whole. That ordering is what guarantees forward // progress when a single artifact is larger than the entire budget — a // post-send test would emit nothing and stall the cursor. bytesSent := 0 var lastTopRID int // nextRID is the cursor reported back to the client: the rid of the first // blob this batch did not send, or 0 once the snapshot is exhausted. nextRID := 0 for rows.Next() { var rid int var uuid string var fullSize int if err := rows.Scan(&rid, &uuid, &fullSize); err != nil { return err } // A higher-rid delta source that a dependent's chain already pulled in // this round is on the wire; skip it before it can consume budget or // advance the cursor a second time. if sent[rid] { continue } isPriv := content.IsPrivate(h.repo.DB(), int64(rid)) if isPriv && !h.syncPrivate { continue // skip private blob, don't count toward batch } if bytesSent >= budget { nextRID = rid break } // §7.2/#141: emit the artifact source-first. A row stored as a delta // goes out as a delta card preceded by the source it names, reclaiming // the delta-transmission bandwidth a plain (expanded) cfile gives up. // content.Deltify deltifies the OLDER artifact against the NEWER one, // so a delta's source almost always has a *greater* rid than the delta // itself and would not yet have been sent under this loop's ascending // order; buildCloneArtifact walks the chain and emits each source ahead // of its dependent so no delta ever forward-references a card that has // not arrived. The delta rides a cfile that names its source, as in // canonical's send_compressed_file. cards, rids, cost, err := h.buildCloneArtifact(rid, uuid, fullSize, sent) if err != nil { return err } // Respect the client's decode bound: the round the client decompresses // must stay within xfer.MaxDecompressedBytes. If this artifact's chain // would push the round past that bound and the round already carries // content, flush now so the chain leads a round of its own. Without this // the budget check above only bounds the accumulated *filler*, and a // large artifact riding behind sub-budget filler carried the round to // budget-plus-artifact, past what the client could decode (issue #109). // // The first artifact of a round (count == 0) is always sent, so a chain // larger than the bound still makes forward progress -- it is simply // unclonable, a property of the artifact alone and not of what precedes // it in the round. if count > 0 && bytesSent+cost > xfer.MaxDecompressedBytes { nextRID = rid break } h.resp = append(h.resp, cards...) for _, r := range rids { sent[r] = true } h.filesSent += len(rids) lastTopRID = rid count += len(rids) sentTop++ bytesSent += cost } if err := rows.Err(); err != nil { return err } // BUGGIFY: truncate — remove the last cfile card to simulate an incomplete // batch. `sentTop > 1` is a liveness constraint, not just a chaos-injection // guard. The removed card is the deepest dependent of the last artifact's // chain (emitted last in source-first order, so nothing else on the wire // depends on it), and nextRID is set to that artifact's top-level rid. With // more than one top-level artifact sent this round that rid strictly // exceeds the cursor we entered with, so the client re-requests from it // rather than looping on the same batch until MaxRounds. if truncate && sentTop > 1 { for i := len(h.resp) - 1; i >= 0; i-- { if _, ok := h.resp[i].(*xfer.CFileCard); ok { h.resp = append(h.resp[:i], h.resp[i+1:]...) h.filesSent-- // The dropped card held lastTopRID, so that rid becomes the // next one owed to the client rather than being skipped. nextRID = lastTopRID break } } } // The cursor must strictly advance or signal exhaustion. Catching a stall // here names the round that caused it; letting it through only surfaces // later as an opaque MaxRounds failure on the client. if nextRID != 0 && nextRID <= cursorAtEntry { return fmt.Errorf( "handler: clone cursor did not advance: next %d, cursor %d", nextRID, cursorAtEntry) } // SeqNo 0 signals completion so the client stops requesting. h.resp = append(h.resp, &xfer.CloneSeqNoCard{SeqNo: nextRID}) return nil } // maxCloneChainWalk bounds the delta-chain walk in walkCloneChain. A real // chain is only as deep as an artifact's revision count and always terminates // at a full-content tip; this cap exists solely so a corrupt or adversarial // delta graph cannot spin the walk forever. content.Deltify forbids cycles, // but the send path does not trust its store to be uncorrupted here. On // reaching the cap the deepest artifact is emitted as full content, which is // always safe (it references no source card). const maxCloneChainWalk = 4096 // cloneArtifact is one artifact the clone will emit. An empty srcUUID means // emit full content as a compressed cfile card; a set srcUUID means emit a // delta against that source, which buildCloneArtifact guarantees precedes this // artifact on the wire. type cloneArtifact struct { rid int uuid string usize int priv bool srcUUID string expanded bool // stored as a delta but sent as full content } // cloneBlobRow holds the columns walkCloneChain needs to decide whether a // delta source can be forwarded to the receiver this clone. type cloneBlobRow struct { uuid string size int priv bool } // loadCloneBlob reads the blob row for rid. ok is false when no such row // exists (a source named by a dangling delta link, which is treated as // unforwardable rather than an error). func (h *handler) loadCloneBlob(rid int) (cloneBlobRow, bool, error) { var row cloneBlobRow err := h.repo.DB().QueryRow("SELECT uuid, size FROM blob WHERE rid = ?", rid). Scan(&row.uuid, &row.size) if errors.Is(err, sql.ErrNoRows) { return cloneBlobRow{}, false, nil } if err != nil { return cloneBlobRow{}, false, fmt.Errorf("handler: clone blob info rid %d: %w", rid, err) } row.priv = content.IsPrivate(h.repo.DB(), int64(rid)) return row, true, nil } // storedContent returns rid's content as the blob table stores it: the // on-disk blob format, compressed, of the full artifact or of its delta. func (h *handler) storedContent(rid int) ([]byte, error) { var stored []byte err := h.repo.DB().QueryRow("SELECT content FROM blob WHERE rid = ?", rid).Scan(&stored) if err != nil { return nil, fmt.Errorf("handler: stored content of rid %d: %w", rid, err) } return stored, nil } // walkCloneChain follows rid's delta chain toward its full-content tip and // returns the artifacts to emit, ordered DEPENDENT-FIRST (rid itself first). // The walk stops at the first source that is already available to the receiver // (emitted earlier this round) or that cannot be forwarded this clone — // outside the snapshot bound, a phantom, or a private source under a public // clone — and tags the artifact below such a source for full-content emission // so it never depends on a card that will not arrive. func (h *handler) walkCloneChain(rid int, uuid string, usize int, sent map[int]bool) ([]cloneArtifact, error) { chain := make([]cloneArtifact, 0, 8) curRID, curUUID, curUSize := rid, uuid, usize curPriv := content.IsPrivate(h.repo.DB(), int64(curRID)) for step := 0; step < maxCloneChainWalk; step++ { src, err := content.DeltaSource(h.repo.DB(), libfossil.FslID(curRID)) if err != nil { return nil, fmt.Errorf("handler: delta source for rid %d: %w", curRID, err) } m := cloneArtifact{rid: curRID, uuid: curUUID, usize: curUSize, priv: curPriv} if src == 0 { return append(chain, m), nil // full-content tip } info, ok, err := h.loadCloneBlob(int(src)) if err != nil { return nil, err } forwardable := ok && info.size >= 0 && int(src) <= h.cloneSnapMax && (!info.priv || h.syncPrivate) if !forwardable { m.expanded = true return append(chain, m), nil // source unsendable: emit curRID full } m.srcUUID = info.uuid chain = append(chain, m) if sent[int(src)] { return chain, nil // source already on the wire this round } curRID, curUUID, curUSize, curPriv = int(src), info.uuid, info.size, info.priv } // Walk cap hit: force the deepest artifact to full content so its delta // does not reference a source we never emitted. chain[len(chain)-1].srcUUID = "" chain[len(chain)-1].expanded = true return chain, nil } // buildCloneArtifact assembles the wire cards that deliver artifact rid to the // receiver, ordered source-first: the chain tip (or the lowest full-content // anchor) leads, each delta follows the source it names, and rid itself is // last. It returns the cards, the artifact rids they cover (for the caller's // sent bookkeeping), and the accumulated wire cost. It does not consult the // byte budget — the caller decides whether the whole sequence fits the round. func (h *handler) buildCloneArtifact( rid int, uuid string, usize int, sent map[int]bool, ) ([]xfer.Card, []int, int, error) { chain, err := h.walkCloneChain(rid, uuid, usize, sent) if err != nil { return nil, nil, 0, err } if len(chain) == 0 { panic("handler.buildCloneArtifact: walkCloneChain returned no artifacts") } cards := make([]xfer.Card, 0, 2*len(chain)) rids := make([]int, 0, len(chain)) cost := 0 // chain is dependent-first; emit it reversed so every source precedes the // delta that names it. for i := len(chain) - 1; i >= 0; i-- { m := chain[i] rids = append(rids, m.rid) // §7.6, Appendix A.2: the refusal takes a SHA3 artifact's place. if sha3Barred(m.uuid, h.peerVersion) { cards = append(cards, sha3Refusal()) continue } var data []byte if m.srcUUID == "" { data, err = content.Expand(h.repo.DB(), libfossil.FslID(m.rid)) if err != nil { return nil, nil, 0, fmt.Errorf("handler: expanding rid %d: %w", m.rid, err) } } else { data, err = blob.Load(h.repo.DB(), libfossil.FslID(m.rid)) if err != nil { return nil, nil, 0, fmt.Errorf("handler: loading delta rid %d: %w", m.rid, err) } } if m.priv { cards = append(cards, &xfer.PrivateCard{}) } // Every clone artifact rides a cfile, a delta naming its source, as // canonical's send_compressed_file does for clone version 3 and as the // draft requires (§7.1, §8.2). USIZE is the full artifact's size even // for a delta, as fossil sends blob.size. // An artifact sent as it is stored goes out as its stored bytes, // as send_compressed_file sends the blob table's content (§7.2). var stored []byte if !m.expanded { if stored, err = h.storedContent(m.rid); err != nil { return nil, nil, 0, err } } cards = append(cards, &xfer.CFileCard{ UUID: m.uuid, DeltaSrc: m.srcUUID, USize: m.usize, Content: data, StoredBlob: stored, }) cost += cloneCardOverheadBytes + len(data) } return cards, rids, cost, nil }