edgesync

package
v0.0.0-...-6de7a98 Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2026 License: AGPL-3.0 Imports: 25 Imported by: 0

Documentation

Overview

Package edgesync implements edge-to-cloud replication: a spoke (an Arc instance at the edge) ships its immutable Parquet files to a hub (a central Arc).

The unit of sync is the FILE, not the row. Arc already produces immutable content-addressed Parquet with a SHA256 in the manifest, so shipping files gives end-to-end integrity for free, costs the hub no re-ingestion, and makes idempotency trivial: (path, sha256) IS the file's identity.

The package is layered so each piece can be tested on its own:

  • ledger.go — the durable record of what has been sent to which hub, and how far. Deliberately dumb: it tracks state and knows nothing about transports, HTTP, or hubs beyond their ID.
  • transport.go — the SyncTransport interface the agent talks to, so the wire format (HTTPS, S3 relay, sneakernet bundle) stays swappable.
  • transport_memory.go — an in-process transport for tests.

See docs/progress/2026-06-04-edge-sync-architecture-converged.md.

Index

Constants

View Source 
const (
	DefaultBundleMaxFiles = 10000
	DefaultBundleMaxBytes = int64(64) << 30 // 64 GiB, roughly a large drive
)

Bundle export defaults, applied when a config value is zero.

View Source 
const BundleVersion = 1

BundleVersion is the on-disk format version.

Written into every manifest and checked on import. A reader that does not recognize a version must refuse rather than guess: a bundle is verified before anything is committed, and "verified" means nothing if the verifier misunderstood the layout.

View Source 
const DefaultHubID = "default"

DefaultHubID is the hub identifier used when multi-hub is not configured.

The ledger is keyed (hub_id, path) from day one even though phase 1 syncs to a single hub. Turning on multi-hub later is then configuration, not a migration of a live edge database — which on a disconnected box in the field is the difference between a config push and a site visit.

View Source 
const DefaultMaxAttempts = 5

DefaultMaxAttempts is how many times a file is retried before the ledger gives up on it.

Deliberately generous: on an intermittent link most failures are the link, not the file, and a spoke that abandons data after two bad contact windows is worse than one that keeps trying. A genuinely broken file (a checksum mismatch that reproduces) still stops after this many.

View Source 
const ImportBatchSize = 1000

ImportBatchSize bounds one manifest proposal.

The Cluster Operations Checklist caps Raft batches at 1000 ops so a single log entry cannot grow unbounded. It matters more here than on the online path: an HTTP transfer naturally rate-limits proposals to one per request, whereas a bundle import is a tight loop, and a 10,000-file bundle would otherwise emit 10,000 individual proposals as fast as the disk allows.

View Source 
const MaxAllowedConcurrent = 64

MaxAllowedConcurrent caps simultaneous transfers regardless of configuration.

Each transfer holds an open file handle and an io.Pipe, and a spoke is by definition a small machine. An operator typo in max_concurrent should not be able to exhaust its file descriptors.

View Source 
const MaxReconcileEntriesDefault = 10_000

MaxReconcileEntriesDefault bounds one reconcile batch.

§5.1 wants the whole pending set in a single round-trip, and notes 100k entries ≈ 20MB. Arc cannot honor that literally: the Fiber app runs with StreamRequestBody=false (api/server.go) so fasthttp buffers the entire request body before routing — and therefore before authentication. An unbounded reconcile would let anyone able to reach the port make the hub hold tens of megabytes per connection.

Capping and letting the spoke page keeps the property that actually matters: discovery costs O(batches), not O(files), so 5,000 pending files is one request rather than 5,000. At ~200 bytes per entry this is ~2MB.

View Source 
const MetaCompactionDeferEpoch = "compaction_defer_epoch"

MetaCompactionDeferEpoch is the sync_meta key recording when defer-compaction-until-synced first became active on this spoke.

View Source 
const NoteCompactedOutput = "compaction output; contents already delivered"

NoteCompactedOutput marks a skipped ledger row as a compacted output whose contents were already delivered (issue #610): every input to the job was in state synced, so the output must never sync — it would duplicate rows on the hub. Rows carrying this note are EXEMPT from PruneSkipped while their file exists (a pruned row would let discovery rediscover and sync the output) and are instead reclaimed by SweepSkippedRows once the file is gone. They are also ELIGIBLE compaction inputs (daily consumes hourly outputs).

View Source 
const NoteOperatorDismissed = "dismissed by an operator"

NoteOperatorDismissed marks a skipped row an operator deliberately dismissed via POST /api/v1/spoke-sync/ledger/dismiss. Distinct from the vanished-file and compacted-output notes: dismissed rows are the only skipped class Requeue may resurrect, and they prune normally.

View Source 
const SecretBytes = 32

SecretBytes is the length of a generated spoke secret before hex encoding.

32 bytes matches the HMAC-SHA256 block security level; anything shorter would be the weakest link in a scheme whose whole point is authenticating a remote party that can write files.

View Source 
const StagingPrefix = ".sync-staging"

StagingPrefix is where partially-received and unverified files live.

It sits outside every database's namespace so a staged file can never be mistaken for queryable data: Arc's storage layout is {database}/{measurement}/{y}/{m}/{d}/{h}/file.parquet, and a leading dot cannot be a database name (validateSyncPath rejects it).

Variables

View Source 
var (
	// ErrSpokeNotFound is returned for an unregistered or deleted spoke.
	ErrSpokeNotFound = errors.New("edgesync: spoke not registered")

	// ErrSpokeExists is returned when registering an ID that is already taken.
	//
	// Deliberately not an upsert: silently replacing a spoke's secret would
	// lock out a live edge box with no signal, and re-registering an existing
	// ID is far more likely to be a mistake than an intent.
	ErrSpokeExists = errors.New("edgesync: spoke already registered")

	// ErrSpokeDisabled is returned when a registered spoke has been disabled.
	ErrSpokeDisabled = errors.New("edgesync: spoke is disabled")
)
View Source 
var ErrAckInvalid = errors.New("edgesync: acknowledgment is invalid")

ErrAckInvalid marks an acknowledgment that must not be trusted.

View Source 
var ErrBundleAlreadyImported = errors.New("edgesync: bundle already imported")

ErrBundleAlreadyImported means this bundle's contents are already on the hub.

View Source 
var ErrBundleInvalid = errors.New("edgesync: bundle is invalid")

ErrBundleInvalid marks a bundle that must not be imported.

One error for every rejection reason so a caller maps it to a single status rather than leaking which specific check failed — a tampered bundle and a truncated one are both simply "do not import this".

View Source 
var ErrDestinationRefused = errors.New("edgesync: bundle destination refused")

ErrDestinationRefused marks a bundle path an operator may not use.

View Source 
var ErrImportInProgress = errors.New("edgesync: a bundle import is already in progress")

ErrImportInProgress means another bundle import is already running.

One import at a time is the physical reality (one drive in one slot) and a hard requirement (the staging area and result collector are shared), so a concurrent request fails FAST with this sentinel rather than queueing silently behind a run that can legally take hours — the operator whose first request timed out client-side would otherwise re-POST and hang too (2026-08-19 audit M3).

View Source 
var ErrInvalidTransition = errors.New("edgesync: invalid state transition")

ErrInvalidTransition is returned when a state change is attempted from a state that does not permit it — e.g. marking a terminally failed entry synced, or re-sending an entry the hub has already acknowledged.

The ledger enforces these rather than trusting callers because §6.2's exactly-once-effect property depends on `synced` being reached only via a hub acknowledgment and never being silently walked back.

View Source 
var ErrNoAck = errors.New("edgesync: bundle carries no acknowledgment")

ErrNoAck means the bundle directory carries no acknowledgment.

Distinct from an invalid one: a drive that has not yet been to the hub is the normal case on the outbound leg, not a problem.

View Source 
var ErrNotFound = errors.New("edgesync: ledger entry not found")

ErrNotFound is returned when a ledger lookup matches no row.

View Source 
var ErrNothingToExport = errors.New("edgesync: nothing to export")

ErrNothingToExport means no file is eligible for a bundle.

Distinct from an error: a spoke with nothing new is the steady state once a backlog has drained, and an operator running a scheduled export should not see a failure for it.

View Source 
var ErrReceiveInternal = errors.New("edgesync: receive failed for a hub-side reason")

ErrReceiveInternal wraps a failure that is the HUB's fault rather than the spoke's — storage I/O, a manifest write during a Raft election, a promote that could not complete.

The distinction is not cosmetic: a spoke told "bad request" has no reason to retry, while a transient hub-side failure is exactly what it should retry. Handlers map this to 503 rather than 400.

View Source 
var ErrReconcileTooLarge = errors.New("edgesync: reconcile batch exceeds the configured maximum")

ErrReconcileTooLarge is returned when a batch exceeds the configured cap. The spoke's remedy is to split the batch, not to retry it unchanged.

View Source 
var ErrTransportClosed = errors.New("edgesync: transport closed")

ErrTransportClosed is returned by a transport that has been shut down.

Functions

func GenerateSecret 

func GenerateSecret() (string, error)

GenerateSecret returns a cryptographically random hex-encoded secret.

func NamespacedPath 

func NamespacedPath(spokeID, sourcePath string) string

NamespacedPath rewrites a spoke's path into the hub's namespace.

This is a HUB-SIDE rewrite by design: the spoke sends its own native paths and stays unaware of namespacing, so the same spoke can sync to several hubs unmodified. Because spoke_id is bound into the request HMAC, a spoke cannot claim another's namespace.

func NewBundleID 

func NewBundleID(now time.Time) (string, error)

NewBundleID returns a lexicographically-sortable, time-prefixed identifier.

ULID-shaped: 48-bit millisecond timestamp then 80 bits of randomness, in Crockford base32. Sortable means a directory listing of bundles is in creation order, which is what an operator holding several drives wants.

func NewCompactedOutputObserver 

func NewCompactedOutputObserver(ledger *Ledger, hubID string, epoch time.Time, logger zerolog.Logger) func(storageKey string)

NewCompactedOutputObserver returns the compacted-output recorder for compaction.Manager.SetOnCompactedOutput. Errors are logged, not returned — the compaction that produced the output has already succeeded, and a lost insert is recovered by discovery's epoch rule on its next pass.

The epoch check exists for the manifest-RECOVERY path: recovery can keep an output from a crash that happened BEFORE the gate was active (an upgrade with an orphaned manifest, or an ungated period), whose inputs were not necessarily delivered. Marking that never-sync would be silent loss — so a pre-epoch output is left untracked and discovery's legacy rule syncs it once instead. Outputs of gated runs are post-epoch by construction.

func NewCompactionEligibility 

func NewCompactionEligibility(ledger *Ledger, hubID string, epoch time.Time, logger zerolog.Logger) func(ctx context.Context, paths []string) (map[string]bool, error)

NewCompactionEligibility returns the sync-eligibility gate for compaction.Manager.SetSyncEligibility.

Eligible: state synced (delivered; acked, on the air-gap path), a compacted-output row (daily consumes hourly outputs), or a tier-suffixed file ABSENT from the ledger whose embedded timestamp is after epoch (a crash orphan: produced under this gate, so its inputs were delivered by construction). Everything else — pending, in_flight, exported (on a drive, unconfirmed), failed, legacy compacted files, and unknown raws — defers. The deferral detail is logged here, broken out by state, so a wedged spoke is diagnosable from its compaction log.

func ValidateBundleID 

func ValidateBundleID(id string) error

ValidateBundleID checks the format of an identifier that arrives from a file.

The ID is attacker-chosen — a compromised spoke signs any manifest it likes — and it reaches a SQLite primary key, a log line, and (via the directory name) a filesystem path. An unbounded arbitrary string in all three is a bad idea regardless of whether a specific exploit is obvious.

func WriteAck 

func WriteAck(dir, secret string, a *Ack) error

WriteAck signs an acknowledgment and writes it into the bundle directory.

Written by the hub after a successful import. A failure here is not fatal to the import — the files are committed either way — but it does cost the spoke its chance to advance, so the caller should say so plainly.

Types

type Ack 

type Ack struct {
	Version int `json:"version"`

	BundleID string `json:"bundle_id"`
	SpokeID  string `json:"spoke_id"`

	// HubID is the hub that imported the bundle. The spoke checks it against
	// its own configured hub: an ack from somewhere else names files this
	// spoke never sent there.
	HubID string `json:"hub_id"`

	// ImportedAt is when the hub committed the bundle. Bound into the MAC and
	// surfaced to operators, not enforced as a freshness window — an ack rides
	// the same drive back and is subject to the same weeks-long latency.
	ImportedAt int64 `json:"imported_at"`

	// Paths are the spoke-relative paths the hub now holds. Committed and
	// already-present files both appear: from the spoke's point of view they
	// are the same fact — the hub has this file — and only that fact licenses
	// advancing the ledger.
	//
	// Conflicted paths are deliberately ABSENT. A conflict means the hub holds
	// DIFFERENT content at that path, so the spoke's copy has not been
	// delivered and must not be marked synced.
	Paths []string `json:"paths"`

	// Conflicts are reported so the spoke can surface them to an operator
	// without a network round-trip. Not acknowledged, only described.
	Conflicts []Conflict `json:"conflicts,omitempty"`

	// PathsDigest is the canonical digest the MAC binds.
	PathsDigest string `json:"paths_digest"`

	MAC string `json:"mac"`
}

Ack is a hub's signed statement that it holds a bundle's files.

The return leg of the air-gap transport. Without it a spoke has no terminal state: `synced` is unreachable, so PruneSynced never prunes and the ledger grows without bound on the box least able to receive a site visit.

func ReadAck 

func ReadAck(dir, secret, expectSpokeID, expectHubID string) (*Ack, error)

ReadAck reads and verifies an acknowledgment from a bundle directory.

Verifies before returning: a caller that gets an Ack back can advance its ledger on the strength of it, so an unverified one must never escape.

type AckResult 

type AckResult struct {
	BundleID   string
	HubID      string
	ImportedAt time.Time

	// Synced is how many entries advanced from exported to synced.
	Synced int

	// AlreadySynced is how many acknowledged paths were already synced. The
	// benign replay case: a drive plugged in twice.
	AlreadySynced int

	// Untracked is how many acknowledged paths this ledger does not know. A
	// spoke restored from a backup legitimately produces these.
	Untracked int

	// Discrepancies is how many acknowledged paths this ledger holds in a
	// state the ack cannot advance — in practice, terminally failed. The hub
	// says it HOLDS a file this spoke gave up on, which an operator should
	// see. Counted apart from the two benign cases because collapsing all
	// three hides the only one that means something is wrong.
	Discrepancies int

	Conflicts []Conflict
}

AckResult summarizes applying one acknowledgment.

func ApplyAck 

func ApplyAck(ctx context.Context, l *Ledger, a *Ack, logger zerolog.Logger) (*AckResult, error)

ApplyAck advances a ledger using a verified acknowledgment.

The caller must have obtained the Ack from ReadAck, which verifies it. This function trusts what it is given, so handing it an unverified ack would let a tampered path list mark files synced that no hub ever received.

An acknowledged entry that is still `pending` — one the spoke never put on the drive, queued for the network path instead — IS advanced, deliberately. ReadAck has proven the hub holds that exact path, so `synced` is factually true however it got there, and re-sending it over a contact window would spend link budget on a file already delivered. The effect is that a drive can satisfy work queued for the network, which is the right outcome when both transports are enabled.

type Agent 

type Agent struct {
	// contains filtered or unexported fields
}

Agent runs one sync pass: discover local files, ask the hub what it is missing, and stream those files to it.

This is the manual half of the design. §8.2 describes a connectivity-adaptive background loop, but that is the Enterprise feature (§9) — here the pass is triggered by an operator and runs once. The internals are the same either way, so phase 2 adds a ticker and a license gate rather than a rewrite.

func NewAgent 

func NewAgent(cfg AgentConfig) (*Agent, error)

NewAgent validates configuration and returns a ready Agent.

func (*Agent) Discover 

func (a *Agent) Discover(ctx context.Context) (int, error)

Discover walks local storage and adds files the ledger does not know about.

§8.2 says to walk the manifest, which does not exist on a standalone spoke — and even in cluster mode it is a different subsystem's index. Walking storage works identically everywhere, and compaction output and backfilled files just appear as new rows on the next pass.

func (*Agent) DismissFailed 

func (a *Agent) DismissFailed(ctx context.Context, path string) (int64, error)

DismissFailed moves failed rows to operator-dismissed skipped.

func (*Agent) EntriesByState 

func (a *Agent) EntriesByState(ctx context.Context, state SyncState, limit int) ([]*LedgerEntry, error)

EntriesByState lists ledger rows in one explicit state.

func (*Agent) RequeueFailed 

func (a *Agent) RequeueFailed(ctx context.Context, path string) (int64, error)

RequeueFailed returns failed (and operator-dismissed) rows to pending.

func (*Agent) Run 

func (a *Agent) Run(ctx context.Context) (*RunResult, error)

Run performs one sync pass.

The order matters and follows §8.2: recover interrupted transfers, discover new files, reconcile the whole backlog in one round-trip, then stream what the hub is missing — newest first, so that if a contact window closes mid-backlog the freshest telemetry has already landed.

func (*Agent) SetCompactionDeferEpoch 

func (a *Agent) SetCompactionDeferEpoch(epoch time.Time)

SetCompactionDeferEpoch forwards the issue-#610 enablement epoch to this agent's discovery passes. Zero deactivates the discrimination.

func (*Agent) SetNamespaceExcluder 

func (a *Agent) SetNamespaceExcluder(fn func(ctx context.Context) (map[string]struct{}, error))

SetNamespaceExcluder forwards the dual-role received-namespace exclusion to this agent's discovery passes. nil deactivates it.

func (*Agent) Status 

func (a *Agent) Status(ctx context.Context) (*Stats, error)

Status reports what the spoke still has to send.

func (*Agent) UnfinishedEntries 

func (a *Agent) UnfinishedEntries(ctx context.Context, limit int) ([]*LedgerEntry, error)

UnfinishedEntries returns files that have not reached the hub — those still queued and those that gave up — with the given-up ones first.

The troubleshooting view. It deliberately includes failed entries: a file that exhausted its retries is the one an operator most needs to see, and a pending-only view would hide exactly that.

Exposed through the agent rather than by handing callers the ledger: the handler has no business driving state transitions, and routing reads through here keeps the ledger's mutating API out of the HTTP layer entirely.

type AgentConfig 

type AgentConfig struct {
	Ledger    *Ledger
	Transport SyncTransport
	Backend   storage.Backend

	// HubID names the hub this agent syncs to. Keyed into every ledger row, so
	// changing it starts a fresh sync history rather than resuming another
	// hub's.
	HubID string

	// SpokeID is this edge instance's identity, as registered on the hub.
	SpokeID string

	// MaxAttempts before a file is marked failed. Zero uses DefaultMaxAttempts.
	MaxAttempts int

	// MaxConcurrent bounds simultaneous transfers. Zero means 2 — edge boxes
	// are small, and §8.2 caps this low deliberately.
	MaxConcurrent int

	// BatchSize caps how many files one reconcile asks about; a larger
	// backlog pages. Zero offers the whole backlog in one reconcile — a page
	// the hub refuses as too large is split and retried either way.
	BatchSize int

	Logger zerolog.Logger
}

AgentConfig configures a sync Agent.

type BundleEntry 

type BundleEntry struct {
	Path      string `json:"path"`
	SHA256    string `json:"sha256"`
	SizeBytes int64  `json:"size_bytes"`
}

BundleEntry is one file in a bundle, one JSON object per line.

type BundleIndex 

type BundleIndex struct {
	// contains filtered or unexported fields
}

BundleIndex records which bundles a hub has imported.

This is the replay protection for the air-gap transport. The online families bind a nonce and a 5-minute timestamp window, which works because a request is in flight and the network guarantees "recent". A bundle legitimately sits on a drive for weeks, so replay protection has to be durable state rather than a measurement — a record that survives a restart, as a nonce cache does not.

func NewBundleIndex 

func NewBundleIndex(db *sql.DB, logger zerolog.Logger) (*BundleIndex, error)

NewBundleIndex creates the table if needed and returns a ready index.

func (*BundleIndex) ListBySpoke 

func (i *BundleIndex) ListBySpoke(ctx context.Context, spokeID string, limit int) ([]*ImportedBundle, error)

ListBySpoke returns a spoke's imported bundles, newest first.

func (*BundleIndex) Record 

func (i *BundleIndex) Record(ctx context.Context, b *ImportedBundle) error

Record marks a bundle imported.

Written AFTER the import completes, so a run that died partway can be retried — re-importing is idempotent because every file the hub already holds resolves to already_present.

Conflicts count as completion, not failure: they are a reported outcome needing a human, and refusing to record would mean re-importing every file to retry a handful — or, if the operator resolves them and re-imports, hitting a 409 on a bundle that was never actually recorded.

func (*BundleIndex) Seen 

func (i *BundleIndex) Seen(ctx context.Context, spokeID, bundleID string) (*ImportedBundle, error)

Seen reports whether this spoke's bundle has already been imported.

type BundleReader 

type BundleReader struct {
	// contains filtered or unexported fields
}

BundleReader reads and verifies a bundle directory.

func OpenBundle 

func OpenBundle(dir string, logger zerolog.Logger) (*BundleReader, error)

OpenBundle reads a bundle's manifest.

Does NOT verify — call Verify before trusting anything here. Parsing the manifest first is what lets a caller look up the right spoke secret.

func (*BundleReader) DataPath 

func (r *BundleReader) DataPath(entryPath string) string

DataPath returns the on-disk location of one entry's file.

func (*BundleReader) Entries 

func (r *BundleReader) Entries(ctx context.Context) ([]BundleEntry, error)

Entries streams the bundle's entries for an importer.

Verify must have succeeded first; this does not re-check.

func (*BundleReader) Manifest 

func (r *BundleReader) Manifest() *Manifest

Manifest returns the parsed, not-yet-verified manifest.

func (*BundleReader) Open 

func (r *BundleReader) Open(entryPath string) (io.ReadCloser, error)

Open returns a reader over one entry's file, for an importer to stream.

The CALLER MUST CLOSE the returned reader. An importer streams thousands of files through this, so a missed Close exhausts descriptors on the hub.

The path is validated even though Verify has already checked every declared entry: this is exported, the importer is a separate component, and a traversal primitive that happens to be unreachable today is one refactor away from being reachable. os.Open on data/../../etc/passwd would otherwise succeed.

func (*BundleReader) Verify 

func (r *BundleReader) Verify(ctx context.Context, secret string) error

Verify checks that this bundle is exactly what its spoke signed.

Everything, before anything is committed:

  • the MAC over (bundle ID, spoke, hub, created-at, entries digest)
  • entries.jsonl's own hash, so a human's sha256sum agrees
  • every declared entry's path, and its file's size and digest
  • no file ANYWHERE in the bundle that the manifest does not declare

That last check is what makes the directory format honest. Verifying only the declared direction would leave unsigned, unverified payload sitting on air-gap media inside something an operator has been told is verified — and it covers the whole directory, not just data/, because that is the whole of what a human is handed.

type BundleWriter 

type BundleWriter struct {
	// contains filtered or unexported fields
}

BundleWriter exports ledger entries to a directory bundle.

func NewBundleWriter 

func NewBundleWriter(cfg BundleWriterConfig) (*BundleWriter, error)

NewBundleWriter validates configuration and returns a ready writer.

func (*BundleWriter) Export 

func (w *BundleWriter) Export(ctx context.Context, parent string, entries []*LedgerEntry, now time.Time) (*ExportResult, error)

Export writes entries to a new bundle directory under parent.

Order matters and is the crash-safety property: data files first, then entries.jsonl, then the manifest LAST. A partial export therefore has no manifest and cannot be mistaken for a complete one.

That ordering protects against a crash HERE. It does NOT survive the copy to removable media — `cp -r` walks in directory order and writes manifest.json before data/, so an interrupted copy leaves a complete manifest over a partial tree. The real completeness signal is BundleReader.Verify, which re-hashes every file. Do not treat the manifest's presence as sufficient.

type BundleWriterConfig 

type BundleWriterConfig struct {
	// Backend reads the spoke's Parquet files.
	Backend storage.Backend

	SpokeID string
	HubID   string

	// Secret is the hub-issued shared secret that signs the manifest.
	Secret string

	Logger zerolog.Logger
}

BundleWriterConfig configures an export.

type CollectingRegistrar 

type CollectingRegistrar struct {
	// contains filtered or unexported fields
}

CollectingRegistrar buffers registered files instead of proposing them.

The seam that makes batching possible without changing Receiver: the online hub passes a RegisterFile that proposes to Raft immediately, and the importer passes this, draining it at ImportBatchSize.

Not safe for concurrent use. That is sound only because Importer.Import holds a mutex for the whole call: one collector is shared across every import, so without that lock two overlapping imports would append to the same buffer and Reset would truncate the other's pending registrations.

func NewCollectingRegistrar 

func NewCollectingRegistrar() *CollectingRegistrar

NewCollectingRegistrar returns an empty collector.

func (*CollectingRegistrar) Drain 

func (c *CollectingRegistrar) Drain() []*ReceivedFile

Drain returns the buffered files and empties the buffer.

func (*CollectingRegistrar) Len 

func (c *CollectingRegistrar) Len() int

Len reports how many files are buffered.

func (*CollectingRegistrar) Register 

func (c *CollectingRegistrar) Register(_ context.Context, f *ReceivedFile) error

Register is the ReceiverConfig.RegisterFile hook.

func (*CollectingRegistrar) Reset 

func (c *CollectingRegistrar) Reset()

Reset discards anything buffered, so a new import does not inherit files left over from one that failed partway.

type Conflict 

type Conflict struct {
	Path string // storage-relative path, as the spoke knows it
	// TheirSHA256 is the digest the hub holds. The spoke's own digest is in
	// its ledger entry for the same path; the pair is what an operator needs
	// to work out which side is wrong.
	TheirSHA256 string
}

Conflict is one same-path-different-content disagreement between spoke and hub.

type DeliveryState 

type DeliveryState struct {
	State SyncState
	Note  string
}

DeliveryState is a ledger row's state plus the note the eligibility gate needs to recognize compacted outputs.

type DestinationPolicy 

type DestinationPolicy struct {
	// contains filtered or unexported fields
}

DestinationPolicy decides which filesystem paths a bundle may use.

Every other Arc write path goes through a storage backend, which confines it to the storage root. A bundle cannot: a USB mount is outside that root by definition, so an operator-supplied path reaches the filesystem directly.

The endpoints are admin-only, so this is not a privilege boundary — an admin can already do worse. It guards against MISTAKES, one of which is not obvious: exporting into Arc's own storage root makes the next discovery pass find the exported copies and queue them for sync, which fans out on every export.

func NewDestinationPolicy 

func NewDestinationPolicy(allowedDirs []string, storageRoot string) (*DestinationPolicy, error)

NewDestinationPolicy resolves the allow-list once, at startup.

Resolving here rather than per-request means a symlink swapped later cannot change what a path means mid-flight, and an unresolvable entry is a startup error rather than a confusing runtime refusal.

func (*DestinationPolicy) Enabled 

func (p *DestinationPolicy) Enabled() bool

Enabled reports whether any destination is permitted.

func (*DestinationPolicy) Resolve 

func (p *DestinationPolicy) Resolve(requested string) (string, error)

Resolve validates a requested path and returns its absolute, symlink-free form.

The returned path is what callers must use — not the requested one — so a later component cannot re-resolve differently.

type Discoverer 

type Discoverer struct {
	// contains filtered or unexported fields
}

Discoverer walks local storage and tracks new files in the ledger.

Separate from Agent because BOTH transports need it and they are independently enabled: a fully air-gapped spoke runs no agent at all, and without its own discovery its ledger would stay empty forever — every export answering "nothing to export" while files piled up on disk.

func NewDiscoverer 

func NewDiscoverer(ledger *Ledger, backend storage.Backend, hubID string, logger zerolog.Logger) (*Discoverer, error)

NewDiscoverer validates configuration and returns a ready discoverer.

func (*Discoverer) Discover 

func (d *Discoverer) Discover(ctx context.Context) (int, error)

Discover tracks every syncable local file not already in the ledger.

Returns how many entries were newly tracked. Idempotent: a file already tracked is skipped before it is hashed, so running this every pass costs a listing rather than a re-read of the whole corpus.

func (*Discoverer) SetCompactionDeferEpoch 

func (d *Discoverer) SetCompactionDeferEpoch(epoch time.Time)

SetCompactionDeferEpoch activates epoch-based compacted-file discrimination (issue #610). Zero deactivates it.

func (*Discoverer) SetNamespaceExcluder 

func (d *Discoverer) SetNamespaceExcluder(fn func(ctx context.Context) (map[string]struct{}, error))

SetNamespaceExcluder installs the received-namespace exclusion for dual-role nodes. nil deactivates it.

type ExportResult 

type ExportResult struct {
	BundleID  string
	Dir       string
	FileCount int
	Bytes     int64
	Duration  time.Duration

	// Skipped is how many eligible entries were dropped because their source
	// file vanished before export (compaction or retention). Set by the
	// Exporter's pre-check, not the writer.
	Skipped int
}

ExportResult summarizes one export.

type Exporter 

type Exporter struct {
	// contains filtered or unexported fields
}

Exporter writes pending ledger entries to an air-gap bundle.

The export half of the sneakernet transport: it selects what to send, writes it, and advances the ledger. It deliberately does NOT delete anything — sync is a copy, and local retention stays the operator's decision.

func NewExporter 

func NewExporter(cfg ExporterConfig) (*Exporter, error)

NewExporter validates configuration and returns a ready exporter.

func (*Exporter) ApplyAck 

func (e *Exporter) ApplyAck(ctx context.Context, dir string) (*AckResult, error)

ApplyAck reads an acknowledgment from a returned bundle directory and advances the ledger.

The return leg. An operator plugs the drive back into the spoke after it has been to the hub, and this is what finally moves those files from `exported` to `synced` — making them prunable, which is the only thing that stops an air-gap ledger growing without bound.

The destination policy applies here too: the path is operator-supplied and reaches the filesystem directly, exactly as on the export side.

func (*Exporter) DismissFailed 

func (e *Exporter) DismissFailed(ctx context.Context, path string) (int64, error)

DismissFailed moves failed rows to operator-dismissed skipped.

func (*Exporter) EntriesByState 

func (e *Exporter) EntriesByState(ctx context.Context, state SyncState, limit int) ([]*LedgerEntry, error)

EntriesByState lists ledger rows in one explicit state.

func (*Exporter) Export 

func (e *Exporter) Export(ctx context.Context, dest string, limit int) (*ExportResult, error)

Export writes one bundle to dest and marks its entries exported.

The ledger advance happens AFTER the bundle is written and signed. The reverse order would mark files exported that a failed write never included, and only an operator noticing the gap would ever bring them back.

func (*Exporter) RequeueFailed 

func (e *Exporter) RequeueFailed(ctx context.Context, path string) (int64, error)

RequeueFailed returns failed (and operator-dismissed) rows to pending.

func (*Exporter) Revert 

func (e *Exporter) Revert(ctx context.Context, bundleID string) (int64, error)

Revert returns a bundle's entries to pending, for a drive that never arrived.

func (*Exporter) Status 

func (e *Exporter) Status(ctx context.Context) (*Stats, error)

Status reports the ledger summary for this hub.

Mirrors Agent.Status: both are pure ledger reads needing no transport, and an air-gap-only spoke has no agent to ask. Without this the operator who most needs the `exported` count — the one whose files are on a drive somewhere — would be the only one unable to see it.

func (*Exporter) UnfinishedEntries 

func (e *Exporter) UnfinishedEntries(ctx context.Context, limit int) ([]*LedgerEntry, error)

UnfinishedEntries returns files that have not reached the hub, given-up ones first. Mirrors Agent.UnfinishedEntries.

type ExporterConfig 

type ExporterConfig struct {
	Ledger *Ledger
	Writer *BundleWriter
	Policy *DestinationPolicy

	// Discoverer finds new local files before an export selects from them.
	//
	// Required: an air-gapped spoke runs no sync agent, so without its own
	// discovery the ledger would never be populated and every export would
	// answer "nothing to export" while files piled up on disk.
	Discoverer *Discoverer

	HubID string

	// MaxFiles and MaxBytes cap one bundle. Zero uses the defaults above.
	MaxFiles int
	MaxBytes int64

	Logger zerolog.Logger
}

ExporterConfig configures air-gap export.

type HTTPTransport 

type HTTPTransport struct {
	// contains filtered or unexported fields
}

HTTPTransport pushes files to a hub over HTTPS.

The v1 transport. Everything protocol-shaped — the identity rule, the outcome taxonomy, resume — lives above this in the interface, so an S3-relay or sneakernet transport reuses it unchanged and only the wire format differs.

func NewHTTPTransport 

func NewHTTPTransport(cfg HTTPTransportConfig) (*HTTPTransport, error)

NewHTTPTransport validates configuration and returns a ready transport.

func (*HTTPTransport) PutFile 

func (t *HTTPTransport) PutFile(ctx context.Context, hubID string, entry *LedgerEntry, body io.Reader, offset int64) (*PutResult, error)

PutFile streams one file to the hub, resuming from offset.

func (*HTTPTransport) Reconcile 

func (t *HTTPTransport) Reconcile(ctx context.Context, hubID string, pending []*LedgerEntry) (*ReconcileResult, error)

Reconcile asks the hub which of the pending files it already holds.

type HTTPTransportConfig 

type HTTPTransportConfig struct {
	// BaseURL is the hub's root, e.g. https://ground-station.example.com.
	BaseURL string

	// SpokeID is this spoke's registered identity, bound into every MAC.
	SpokeID string

	// Secret is the hub-issued shared secret. Never logged, never persisted
	// by this package — it arrives from the environment and stays in memory.
	Secret string

	// APIToken is an Arc API token for the hub's token middleware, which
	// gates /api/v1/sync at write level ahead of the per-spoke HMAC. Optional:
	// empty means no Authorization header, which only works against a hub
	// running with auth disabled. Same handling discipline as Secret.
	APIToken string

	// Timeout bounds a single request. Zero means 30 minutes, matching the
	// hub's receive timeout: a large Parquet file over a constrained link is
	// the expected case, not an anomaly.
	Timeout time.Duration

	// Client overrides the HTTP client, for tests and for callers that need a
	// custom TLS config.
	Client *http.Client
}

HTTPTransportConfig configures an HTTPTransport.

type HeldFile 

type HeldFile struct {
	SHA256    string
	Compacted bool
}

HeldFile is what Lookup reports for one receipt: the delivered content's digest, and whether the hub's own compaction has since consumed the FILE (the content lives on inside a compacted output — the receipt stays valid, but nothing at the original path exists to stat).

type HubIndex 

type HubIndex struct {
	// contains filtered or unexported fields
}

HubIndex records what a hub has received, so reconcile can answer without touching parquet bytes.

§5.1 assumes the hub answers missing/present/conflicts "from its manifest in O(N) lookups, no I/O on the parquet bytes". That manifest is the Raft FSM, which exists only in cluster mode — in OSS standalone there is no file index at all, and distinguishing present from conflict would mean reading and hashing every candidate file. For a spoke returning from a long outage that is gigabytes of disk reads on the one request that exists to be cheap.

So the hub keeps its own index. The receive path already knows every file's path, digest, and size at commit time, so recording them costs one small write on a path that has just done a full hash and a promote.

It is keyed by the SPOKE's path rather than the hub's namespaced path, because that is what a spoke asks about: it has no idea the hub prepends a namespace.

func NewHubIndex 

func NewHubIndex(db *sql.DB, logger zerolog.Logger) (*HubIndex, error)

NewHubIndex creates the index and initializes its schema.

func (*HubIndex) CountForSpoke 

func (h *HubIndex) CountForSpoke(ctx context.Context, spokeID string) (int64, error)

CountForSpoke returns how many files the hub holds from a spoke.

func (*HubIndex) Forget 

func (h *HubIndex) Forget(ctx context.Context, spokeID, sourcePath string) error

Forget removes a spoke's record for a path.

TODO(#611): NOT YET CALLED IN PRODUCTION, and the gap has teeth. If a GENUINE hub-side removal deletes a synced file without calling this, the index keeps claiming the hub holds it, so reconcile reports it `present` and the spoke marks it synced. A spoke configured with delete_after_sync (phase 3) would then be free to delete its only copy of data the hub no longer has.

Forget is for genuine removals ONLY — content the hub really no longer holds. Hub-side COMPACTION of spoke namespaces (#619) deletes files too, but it PRESERVES their content inside the compacted output, so it goes through MarkCompacted instead: the receipt stays valid and reconcile keeps vouching, which is correct. Do not "fix" that to Forget.

What keeps the Forget gap from being a live hazard today: delete_after_sync does not exist yet, and no hub-side path REMOVES spoke content — retention operates on Arc's own ingested databases (and its partition parsing does not currently reach spoke namespaces). Whoever adds hub retention over spoke prefixes, or ships phase 3, MUST wire this first.

func (*HubIndex) ForgetBatch 

func (h *HubIndex) ForgetBatch(ctx context.Context, spokeID string, paths []string) error

ForgetBatch removes several of a spoke's records in one statement per chunk.

Batched because the alternative — one DELETE per stale row inside a request handler — lands on the SQLite handle shared with ingest file-registration, auth, and audit. A spoke reconciling after a retention sweep can present thousands of stale rows at once, and issuing that many separate implicit transactions on the single writer would contend with ingest for the duration of the request.

func (*HubIndex) Lookup 

func (h *HubIndex) Lookup(ctx context.Context, spokeID string, paths []string) (map[string]HeldFile, error)

Lookup returns the receipts the hub holds for the given paths from one spoke.

Batched deliberately: reconcile asks about thousands of paths at once, and issuing one query per path would make the round-trip the design saved on the wire reappear as a query storm against SQLite. A path absent from the map is one the hub does not hold.

func (*HubIndex) MarkCompacted 

func (h *HubIndex) MarkCompacted(ctx context.Context, spokeID string, sourcePaths []string) error

MarkCompacted stamps compacted_at on the receipts for source paths whose files the hub's own compaction consumed (#619). Idempotent — recovery can re-fire it — and deliberately an UPDATE, never an INSERT: a path with no receipt was not received from this spoke and gets no bookkeeping. Chunked like Lookup.

func (*HubIndex) Record 

func (h *HubIndex) Record(ctx context.Context, r *ReceivedRecord) error

Record notes that the hub holds a file.

Upserts on (spoke_id, source_path): the digest is allowed to change here because the receive path only reaches this point after verifying the bytes, so a differing digest means the file was legitimately replaced rather than that two spokes collided. A conflicting upload never gets this far — it is refused with 409 before promotion.

type ImportResult 

type ImportResult struct {
	BundleID  string
	SpokeID   string
	CreatedAt time.Time

	Committed      int
	AlreadyPresent int
	BytesWritten   int64

	// AckPaths are the paths the hub now holds, for the acknowledgment.
	// Committed and already-present both qualify: from the spoke's side they
	// are the same fact. Conflicted paths are excluded — the hub holds
	// DIFFERENT content there, so the spoke's copy was never delivered.
	AckPaths []string

	// AckWritten reports whether the acknowledgment reached the drive. False
	// means the import succeeded but the spoke cannot learn of it from this
	// drive, so its files stay `exported` and ride the next bundle.
	AckWritten bool

	// Conflicts are reported in full, not counted: each needs a human to
	// decide which copy is right, and a count alone would not say which.
	Conflicts []Conflict

	// Recorded reports whether the dedup ledger write succeeded. False means
	// the files are committed but a re-import of the same drive will not be
	// refused — harmless, since it resolves to already_present, but it makes
	// /history disagree with reality and the operator should know why.
	Recorded bool

	Duration time.Duration
}

ImportResult summarizes one import.

type ImportedBundle 

type ImportedBundle struct {
	SpokeID    string
	BundleID   string
	CreatedAt  time.Time
	ImportedAt time.Time
	FileCount  int64
	BytesTotal int64
	Conflicts  int64
}

ImportedBundle is one row of the dedup ledger.

type Importer 

type Importer struct {
	// contains filtered or unexported fields
}

Importer reads a verified bundle into hub storage.

Imports are serialized. One collector and one staging area are shared across calls, and a hub takes one drive at a time by nature — but the endpoint is a concurrent Fiber handler, so nothing structural stops two operators (or one retry after a client timeout on a four-hour request) from overlapping. Without the lock they collide in the staging area: both stage the same path, one promotes, the other fails with "file not found" — and Reset truncates the other import's buffered registrations, leaving committed files outside the manifest with no error and no log.

func NewImporter 

func NewImporter(cfg ImporterConfig) (*Importer, error)

NewImporter validates configuration and returns a ready importer.

func (*Importer) Import 

func (i *Importer) Import(ctx context.Context, dir string) (*ImportResult, error)

Import verifies a bundle and commits its files.

Order: open, identity checks, dedup, verify EVERYTHING, then commit. No byte reaches its final path until the whole bundle has been verified — the property that makes an artifact carried on removable media safe to trust.

type ImporterConfig 

type ImporterConfig struct {
	// Receiver commits each file: staging, hash, verify, promote, index. Reused
	// unchanged from the online path so both transports land bytes identically.
	//
	// IMPORTANT: this Receiver must be constructed with a RegisterFile that
	// COLLECTS rather than proposes — see CollectingRegistrar. The online
	// Receiver proposes to the Raft manifest once per file, which is fine at
	// one-file-per-HTTP-request but would emit 10,000 proposals in a tight
	// loop for a 10,000-file bundle.
	Receiver *Receiver

	// Collector is the sink the Receiver's RegisterFile writes into. The
	// importer drains it in batches of ImportBatchSize.
	Collector *CollectingRegistrar

	// Index is the dedup ledger.
	Index *BundleIndex

	// Registry resolves a spoke's secret to verify the manifest signature.
	Registry *Registry

	// HubID is this hub's identity. A bundle naming a different hub is
	// refused: the MAC alone does not stop that, since a bundle for another
	// hub validates fine under the same spoke's secret, and a scavenged drive
	// would otherwise import anywhere that spoke is registered.
	HubID string

	// MaxFiles refuses a manifest declaring more than this. Enforced
	// independently of the exporting spoke's own cap, which is advisory
	// against a hostile manifest.
	MaxFiles int64

	// FlushManifest commits a batch of registered files to the Raft manifest.
	// Nil in standalone mode, where there is no manifest.
	FlushManifest func(ctx context.Context, files []*ReceivedFile) error

	Logger zerolog.Logger
}

ImporterConfig configures bundle import.

type Ledger 

type Ledger struct {
	// contains filtered or unexported fields
}

Ledger is the spoke-side record of sync progress, backed by the shared SQLite database (cfg.Auth.DBPath).

Concurrency: *sql.DB is already safe for concurrent use and this type holds no in-memory state, so it takes no mutex of its own. Holding an application lock across SQLite I/O would serialize readers behind writers for no benefit (see the SQLite Review Checklist in CLAUDE.md).

func NewLedger 

func NewLedger(db *sql.DB, logger zerolog.Logger) (*Ledger, error)

NewLedger creates the ledger and initializes its schema.

func (*Ledger) ClearMeta 

func (l *Ledger) ClearMeta(ctx context.Context, key string) error

ClearMeta removes a sync_meta key. Used when the compaction defer gate is INACTIVE while compaction may run (issue #610): the epoch's meaning is "the gate has been continuously active since this instant", so any ungated run must invalidate it — the next activation re-stamps a fresh epoch and outputs from the ungated period classify as legacy (sync once; a bounded duplicate, never a loss).

func (*Ledger) DeliveryStates 

func (l *Ledger) DeliveryStates(ctx context.Context, hubID string, paths []string) (map[string]DeliveryState, error)

DeliveryStates returns state+note for each of paths that has a ledger row. Paths without a row are absent from the result. IN-chunked at 500 binds.

func (*Ledger) DismissFailed 

func (l *Ledger) DismissFailed(ctx context.Context, hubID, path string) (int64, error)

DismissFailed moves failed rows to skipped with the operator note — terminal, pruned after retention, and reversible via RequeueFailed. path == "" dismisses every failed row for the hub.

Skipped, not deleted: a deleted row whose file still exists would be re-tracked by the next discovery pass and resurrect as pending — the exact noise the operator asked to silence.

func (*Ledger) EnsureMetaOnce 

func (l *Ledger) EnsureMetaOnce(ctx context.Context, key, value string) (string, error)

EnsureMetaOnce stores value under key unless the key already exists, and returns the stored (possibly pre-existing) value. Used for the compaction defer epoch: set on first activation, stable forever after.

func (*Ledger) EntriesByState 

func (l *Ledger) EntriesByState(ctx context.Context, hubID string, state SyncState, limit int) ([]*LedgerEntry, error)

EntriesByState lists rows in one explicit state, newest partition first. The troubleshooting view for states Unfinished hides — above all `skipped`, whose rows (with their notes) are otherwise invisible: an operator cannot selectively requeue a dismissed entry they cannot list (#612 follow-up).

func (*Ledger) Get 

func (l *Ledger) Get(ctx context.Context, hubID, path string) (*LedgerEntry, error)

Get returns a single entry, or ErrNotFound.

func (*Ledger) MarkConflicted 

func (l *Ledger) MarkConflicted(ctx context.Context, hubID, path, errMsg string) error

MarkFailed records a transfer failure. If the entry has reached maxAttempts it becomes terminally failed; otherwise it returns to pending for retry.

bytes_sent is deliberately preserved on the retry path — a failure is usually a dropped link, and the bytes the hub already accepted are still valid. Discarding the checkpoint would restart a large file from zero on exactly the link least able to afford it. MarkConflicted terminally fails an entry the hub reports as conflicting.

Separate from MarkFailed because that transition is pending→in_flight→failed and requires the row to be in_flight: a conflict found during RECONCILE is still 'pending' — no transfer was ever started for it — so MarkFailed would match no rows and silently leave it pending, to be re-offered in every future reconcile payload forever.

Terminal immediately, with no attempt counting: retrying cannot resolve a content disagreement, and the same path holding different bytes on both sides needs a human to decide which copy is right.

func (*Ledger) MarkExported 

func (l *Ledger) MarkExported(ctx context.Context, hubID, path, bundleID string) error

MarkExported records that a file was written to an air-gap bundle.

Only from pending: an in_flight row belongs to a running network transfer, and exporting it concurrently would put the same file on two paths with no way to reconcile which acknowledgment arrived.

This is what stops a capped export from starving old files. Pending orders partition_time DESC, so without a state change every export would re-take the newest N and the oldest files would never leave the box.

func (*Ledger) MarkFailed 

func (l *Ledger) MarkFailed(ctx context.Context, hubID, path, errMsg string, maxAttempts int) error

func (*Ledger) MarkInFlight 

func (l *Ledger) MarkInFlight(ctx context.Context, hubID, path string) error

MarkInFlight moves an entry to in_flight and increments its attempt count.

func (*Ledger) MarkSkipped 

func (l *Ledger) MarkSkipped(ctx context.Context, hubID, path, note string) error

MarkSkipped records that an entry's source file vanished from storage before delivery — compaction or retention got to it first.

Legal from pending (the export pre-check finds the file gone) AND from in_flight (the network path only learns mid-transfer: openAt streams through an io.Pipe, so a missing file surfaces as a PutFile error after MarkInFlight has already run). last_attempt is stamped because it is the column PruneSkipped keys on — synced_at stays NULL on this path.

func (*Ledger) MarkSynced 

func (l *Ledger) MarkSynced(ctx context.Context, hubID, path string) error

MarkSynced records that the hub has the file.

Call this ONLY on a 2xx or AlreadyPresent from the hub — never optimistically before the ack. The whole exactly-once-effect property rests on this rule: advancing early converts a lost ack into permanent data loss, whereas advancing late costs one redundant entry in the next reconcile.

bytes_sent is set to size_bytes so a synced row reads consistently.

func (*Ledger) Pending 

func (l *Ledger) Pending(ctx context.Context, hubID string, limit int) ([]*LedgerEntry, error)

func (*Ledger) PendingPage 

func (l *Ledger) PendingPage(ctx context.Context, hubID string, limit int, after *LedgerEntry) ([]*LedgerEntry, error)

PendingPage returns one page of pending entries strictly AFTER the cursor row, in Pending's order (partition_time DESC, id ASC). A nil cursor starts from the top. Keyset pagination: the agent's old offered-map approach re-fetched every previously-offered row on every page — O(pages²) row scans across a large backlog (2026-08-19 audit M5). Rows the pass leaves unresolved (conflicts, exhausted retries) sit BEFORE the cursor and are naturally not re-offered within the pass, which is the same property the offered map bought, at the cost of one range-bounded scan (plus the same ORDER BY sort Pending already pays) per page.

Time comparison stays in the Go time.Time domain on both sides — partition_time is written as a Go time.Time parameter (see the domain note on PruneSynced).

func (*Ledger) PruneSkipped 

func (l *Ledger) PruneSkipped(ctx context.Context, retentionDays int) (int64, error)

PruneSkipped deletes skipped rows older than retentionDays.

Same shape and discipline as PruneSynced (resolved id bound, 1000-row batches, ctx checked between batches, Go-time-vs-Go-time comparison — see the domain note there). Keys on last_attempt, which MarkSkipped stamps: synced_at is NULL for a file that never arrived anywhere, so PruneSynced's predicate can never reclaim these rows.

func (*Ledger) PruneSynced 

func (l *Ledger) PruneSynced(ctx context.Context, retentionDays int) (int64, error)

PruneSynced deletes synced entries older than retentionDays, across ALL hubs. Unlike every other method here it is deliberately not hub-scoped: retention is a local disk-space concern, not a per-hub policy. When multi-hub becomes real and per-hub retention is wanted, this needs a hubID parameter — the signature is the place that will have to change.

Batched at 1000 rows with an ORDER BY on the primary key: a single unbounded DELETE on a table with millions of rows holds the SQLite write lock for the duration, blocking ingest file registration and auth token updates. The context is checked between batches so shutdown is not delayed.

No incremental_vacuum: the freed pages are reused by subsequent inserts, and vacuuming a continuously-written table just causes write amplification.

func (*Ledger) RecordProgress 

func (l *Ledger) RecordProgress(ctx context.Context, hubID, path string, bytesSent int64) error

RecordProgress updates the resume checkpoint for an in-flight transfer.

bytesSent is the absolute offset the hub has accepted, not a delta, so a retry that re-sends from an earlier offset cannot inflate it.

func (*Ledger) RecoverInFlight 

func (l *Ledger) RecoverInFlight(ctx context.Context) (int64, error)

RecoverInFlight reverts in_flight entries to pending. Call once at startup.

An in_flight row can only have been written by a transfer that is no longer running — the process that owned it is gone. Reverting makes those files eligible for the next reconcile, where the hub reports any that actually landed as `present` and the spoke advances them without re-sending a byte.

func (*Ledger) RequeueFailed 

func (l *Ledger) RequeueFailed(ctx context.Context, hubID, path string) (int64, error)

RequeueFailed returns failed rows — and operator-dismissed rows, so a dismissal is reversible — to pending for a fresh delivery attempt. path == "" requeues every eligible row for the hub. Attempts reset to 0: the operator is explicitly granting a new retry budget. bytes_sent is kept so a partial transfer resumes; a checkpoint the hub no longer holds already restarts from zero via the stale-checkpoint path.

The state predicate lives in SQL so only legal sources transition: vanished-file and compacted-output skips have nothing to deliver (or would duplicate) and must never re-enter the queue.

func (*Ledger) RevertExported 

func (l *Ledger) RevertExported(ctx context.Context, hubID, bundleID string) (int64, error)

RevertExported returns a bundle's files to pending, for a drive that was lost, damaged, or never delivered.

Without this an exported row is a dead end until an ack that will never come: the files are neither retryable nor visible as a problem. Scoped to one bundle ID so recovering one lost drive does not disturb others in transit.

Returns the number of rows reverted.

func (*Ledger) Stats 

func (l *Ledger) Stats(ctx context.Context, hubID string) (*Stats, error)

Stats returns a summary for a hub, for the status endpoint and operators.

func (*Ledger) SweepSkippedRows 

func (l *Ledger) SweepSkippedRows(ctx context.Context, hubID string, exists func(context.Context, string) (bool, error)) (int64, error)

SweepSkippedRows deletes the FILE-BACKED skipped rows — compacted outputs and operator-dismissed entries — whose file no longer exists (consumed by compaction, or removed by retention). These two classes are exempt from PruneSkipped: pruning one whose file still exists would let discovery re-track the file (re-syncing a compacted output, or resurrecting a dismissed failure as pending). This sweep is their only reclamation path and runs regardless of the retention setting. exists is consulted per row; an exists ERROR keeps the row (fail-safe: never delete bookkeeping on an uncertain answer).

func (*Ledger) Track 

func (l *Ledger) Track(ctx context.Context, e *LedgerEntry) error

Track records a file as pending sync to a hub. It is idempotent: a file already tracked for this hub is left untouched, whatever state it is in.

This matters because discovery re-walks the manifest every tick. Without DO NOTHING, a file already synced would be reset to pending and re-sent forever.

PRECONDITION — caller must guarantee path immutability. Because a conflict does nothing, re-tracking a path whose content changed keeps the ORIGINAL sha256 and size, and the ledger would then assert the hub holds content it has never seen. Arc satisfies this: compaction and retention produce new immutable paths rather than rewriting one in place (§13), so a given path's bytes never change. If a future producer breaks that, this must become an upsert that detects the divergence — §6.1 treats same-path-different-SHA as a 409-class alarm, and silently discarding the evidence here would hide it.

func (*Ledger) TrackBatch 

func (l *Ledger) TrackBatch(ctx context.Context, entries []*LedgerEntry) (int, error)

TrackBatch records many files in one transaction.

Discovery after a long disconnection can surface thousands of files at once; inserting them individually would mean one implicit transaction (and one fsync) each. A single transaction turns that into one fsync.

func (*Ledger) TrackCompactedOutput 

func (l *Ledger) TrackCompactedOutput(ctx context.Context, hubID, path string) error

TrackCompactedOutput records a compacted output as already-delivered content that must never sync (state skipped, NoteCompactedOutput).

INSERT OR IGNORE: the observer can fire more than once for one output — manifest recovery retries until input deletion fully succeeds — and a legacy compacted file may already be tracked as a synced row, which must be left alone (it DID sync).

func (*Ledger) TrackedPaths 

func (l *Ledger) TrackedPaths(ctx context.Context, hubID string) (map[string]struct{}, error)

TrackedPaths returns every path the ledger knows for a hub, as a set.

Discovery used to point-SELECT each stored object against the ledger — 100k files meant 100k queries per pass on the SQLite handle shared with auth and ingest (2026-08-19 audit M4). One scan into a set costs one query and O(corpus) transient memory on a box that already listed the same corpus to discover it.

func (*Ledger) Unexported 

func (l *Ledger) Unexported(ctx context.Context, hubID string, limit int) ([]*LedgerEntry, error)

Unexported returns files eligible for an air-gap bundle, newest first.

Distinct from Pending only in intent today — both select pending rows — but they diverge the moment anything else can enqueue work, and the export path should not silently inherit a change made for the network path. Ordering matches Pending so a bundle carries the freshest telemetry first.

A limit <= 0 returns everything eligible.

func (*Ledger) Unfinished 

func (l *Ledger) Unfinished(ctx context.Context, hubID string, limit int) ([]*LedgerEntry, error)

Pending returns entries awaiting transfer to a hub, newest partition first.

Newest-first is a deliberate ordering, not an incidental one: when a contact window closes mid-backlog, the freshest telemetry has already reached the hub and backfill catches up on a later pass.

A limit <= 0 returns all pending entries. Unfinished returns entries that have not reached the hub, in either state an operator cares about: still queued, or given up on.

Separate from Pending rather than a widening of it, because the sync pass depends on Pending returning ONLY 'pending' rows — a failed entry re-offered to the hub would restart a transfer the retry cap deliberately stopped. This is the troubleshooting view: a file that exhausted its attempts is the one an operator most needs to see, and it is exactly the one Pending hides.

Failed entries sort first: they need a decision, whereas pending ones are simply waiting for the next pass.

A limit <= 0 returns everything unfinished.

type LedgerEntry 

type LedgerEntry struct {
	ID            int64
	HubID         string
	Path          string // storage-relative path, as the spoke knows it
	SHA256        string // from the manifest; the integrity anchor
	SizeBytes     int64
	Database      string
	Measurement   string
	PartitionTime time.Time
	DiscoveredAt  time.Time
	State         SyncState
	Attempts      int
	LastAttempt   *time.Time
	SyncedAt      *time.Time

	// BytesSent is the resume checkpoint: how many bytes of this file the hub
	// has already accepted. A transfer that dies mid-file resumes from here
	// rather than restarting, which is the difference between "eventually
	// drains" and "never completes" on a link whose contact window is shorter
	// than the file.
	BytesSent int64

	LastError string

	// ExportedAt / ExportedBundleID identify the air-gap bundle a file left on,
	// for the operator whose drive did not arrive. Nil and empty unless the
	// entry is (or once was) exported. RevertExported filters on the bundle ID
	// in SQL, but an operator needs to READ it to know which ID to revert.
	ExportedAt       *time.Time
	ExportedBundleID string
}

LedgerEntry is one file's sync state with respect to one hub.

type Manifest 

type Manifest struct {
	Version int `json:"version"`

	// BundleID is a ULID, and the hub's replay key. Format-validated on import
	// because it is attacker-chosen: a compromised spoke signs whatever it
	// likes, and this string reaches a SQLite key and operator log lines.
	BundleID string `json:"bundle_id"`

	SpokeID string `json:"spoke_id"`

	// HubID is the hub this bundle is FOR. The import side must reject a
	// mismatch: the MAC alone does not, since a bundle for another hub
	// validates fine under the same spoke secret, and a scavenged drive would
	// otherwise import anywhere that spoke is registered.
	HubID string `json:"hub_id"`

	// CreatedAt is bound into the MAC but NOT enforced as a freshness window —
	// a bundle legitimately crosses an air gap over weeks. Surfaced to
	// operators so a 409 on a very old bundle is diagnosable.
	CreatedAt int64 `json:"created_at"`

	EntryCount int64 `json:"entry_count"`
	TotalBytes int64 `json:"total_bytes"`

	// EntriesSHA256 covers entries.jsonl as bytes, so a human with sha256sum
	// can check it without understanding the canonical entry encoding.
	EntriesSHA256 string `json:"entries_sha256"`

	// EntriesDigest is the canonical digest the MAC binds. Distinct from
	// EntriesSHA256: this one is order- and formatting-independent, so it
	// survives a reader that rewrites the file, while the raw hash does not.
	EntriesDigest string `json:"entries_digest"`

	MAC string `json:"mac"`
}

Manifest is the signed header of a bundle.

Deliberately small and fixed-size: the entry list lives in a separate newline-delimited file so neither export nor import has to hold hundreds of thousands of entries in memory, and so a human can run `sha256sum entries.jsonl` — which is the point of a directory bundle over an archive.

type MemoryTransport 

type MemoryTransport struct {

	// MaxReconcileEntries, when > 0, refuses larger reconcile batches with
	// ReconcileTooLargeError — mirroring the real hub's cap so agent paging
	// and 413 splitting are testable. Set before use; not synchronized.
	MaxReconcileEntries int
	// contains filtered or unexported fields
}

MemoryTransport is an in-process SyncTransport backed by a map.

It exists for two reasons. First, it keeps SyncTransport honest: an interface with no implementation is a guess, and writing this one is what surfaced the fact that PutFile needs a result type rather than a bare error. Second, the agent (PR 8) needs a hub it can drive deterministically — exercising lost acks, conflicts, mid-stream drops, and backpressure against a real HTTP server means either a fragile fake server or no test at all.

It implements the same identity rule as a real hub (§6.1): a path is absent (write), present with the same digest (no-op), or present with a different digest (conflict, never overwrite). It is NOT a hub — it does no authentication, no namespacing, and keeps bytes in memory.

func NewMemoryTransport 

func NewMemoryTransport() *MemoryTransport

NewMemoryTransport returns an empty in-memory hub.

func (*MemoryTransport) Close 

func (m *MemoryTransport) Close() error

Close makes every subsequent call return ErrTransportClosed.

func (*MemoryTransport) Has 

func (m *MemoryTransport) Has(hubID, path string) (string, bool)

Has reports whether the hub holds a path, and its digest.

func (*MemoryTransport) PutFile 

func (m *MemoryTransport) PutFile(ctx context.Context, hubID string, entry *LedgerEntry, body io.Reader, offset int64) (*PutResult, error)

PutFile stores the streamed bytes, applying the §6.1 identity rule and verifying the digest before committing — the same verify-before-commit ordering a real hub uses, so a mismatch never lands as stored content.

func (*MemoryTransport) PutOrder 

func (m *MemoryTransport) PutOrder() []string

PutOrder returns the paths PutFile was called with, in order.

func (*MemoryTransport) Reconcile 

func (m *MemoryTransport) Reconcile(ctx context.Context, hubID string, pending []*LedgerEntry) (*ReconcileResult, error)

Reconcile partitions the pending set exactly as a hub would: matching digest is present, differing digest is a conflict, unknown path is missing.

func (*MemoryTransport) ScriptPut 

func (m *MemoryTransport) ScriptPut(hubID, path string, results ...*PutResult)

ScriptPut queues results for the next PutFile calls on (hubID, path), letting a test drive the partial/backpressure/mismatch branches deterministically. Queued results are consumed in order; once empty, PutFile behaves normally.

Keyed by hub as well as path: with a path-only key, a script intended for one hub would be consumed by a transfer to another, so a "hub A throttles, hub B does not" test would silently assert the opposite.

func (*MemoryTransport) Seed 

func (m *MemoryTransport) Seed(hubID, path, sha256Hex string, size int64)

Seed pre-populates the hub with a file, as though a previous sync had delivered it. Used to set up the lost-ack and conflict paths.

type PutOutcome 

type PutOutcome string

PutOutcome is the machine-readable result of a PutFile call.

It is a distinct type rather than an HTTP status because the caller's response differs per outcome and must not depend on parsing an error string — and because a non-HTTP transport (S3 relay, sneakernet bundle) has to express the same set without inventing status codes. 

const (
	// OutcomeCommitted — the hub verified the checksum and durably stored the
	// file. Advance the ledger to synced.
	OutcomeCommitted PutOutcome = "committed"

	// OutcomeAlreadyPresent — the hub already had this exact content at this
	// path, so the write was a no-op. Treated identically to committed: this
	// is what makes redelivery harmless and turns at-least-once delivery into
	// exactly-once effect.
	OutcomeAlreadyPresent PutOutcome = "already_present"

	// OutcomePartial — the hub accepted a prefix but not the whole file,
	// typically because the link dropped mid-stream. PutResult.BytesAccepted
	// is the new resume checkpoint; the file stays pending.
	OutcomePartial PutOutcome = "partial"

	// OutcomeConflict — same path, different content. The hub refused to
	// overwrite. Do NOT retry: this needs an operator, not a backoff.
	OutcomeConflict PutOutcome = "conflict"

	// OutcomeChecksumMismatch — the bytes arrived corrupted and the hub
	// discarded them without committing. Retrying is correct and worthwhile:
	// the corruption may be in flight, or in the spoke's own storage (edge
	// hardware in the field is exactly where bit-rot shows up).
	OutcomeChecksumMismatch PutOutcome = "checksum_mismatch"

	// OutcomeBackpressure — the hub is overloaded and asked the spoke to slow
	// down. PutResult.RetryAfter carries how long to wait. Not a failure: at
	// fan-in scale with many spokes, this is the hub's normal flow control.
	OutcomeBackpressure PutOutcome = "backpressure"
)

func (PutOutcome) Done 

func (o PutOutcome) Done() bool

Done reports whether the hub now holds the file, so the ledger entry can advance to synced.

func (PutOutcome) Retryable 

func (o PutOutcome) Retryable() bool

Retryable reports whether re-attempting the transfer unchanged could succeed.

Conflict is the one outcome that is emphatically not retryable — resending cannot resolve a content disagreement and would risk overwriting good data. Committed and AlreadyPresent are "not retryable" only because there is nothing left to do.

type PutResult 

type PutResult struct {
	Outcome PutOutcome

	// BytesAccepted is the absolute offset the hub has durably accepted, and
	// becomes the ledger's resume checkpoint. Meaningful for OutcomePartial;
	// equals the file size for a committed transfer.
	BytesAccepted int64

	// RetryAfter is how long the hub asked the spoke to wait. Only set for
	// OutcomeBackpressure.
	RetryAfter time.Duration

	// TheirSHA256 is the hub's digest for the path. Only set for
	// OutcomeConflict, where it is the evidence an operator needs.
	TheirSHA256 string
}

PutResult is the outcome of one PutFile call.

func BackpressureResult 

func BackpressureResult(d time.Duration) *PutResult

BackpressureResult builds the result a hub returns when it wants the spoke to slow down. Provided so tests and future transports produce a valid one (a zero delay would busy-loop).

func (*PutResult) Validate 

func (r *PutResult) Validate(entry *LedgerEntry) error

Validate reports whether the result is internally consistent for its outcome.

A transport is remote code as far as the agent is concerned — a buggy or hostile hub could answer OutcomePartial with a negative offset, or claim backpressure with no delay, and the agent would write nonsense into the ledger. Implementations should call this before returning, and the agent should call it on anything it receives.

type ReceivedFile 

type ReceivedFile struct {
	// SpokeID is the sender. The hub namespaces by it, so this is what keeps
	// two edges writing the same measurement from colliding.
	SpokeID string

	// Path is the final storage-relative path, already namespaced.
	Path string

	// SourcePath is the path as the spoke knows it, before namespacing —
	// carried so logs and manifests can be correlated with the spoke's ledger.
	SourcePath string

	SHA256    string
	SizeBytes int64
}

ReceivedFile describes a file that has been verified and promoted.

func (*ReceivedFile) Database 

func (f *ReceivedFile) Database() string

Database and Measurement extract the Arc namespace a received file belongs to, from the spoke's own path.

Arc's layout is {database}/{measurement}/{y}/{m}/{d}/{h}/file.parquet, and SourcePath is the spoke's path BEFORE hub namespacing — so the database and measurement are its first two segments. Deriving them from SourcePath rather than the namespaced Path is deliberate: the hub prepends the spoke ID, which would otherwise be read as the database name.

Both return "" for a path too short to carry them. validateSyncPath has already rejected traversal and absolute paths by the time these are called.

func (*ReceivedFile) Measurement 

func (f *ReceivedFile) Measurement() string

func (*ReceivedFile) PartitionTime 

func (f *ReceivedFile) PartitionTime() time.Time

PartitionTime is the hour partition the file belongs to, parsed from the spoke's path.

This is not cosmetic metadata. raft.FileEntry.PartitionTime drives hot/cold routing: tiering/router.go filters files by it when a query carries a time range, and tiering/migrator.go computes file age from it to decide what to migrate. A zero value would make every synced file look infinitely old to the migrator and drop it from time-ranged queries entirely.

Returns the zero time when the path does not carry a parseable partition — callers should treat that as "unknown" rather than "epoch".

type ReceivedRecord 

type ReceivedRecord struct {
	SpokeID    string
	SourcePath string // the spoke's own path, before hub namespacing
	HubPath    string // where the hub actually stored it
	SHA256     string
	SizeBytes  int64
	ReceivedAt time.Time
}

ReceivedRecord is one file the hub holds, as the spoke knows it.

type Receiver 

type Receiver struct {
	// contains filtered or unexported fields
}

Receiver implements the hub side of a file transfer: it accepts bytes from a spoke, verifies them, and only then makes them visible.

The ordering is the point. Bytes stream into a staging path while a SHA-256 runs over them; the digest is compared against what the spoke declared; and only on a match is the file promoted into its final namespaced location. A mismatch never produces a byte at the path a reader would look at.

func NewReceiver 

func NewReceiver(cfg ReceiverConfig) (*Receiver, error)

NewReceiver validates configuration and returns a ready Receiver.

func (*Receiver) Receive 

func (r *Receiver) Receive(ctx context.Context, spokeID, sourcePath, declaredSHA256 string, declaredSize, offset int64, body io.Reader) (*PutResult, error)

Receive streams one file from a spoke, verifies it, and promotes it.

offset resumes a previous partial transfer: body must carry only the bytes from offset onward. declaredSHA256 is the digest of the WHOLE file, so a resumed transfer is still verified end to end — the staged prefix is hashed before the tail is appended.

The returned PutResult mirrors what the spoke's transport expects, so an HTTP handler maps it to a status code without re-deriving the semantics.

func (*Receiver) SupportsResume 

func (r *Receiver) SupportsResume() bool

SupportsResume reports whether this hub can accept a partial transfer and continue it later.

Only backends implementing storage.AppendingBackend can — S3 and Azure cannot append to a block object. On those, a dropped transfer restarts from zero rather than resuming, which is a throughput cost on intermittent links, not a correctness problem. Handlers surface this so a spoke does not send an offset the hub cannot honor.

func (*Receiver) SweepStaging 

func (r *Receiver) SweepStaging(ctx context.Context, maxAge time.Duration, now time.Time) (int, error)

SweepStaging deletes abandoned staging files older than maxAge and reports how many it removed.

Without this the staging area grows without bound: a spoke that declares a large file, sends a few bytes, and never returns leaves a partial behind, and nothing else in the system reclaims it. A compromised or merely buggy spoke can repeat that with fresh paths until the hub's disk is full.

maxAge must be comfortably longer than a plausible contact gap, because a staged prefix IS a legitimate resume checkpoint — sweeping too eagerly turns a recoverable transfer into a restart from zero on exactly the link least able to afford it.

Cluster note: this deletes from storage only. Staged files are never in the manifest (they are unverified by definition), so there is no manifest-before- storage ordering to preserve and no Raft proposal to batch.

type ReceiverConfig 

type ReceiverConfig struct {
	Backend storage.Backend
	Logger  zerolog.Logger

	// RecordActivity is called after each committed file. Optional.
	RecordActivity func(ctx context.Context, spokeID string, files, bytes int64)

	// Index records received files for reconcile. Optional but strongly
	// recommended; without it reconcile cannot report anything as present.
	Index *HubIndex

	// RegisterFile is called after a file is verified and promoted. Leave nil
	// in standalone mode.
	RegisterFile func(ctx context.Context, f *ReceivedFile) error
}

ReceiverConfig configures a Receiver.

type ReconcileEntry 

type ReconcileEntry struct {
	Path      string `json:"path"`
	SHA256    string `json:"sha256"`
	SizeBytes int64  `json:"size,omitempty"`
}

ReconcileEntry is one file a spoke is asking about.

type ReconcileResult 

type ReconcileResult struct {
	// Missing lists storage-relative paths the hub does not have. These are
	// what the agent streams, newest-first.
	Missing []string

	// Present lists paths the hub already holds with a matching SHA256. The
	// agent advances these straight to synced without sending a byte.
	//
	// This is the lost-ack recovery path: a transfer that completed but whose
	// acknowledgment never arrived leaves the spoke believing the file is
	// pending. Reconcile discovers the truth in bulk, which is why a lost ack
	// costs one redundant entry in the next batch rather than a re-upload.
	Present []string

	// Conflicts lists paths the hub holds with a DIFFERENT SHA256.
	//
	// This is an alarm, not a retry: it means either two spokes are writing
	// the same namespaced path (a spoke_id collision) or one side's bytes are
	// corrupt. The agent must not resend — overwriting would destroy whichever
	// copy is correct. Surfacing conflicts here catches the whole backlog at
	// once, rather than discovering them one 409 at a time during transfer.
	Conflicts []Conflict
}

ReconcileResult is the hub's answer to a reconcile request: a partition of the spoke's pending set into what must be sent, what is already there, and what disagrees.

func (*ReconcileResult) Validate 

func (r *ReconcileResult) Validate() error

Validate reports whether the reconcile result is internally consistent.

The agent drives ledger state directly from these lists, so a hub that reports the same path as both missing and present would produce contradictory transitions. Checking here keeps that from reaching the ledger.

type ReconcileTooLargeError 

type ReconcileTooLargeError struct {
	MaxEntries int
}

ReconcileTooLargeError reports that the hub refused a reconcile batch with 413. MaxEntries is the hub's advertised entry cap, or 0 when the refusal came from a byte limit that advertises none (the route-level body cap). The agent reacts by splitting the page and retrying, so this error never fails a pass on its own.

func (*ReconcileTooLargeError) Error 

func (e *ReconcileTooLargeError) Error() string

type Reconciler 

type Reconciler struct {
	// contains filtered or unexported fields
}

Reconciler answers "which of these files do you already have?" for a spoke.

func NewReconciler 

func NewReconciler(cfg ReconcilerConfig) (*Reconciler, error)

NewReconciler validates configuration and returns a ready Reconciler.

func (*Reconciler) MaxEntries 

func (r *Reconciler) MaxEntries() int

MaxEntries reports the configured per-batch cap, so a handler can reject an oversized request before decoding it.

func (*Reconciler) Reconcile 

func (r *Reconciler) Reconcile(ctx context.Context, spokeID string, entries []ReconcileEntry) (*ReconcileResult, error)

Reconcile partitions a spoke's pending set into what the hub is missing, what it already holds, and what disagrees.

Answered entirely from the hub index — one batched SQLite lookup, no reads of parquet bytes — which is what makes this affordable for a spoke returning from a long outage. §6.1's identity rule decides each entry: absent is missing, a matching digest is present, and a differing digest is a conflict.

type ReconcilerConfig 

type ReconcilerConfig struct {
	Index *HubIndex

	// Backend confirms a file the index claims still exists in storage.
	//
	// Required, and the reason is data loss rather than tidiness. The index
	// records what the hub RECEIVED; it does not learn about deletions.
	// Anything that removes a file from the hub — Arc retention pointed at a
	// spoke's namespace (its prefix is an operator-chosen string, so
	// `database = "rocket-01"` sweeps that spoke), a cold-tier migration, an
	// operator with rm — leaves the index asserting a file the hub no longer
	// has. Reporting that as `present` makes the spoke mark it synced, and a
	// spoke configured to reclaim space would then delete its only copy.
	//
	// Confirming costs one stat per candidate the index claims (~3µs on local
	// disk, so ~31ms for a 10k batch) and no parquet reads, which keeps §5.1's
	// actual promise: reconcile does not read file contents.
	Backend storage.Backend

	// MaxEntries caps one batch. Zero uses MaxReconcileEntriesDefault.
	MaxEntries int
}

ReconcilerConfig configures a Reconciler.

type Registry 

type Registry struct {
	// contains filtered or unexported fields
}

Registry stores which spokes may sync to this hub, and their secrets.

func NewRegistry 

func NewRegistry(db *sql.DB, cipher SecretCipher, logger zerolog.Logger) (*Registry, error)

NewRegistry creates the registry and initializes its schema.

A cipher is required. §8.1 specifies a `secret_hash` column, but a one-way hash cannot work here: HMAC verification recomputes the MAC from the secret, so the hub needs the plaintext at request time — unlike an API token, which is only ever checked against a presented value. The secret is therefore encrypted at rest rather than hashed, and a hub without a key refuses to start rather than silently storing write credentials in the clear.

func (*Registry) Count 

func (r *Registry) Count(ctx context.Context) (int64, error)

Count returns how many spokes are registered.

Used at startup to tell an operator whether the hub can accept anything. A COUNT rather than len(List(...)) so it does not scan and allocate a row per spoke just to produce a number.

func (*Registry) Delete 

func (r *Registry) Delete(ctx context.Context, spokeID string) error

Delete removes a spoke entirely.

The spoke's received files and index entries are deliberately left alone: deleting a registration must not delete data the hub was trusted with. An operator reclaiming that storage does it explicitly.

func (*Registry) Get 

func (r *Registry) Get(ctx context.Context, spokeID string) (*Spoke, error)

Get returns a spoke's metadata, never its secret.

func (*Registry) List 

func (r *Registry) List(ctx context.Context) ([]*Spoke, error)

List returns every registered spoke, newest first.

func (*Registry) RecordActivity 

func (r *Registry) RecordActivity(ctx context.Context, spokeID string, files, bytes int64) error

RecordActivity updates a spoke's last-seen time and transfer counters.

Best-effort by design: this is observability, and failing a verified, committed transfer because a counter could not be bumped would trade real data for a statistic. Callers log the error and carry on.

func (*Registry) Register 

func (r *Registry) Register(ctx context.Context, spokeID, name string) (secret string, err error)

Register creates a spoke and returns its secret.

The secret is generated here rather than supplied by the caller, and this is the ONLY time it is readable — the operator must capture it from this response to configure the edge box. Generating it removes the failure mode where an operator or an automation picks something weak or reuses one across the fleet.

func (*Registry) RotateSecret 

func (r *Registry) RotateSecret(ctx context.Context, spokeID string) (string, error)

RotateSecret issues a new secret for an existing spoke and returns it once.

The old secret stops working immediately, so an edge box mid-transfer will start failing authentication until it is reconfigured. That is the point of rotation, but it means this is not a routine operation.

func (*Registry) Secret 

func (r *Registry) Secret(ctx context.Context, spokeID string) (string, error)

Secret returns a spoke's decrypted secret for HMAC verification.

Returns ErrSpokeDisabled for a registered-but-disabled spoke, so a caller can log the difference — but callers on the request path MUST NOT surface that distinction to the client, since it would let an attacker enumerate which spoke IDs exist.

func (*Registry) SetEnabled 

func (r *Registry) SetEnabled(ctx context.Context, spokeID string, enabled bool) error

SetEnabled enables or disables a spoke without deleting it.

Disabling is the reversible way to cut a spoke off — its history and byte counters survive, and re-enabling does not require re-provisioning a secret.

func (*Registry) VerifyStoredSecrets 

func (r *Registry) VerifyStoredSecrets(ctx context.Context) (registered int64, err error)

VerifyStoredSecrets checks that the configured key can still decrypt what is already stored, and reports how many spokes are registered.

Called at startup because a changed or lost ARC_ENCRYPTION_KEY is systemic and otherwise invisible: the admin endpoints keep returning 200 with a full spoke list (metadata is not encrypted), so the hub looks healthy while every spoke fails authentication. The first failed sync would reveal it, but that might be a contact window away — and on an edge deployment, a missed window can be hours.

One spoke is enough to prove the key: they are all encrypted under it, so either it works or nothing does. Decrypting the whole fleet would cost a linear scan at every boot for no extra signal.

type RunResult 

type RunResult struct {
	// Discovered is how many local files were newly added to the ledger.
	Discovered int

	// Recovered is how many interrupted transfers were reset to pending.
	//
	// int like its sibling counters, though the ledger reports RowsAffected as
	// int64 — a RunResult is a per-pass summary, and a mixed-width struct is
	// awkward for every caller that formats it.
	Recovered int

	// AlreadyPresent is how many files the hub reported it already had —
	// the lost-acknowledgment path, resolved without sending bytes.
	AlreadyPresent int

	// Sent is how many files were transferred and acknowledged this pass.
	Sent int

	// BytesSent counts only bytes actually put on the wire, so a resumed
	// transfer contributes its tail rather than the whole file.
	BytesSent int64

	// Partial is how many transfers ended mid-file. Not failures: each left a
	// resume checkpoint and continues on the next pass.
	Partial int

	// Failed is how many transfers errored.
	Failed int

	// Skipped is how many entries were dropped because their source file
	// vanished (compaction or retention) before delivery.
	Skipped int

	// Conflicts are same-path-different-content disagreements. These need an
	// operator, not a retry, so they are surfaced rather than counted away.
	Conflicts []Conflict

	Duration time.Duration
}

RunResult summarizes one sync pass.

type SecretCipher 

type SecretCipher interface {
	Encrypt(plaintext string) (string, error)
	Decrypt(ciphertext string) (string, error)
}

SecretCipher encrypts and decrypts spoke secrets at rest.

Satisfied by mqtt.AESEncryptor. Declared here as an interface so this package does not depend on the MQTT one, and so a test can substitute a stub without an encryption key.

type Spoke 

type Spoke struct {
	SpokeID       string
	Name          string
	Enabled       bool
	LastSeenAt    *time.Time
	FilesReceived int64
	BytesReceived int64
	RegisteredAt  time.Time
}

Spoke is a registered edge instance.

The secret is deliberately absent: it is returned exactly once, from Register or RotateSecret, and is never readable again.

type Stats 

type Stats struct {
	HubID    string
	Pending  int64
	InFlight int64

	// Exported — on physical media, awaiting an ack. Counted separately
	// because it is neither "still queued" nor "delivered", and folding it
	// into either would misreport how far behind an air-gap spoke is.
	Exported int64

	Synced int64
	Failed int64

	// Skipped — the source file vanished (compaction/retention) before
	// delivery. Terminal bookkeeping, not backlog: excluded from
	// PendingBytes because there is nothing left to send.
	Skipped int64

	// PendingBytes is what has not reached a hub, INCLUDING exported bytes:
	// a file on a drive in transit has not arrived. Excluding it would make an
	// air-gap spoke's backlog appear to shrink the moment a bundle is written,
	// which is precisely when nothing has been delivered yet.
	PendingBytes int64

	LastSyncedAt *time.Time
}

Stats summarizes ledger state for one hub.

type SyncState 

type SyncState string

SyncState is the lifecycle of one file's journey to one hub. 

const (
	// StatePending — discovered locally, not yet sent. The starting state, and
	// the state an interrupted transfer reverts to on restart.
	StatePending SyncState = "pending"

	// StateInFlight — a transfer is currently running. Any row left in this
	// state after a crash is stale by definition (the transfer died with the
	// process), which is why RecoverInFlight reverts them at startup.
	StateInFlight SyncState = "in_flight"

	// StateSynced — the hub acknowledged receipt. Terminal on the happy path.
	// Only ever set from a 2xx/AlreadyPresent, never optimistically: the
	// design's ack-then-advance rule means a lost ack costs one extra entry in
	// the next reconcile, but never a silent gap.
	StateSynced SyncState = "synced"

	// StateExported — written to an air-gap bundle, awaiting acknowledgment.
	//
	// A file on a physical drive in transit is neither pending (re-exporting or
	// re-sending it wastes the scarce resource: media, or a contact window) nor
	// synced (no hub has confirmed anything). Without this state, exported rows
	// stay pending, and because Pending orders partition_time DESC a capped
	// export takes the newest N every time — the oldest files are never
	// exported at all. That is a treadmill, not eventual consistency.
	//
	// NOT terminal. It advances to synced when an ack bundle returns, and an
	// operator can revert it to pending if the drive is lost.
	StateExported SyncState = "exported"

	// StateFailed — exhausted retries. Terminal until an operator intervenes.
	StateFailed SyncState = "failed"

	// StateSkipped — deliberately not (or no longer) deliverable. Three
	// classes, told apart by last_error: the source file VANISHED before
	// delivery (compaction/retention got it first); a COMPACTED OUTPUT whose
	// contents were already delivered (NoteCompactedOutput — must never
	// sync); or an OPERATOR-DISMISSED failure (NoteOperatorDismissed —
	// reversible via requeue while its file survives). Counted in Stats and
	// /status. Vanished rows prune normally; the two file-backed classes are
	// prune-exempt and reclaimed by SweepSkippedRows once their file is gone.
	StateSkipped SyncState = "skipped"
)

type SyncTransport 

type SyncTransport interface {
	// Reconcile asks the hub which of the spoke's pending files it already
	// holds. This is ONE round-trip for the whole backlog regardless of size —
	// the property that makes a long disconnection survivable, since 5,000
	// pending files cost one request rather than 5,000.
	//
	// The signature materializes both lists, which is fine at phase-1 scale
	// and keeps the interface simple. The WIRE format is where streaming
	// matters: a spoke returning from a months-long outage can present
	// hundreds of thousands of entries (~20MB compressed), so the HTTPS
	// implementation must stream the request and response bodies rather than
	// buffering them, even though the values it produces are materialized.
	// If a deployment ever outgrows an in-memory pending slice, this becomes
	// an iterator — a breaking change deliberately deferred until real.
	Reconcile(ctx context.Context, hubID string, pending []*LedgerEntry) (*ReconcileResult, error)

	// PutFile streams one file's bytes to the hub, resuming from offset.
	//
	// body must yield the file's content starting at that offset; the caller
	// owns opening, seeking, and closing it. offset is absolute within the
	// file, so a resumed transfer sends only the remaining tail.
	//
	// Implementations must read ONLY the content-describing fields of entry —
	// Path, SHA256, SizeBytes, Database, Measurement, PartitionTime. The rest
	// (State, Attempts, BytesSent, SyncedAt, LastError, ID) is spoke-private
	// bookkeeping that no hub has any business seeing, and a transport that
	// branched on it would be reading state a sneakernet bundle or S3 relay
	// cannot meaningfully carry. The whole entry is passed rather than a
	// narrower descriptor to avoid a lossy conversion at every call site.
	//
	// A non-nil error means the transfer did not complete. A nil error with a
	// result whose Outcome is not OutcomeCommitted or OutcomeAlreadyPresent
	// means the hub answered deliberately (partial, conflict, backpressure)
	// and the caller must act on the Outcome rather than assume success.
	PutFile(ctx context.Context, hubID string, entry *LedgerEntry, body io.Reader, offset int64) (*PutResult, error)
}

SyncTransport moves reconcile requests and file bytes from a spoke to a hub.

It exists so the sync agent is written once against an abstract hub and the wire format stays swappable. Phase 1 ships HTTPS; an S3/Azure relay (where spoke and hub never connect directly, exchanging objects through a shared bucket) and a sneakernet bundle (a signed directory carried on physical media) are planned, and both reuse the reconcile/identity/idempotency logic unchanged because none of it is HTTP-specific.

Implementations must be safe for concurrent use: the agent transfers several files at once (§8.2, sync.max_concurrent_files defaults to 2).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.