sync

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 29, 2026 License: MIT Imports: 31 Imported by: 0

Documentation

Overview

Package sync implements Fossil's multi-round sync and clone protocol.

The client side drives convergence: Sync opens a session against a Transport, exchanges xfer cards in rounds until both sides agree on the set of blobs, then returns a SyncResult. Clone creates a new repo file and populates it from the remote.

The server side is stateless per round: HandleSync processes one request and produces one response. Use XferHandler to embed sync in an http.ServeMux alongside operational routes like /healthz.

Observability is injected via the Observer interface — pass nil for zero-cost no-ops, or supply an implementation (e.g. OTelObserver from leaf/telemetry) for traces, metrics, and structured logs.

Index

Constants

View Source
const (
	// MaxExchangeAttempts is how many times one round is sent before the
	// clone or sync gives up, the first attempt included.
	MaxExchangeAttempts = 4
	// ExchangeRetryBaseDelay is the wait before the second attempt. It
	// doubles before each further attempt, capped at ExchangeRetryMaxDelay.
	ExchangeRetryBaseDelay = 250 * time.Millisecond
	// ExchangeRetryMaxDelay caps the backoff between attempts.
	ExchangeRetryMaxDelay = 2 * time.Second
)

Bounds on retrying a single exchange round. A clone of a large repository is a conversation of many rounds spanning minutes, so one transient drop anywhere in it used to discard the whole transfer (#185). Retrying is bounded so a peer that is genuinely gone still fails promptly: the delays below sum to under two seconds before the last attempt.

View Source
const (
	// DefaultMaxSend is the default byte budget per round for file payloads.
	DefaultMaxSend = 250000
	// MaxRounds caps the number of sync rounds before giving up.
	MaxRounds = 100
	// MaxGimmeBase is the minimum gimme cap per round.
	MaxGimmeBase = 200
)
View Source
const DefaultCkinLockTimeout = 60 * time.Second

DefaultCkinLockTimeout is the default duration after which a ci-lock expires.

View Source
const DefaultCloneBatchBytes = 16_000_000

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.

Variables

This section is empty.

Functions

func HandleSync

func HandleSync(ctx context.Context, r *repo.Repo, req *xfer.Message) (*xfer.Message, error)

HandleSync processes an incoming xfer request and produces a response. Stateless per-round — the client drives convergence.

func HandleSyncWithOpts

func HandleSyncWithOpts(ctx context.Context, r *repo.Repo, req *xfer.Message, opts HandleOpts) (*xfer.Message, error)

HandleSyncWithOpts processes an incoming xfer request with optional fault injection. Used by DST harness; production callers use HandleSync.

func RetryableFault added in v0.8.0

func RetryableFault(err error) error

RetryableFault tags an error raised while bytes were still crossing the wire, so the round loop may resend the same request. Transports outside this package use it for their read stage, where a short read means the reply was cut off rather than malformed; a failure raised while interpreting a reply that arrived whole must never be tagged.

func ServeHTTP

func ServeHTTP(ctx context.Context, addr string, r *repo.Repo, h HandleFunc) error

ServeHTTP starts an HTTP server that accepts Fossil xfer requests. Blocks until ctx is cancelled. Stock fossil clone/sync can connect.

func XferHandler

func XferHandler(r *repo.Repo, h HandleFunc) http.HandlerFunc

XferHandler returns an http.HandlerFunc that decodes Fossil xfer requests, dispatches to the HandleFunc, and encodes the response. Use this to compose a custom mux when you need additional routes (e.g. /healthz) alongside xfer.

Types

type BuggifyChecker

type BuggifyChecker interface {
	Check(site string, probability float64) bool
}

BuggifyChecker is an optional fault injection interface. Pass nil in production — implementations should be nil-safe.

type CkinLockFail

type CkinLockFail struct {
	HeldBy string
	Since  time.Time
}

CkinLockFail reports that another client holds the check-in lock.

type CkinLockReq

type CkinLockReq struct {
	ParentUUID string
	ClientID   string
}

CkinLockReq requests a server-side check-in lock.

type CloneOpts

type CloneOpts struct {
	User     string // Credentials for clone auth (also default admin user)
	Password string
	Version  int            // Protocol version (default 3)
	Env      *simio.Env     // nil defaults to RealEnv
	Observer Observer       // nil defaults to no-op
	Buggify  BuggifyChecker // fault injection for DST (nil = no faults)
}

CloneOpts configures a clone operation.

type CloneResult

type CloneResult struct {
	Rounds          int
	BlobsRecvd      int
	ArtifactsLinked int // Manifests crosslinked into event table
	ProjectCode     string
	ServerCode      string
	Messages        []string // Informational messages from server
}

CloneResult reports what happened during a clone.

func Clone

func Clone(ctx context.Context, path string, t Transport, opts CloneOpts) (r *repo.Repo, result *CloneResult, err error)

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.

type HTTPTransport

type HTTPTransport struct {
	URL string // repo root, e.g. "http://localhost:8080"
}

HTTPTransport speaks Fossil's HTTP /xfer protocol. Fossil routes to /xfer based on Content-Type: application/x-fossil, NOT the URL path. URL should be the repo root (e.g. "http://localhost:8080").

func (*HTTPTransport) Exchange

func (t *HTTPTransport) Exchange(ctx context.Context, req *xfer.Message) (*xfer.Message, error)

type HandleEnd

type HandleEnd struct {
	CardsProcessed int
	FilesSent      int
	FilesReceived  int
	Err            error
}

HandleEnd describes the result of a server-side sync request.

type HandleFunc

type HandleFunc func(ctx context.Context, r *repo.Repo, req *xfer.Message) (*xfer.Message, error)

HandleFunc is the server-side sync handler signature. Transport listeners call this with decoded requests and write back the response.

type HandleOpts

type HandleOpts struct {
	Buggify      BuggifyChecker // nil in production.
	Observer     Observer       // nil defaults to no-op.
	ContentCache *content.Cache // nil = no caching.
}

HandleOpts configures optional behavior for HandleSync.

type HandleStart

type HandleStart struct {
	Operation   string // "sync" or "clone"
	ProjectCode string
	RemoteAddr  string
}

HandleStart describes the beginning of a server-side sync request.

type MockTransport

type MockTransport struct {
	Handler func(req *xfer.Message) *xfer.Message
}

MockTransport replays canned responses for testing.

func (*MockTransport) Exchange

func (t *MockTransport) Exchange(ctx context.Context, req *xfer.Message) (*xfer.Message, error)

type Observer

type Observer interface {
	// Client-side session lifecycle.
	Started(ctx context.Context, info SessionStart) context.Context
	RoundStarted(ctx context.Context, round int) context.Context
	RoundCompleted(ctx context.Context, round int, stats RoundStats)
	Completed(ctx context.Context, info SessionEnd, err error)

	// Per-error recording — called on individual protocol errors.
	Error(ctx context.Context, err error)

	// Server-side request lifecycle.
	HandleStarted(ctx context.Context, info HandleStart) context.Context
	HandleCompleted(ctx context.Context, info HandleEnd)

	// Table sync lifecycle.
	TableSyncStarted(ctx context.Context, info TableSyncStart)
	TableSyncCompleted(ctx context.Context, info TableSyncEnd)
}

Observer receives lifecycle callbacks during sync and clone operations. A single Observer instance may be shared across multiple concurrent sessions. Pass nil for no-op default.

Performance: nopObserver implements all methods as empty functions. The only cost on the hot path is one indirect call per invocation (~2ns).

type PhantomStallError

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
}

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.

func (*PhantomStallError) Error

func (e *PhantomStallError) Error() string

type RoundStats

type RoundStats struct {
	FilesSent     int
	FilesReceived int
	GimmesSent    int
	IgotsSent     int
	BytesSent     int64
	BytesReceived int64
}

RoundStats reports per-round activity for observer callbacks. Value type — no allocations on the nop path.

type SessionEnd

type SessionEnd struct {
	Operation                 string
	Rounds                    int
	FilesSent, FilesRecvd     int
	UVFilesSent, UVFilesRecvd int
	UVGimmesSent              int
	BytesSent, BytesRecvd     int64
	ProjectCode               string
	Errors                    []string
}

SessionEnd describes the result of a sync or clone operation.

type SessionStart

type SessionStart struct {
	Operation   string // "sync" or "clone"
	Push, Pull  bool
	UV          bool
	ProjectCode string
	PeerID      string // identifies the leaf agent instance
}

SessionStart describes the beginning of a sync or clone operation.

type StatusError added in v0.8.0

type StatusError struct {
	Code   int    // HTTP status code
	Status string // HTTP status line, e.g. "503 Service Unavailable"
}

StatusError reports a non-2xx HTTP reply from a sync peer. It is raised instead of decoding the body, because an error page is not an xfer reply and reporting it as a decode failure hides what actually happened.

func (*StatusError) Error added in v0.8.0

func (e *StatusError) Error() string

func (*StatusError) RetryableTransportFault added in v0.8.0

func (e *StatusError) RetryableTransportFault() bool

RetryableTransportFault treats 5xx, 429 and 408 as worth another attempt: each says the peer could not serve this request now, not that the request was wrong. Every other status — 4xx above all — is the peer rejecting the request itself, which repeating cannot fix and would only obscure.

type SyncOpts

type SyncOpts struct {
	Push, Pull   bool           // enable push/pull directions (at least one must be true)
	ProjectCode  string         // repo's project-code — sent to identify the repository
	ServerCode   string         // server-code from a previous session (cookie-like, speeds up sync)
	User         string         // login user — empty means unauthenticated "nobody" sync
	Password     string         // login password
	PeerID       string         // identifies this leaf agent instance (for observability)
	MaxSend      int            // byte budget per round for file payloads (0 defaults to DefaultMaxSend)
	UV           bool           // enable unversioned file sync (wiki, forum, attachments)
	XTableSync   bool           // enable extension table sync (peer_registry, etc.) — only between EdgeSync peers
	Private      bool           // enable private artifact sync
	Env          *simio.Env     // nil defaults to RealEnv
	Buggify      BuggifyChecker // nil in production — used by DST for fault injection
	Observer     Observer       // nil defaults to no-op
	CkinLock     *CkinLockReq   // nil = no lock requested
	ContentCache *content.Cache // nil = no caching (every Expand walks the full delta chain)
}

SyncOpts configures a sync session.

type SyncResult

type SyncResult struct {
	Rounds, FilesSent, FilesRecvd int
	UVFilesSent, UVFilesRecvd     int
	UVGimmesSent                  int
	BytesSent, BytesRecvd         int64
	ArtifactsLinked               int
	Errors                        []string
	CkinLockFail                  *CkinLockFail // nil = no conflict
}

SyncResult reports what happened during a sync.

func Sync

func Sync(ctx context.Context, r *repo.Repo, t Transport, opts SyncOpts) (result *SyncResult, err error)

Sync runs the client sync loop against the given transport. It returns once the protocol has converged or a fatal error occurs.

type SyncedTable

type SyncedTable struct {
	Info repo.TableInfo
	Def  repo.TableDef
}

SyncedTable caches a table definition along with its metadata.

type TableSyncEnd

type TableSyncEnd struct {
	Table    string
	Sent     int
	Received int
}

TableSyncEnd describes the result of a table sync operation.

type TableSyncStart

type TableSyncStart struct {
	Table     string
	LocalRows int
}

TableSyncStart describes the beginning of a table sync operation.

type Transport

type Transport interface {
	Exchange(ctx context.Context, request *xfer.Message) (*xfer.Message, error)
}

Transport sends an xfer request and returns the response. Implementations handle encoding (the request Message is already decoded), network I/O, and decoding. The xfer.Message uses zlib-compressed payloads internally — see xfer.Encode and xfer.Decode for wire format details.

Built-in implementations: HTTPTransport (Fossil HTTP /xfer protocol), MockTransport (canned responses for testing). The leaf agent adds NATSTransport for NATS-based peer-to-peer sync.

type TransportFault added in v0.8.0

type TransportFault interface {
	error
	// RetryableTransportFault reports whether sending the same request
	// again may succeed. It is true only when no reply was interpreted, so
	// a resend cannot mask a defect in one that was.
	RetryableTransportFault() bool
}

TransportFault is implemented by an error that classifies itself for the retry decision. A Transport knows which stage of the exchange failed — whether bytes were still crossing the wire or a reply had already arrived and was being interpreted — and that distinction cannot be recovered reliably from the error's shape afterwards, since a truncated zlib body and a truncated wire read both surface as io.ErrUnexpectedEOF.

It is optional in the same way [FramedTransport] is: a Transport that does not implement it has its errors classified by inspection instead, which is deliberately conservative — anything not recognisably a network fault is treated as fatal.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL