hub

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 5, 2026 License: MIT Imports: 27 Imported by: 0

Documentation

Overview

Package hub implements the Cloudflare R2 zero-knowledge production backend for the DevStrap sync hub.

The logical DevStrap Hub (HUB-08) has three explicit implementation backends, each behind the one sync.Hub interface defined in internal/sync:

  • file: the file-backed test backend (internal/sync.FileHub), retained ONLY for tests and the --hub-file spike.
  • s3/r2: the direct Cloudflare R2 backend (R2Hub in this package), using S3 credentials and cursor-based polling with backoff. This is the production backend for the single-user fleet.
  • http/sse: a future relay backend (deferred) that adds live push and multi-tenant routing. mTLS and SSE belong only to this later relay, not to the R2-direct path.

R2-direct uses S3 credentials and cursor polling (ListObjectsV2 with start-after) with backoff. It does not require a bespoke server — R2 is a managed object store. The HTTP/SSE relay is revisited only if a transport need (live push, multi-tenant routing) outgrows R2-direct.

FolderHub is the local-folder / cloud-drive-folder hub carrier (AD-1 final slice): a plain shared directory — a Dropbox/iCloud/Google Drive folder, an SMB/NFS mount, anything the OS presents as a filesystem path — carries the same zero-knowledge object set the R2/S3 and git-carrier backends store (enc.v2 event ciphertext, age blobs, sealed snapshots, signed manifests and acks). No bucket, no git remote, no credential plane beyond the drive the user already syncs.

Architecture: like the git carrier, the folder carrier composes the ALREADY-PROVEN R2Hub semantics over the plain-filesystem fsObjectStore, but rooted DIRECTLY in the shared folder — there is no fetch/commit/push loop, because the cloud drive (or network mount) is the replication transport. Each dssync.Hub method is just: acquire the cross-process lock → delegate to R2Hub → release.

Lock/observation placement: the cross-process lock file and the per-clone observation floor (observed.json) live in the LOCAL home cache (~/.devstrap/hub-folder/<hash>/), NEVER inside the shared folder. Replicating lock churn through a cloud drive would cause false contention and "conflicted copy" duplicates, and the observation floor is inherently per-device local state. Only the object payloads and their RFC3339Nano timestamp sidecars (.devstrap-meta/times/, the same freshness mechanism the git carrier uses) live in the shared folder, where they must replicate to converge.

CAS is best-effort across devices. The fsObjectStore conditional-put (PutObjectIfMatch / If-None-Match) is atomic only under the cross-process lock, and that lock lives in each device's LOCAL cache — so it serializes multiple processes of the SAME device but NOT two different devices writing through the same drive. A cloud drive gives no cross-writer linearization point (unlike the git carrier's atomic push-ref CAS or R2's conditional PUT), so a simultaneous retention/sweep-lock race between two devices can in principle produce two "winners"; the drive resolves it as a conflicted copy. This is the documented residual (spec/15), in the same advisory-cooperation class as the sweep lock's byzantine residuals — acceptable because the folder carrier targets the single-user, few-devices, rarely-simultaneous case, and every object is content-addressed or (device,seq)-unique so ordinary convergence never collides.

GitCarrierHub is the zero-infrastructure hub backend (AD-1 first slice): a private git repository — any host the user can already push to — carries the same zero-knowledge object set the R2/S3 backend stores (enc.v2 event ciphertext, age blobs, sealed snapshots, signed manifests/acks). No bucket, no credentials plane beyond the user's existing git auth.

Architecture: rather than re-implementing the 24-method Hub contract, the carrier composes the ALREADY-PROVEN R2Hub semantics over a plain-filesystem S3Client (fsObjectStore) rooted in a local clone of the carrier repo, and adds the git transport around it: 

read  = fetch + reset to the remote head, then delegate to R2Hub
write = fetch + reset, delegate to R2Hub (file mutations), one commit
        for the whole batch, push; on a non-fast-forward rejection
        refetch and re-apply with capped backoff (mutations are
        idempotent: every key is content-addressed or (device,seq)-unique)

The atomic push-ref compare-and-swap is the linearization point that replaces S3 conditional PUT: If-None-Match/If-Match are evaluated against the freshly fetched head inside every attempt, so a lost push race re-evaluates them against the winner's state and surfaces the same ErrSweepLockHeld / ErrRetentionConflict outcomes as R2.

Object LastModified (gc grace windows, sweep-lock TTL) cannot ride git commit times — a dedup re-put changes no bytes and history rewrites reset commit times — so fsObjectStore keeps an RFC3339Nano timestamp sidecar per object under .devstrap-meta/times/. Sidecars live OUTSIDE the workspaces/ prefix, so no Hub listing ever sees them, and they travel with the tree through compaction squashes. The time is client-reported, which is acceptable for the advisory sweep lock's cooperating-clients contract (spec/15: a hostile writer with repo access defeats any dumb carrier).

hub compact is the only history-bounding operation: after deleting cold event objects it rewrites the branch to a single parentless commit of the surviving tree and pushes with --force-with-lease, so the carrier's git history stops growing monotonically (deleting files never shrinks a repo); the host garbage-collects the unreachable objects. Concurrent pushers recover through their own fetch-and-reapply loop. The caller (hub compact) already holds the advisory sweep lock.

R2 is S3-compatible with zero egress, strong consistency for object writes/listing, and conditional puts. Event-log payloads are envelope-encrypted (XChaCha20-Poly1305 under a per-epoch Workspace Content Key, P4-SEC-02/SEC-07) and Ed25519-signed before upload; blobs are age-encrypted. R2 stores only ciphertext plus a signed carrier map — it can decrypt nothing and holds no private key.

The event log is NOT one overwritten manifest object. Every event is an immutable, unique, lexicographically sortable object (HUB-06). Since P5-SYNC-01 the transport coordinate is the per-origin-device sequence number, so Push writes per-device, seq-ordered keys: 

workspaces/<workspace_id>/eventlog/<device_id>/<seq-padded>_<event_id>.json

and Pull resumes each device's stream with a per-device Seq cursor (delimiter-discovered device prefixes + StartAfter within each). The retired HLC-keyed legacy layout 

workspaces/<workspace_id>/events/<hlc-padded>/<device_id>/<seq>/<event_id>.json

is still READ (dual-read) so pre-migration hubs keep working — Pull parses (device, seq) out of legacy keys and applies the same per-device cursor; unparseable legacy keys fail open toward fetching so a parse bug can never silently lose events. Push never writes the legacy layout. A follow-up `hub migrate-events` can re-key legacy objects and delete the old prefix; the dual-read is O(1) on an empty prefix.

Blobs are content-addressed (HUB-06): 

workspaces/<workspace_id>/blobs/<sha256>

The S3 operations are abstracted behind the S3Client interface so the keying scheme and Hub contract are unit-testable with an in-memory double (the conformance suite). A real implementation plugs in the AWS SDK v2 S3 client.

s3client_awssdk.go is the production S3Client adapter (P5-HUB-01). It wraps the aws-sdk-go-v2 S3 client pointed at a Cloudflare R2 (or any S3-compatible) endpoint and translates SDK errors into the four hub sentinels (ErrPreconditionFailed, ErrS3Throttle, ErrS3Transient, dssync.ErrBlobNotFound) so R2Hub.Retry is the single retry/classification layer.

It is constructed with s3.New(s3.Options{...}) — not config.LoadDefaultConfig — to keep the dependency/govulncheck surface to the S3 data path only (no SSO/IMDS/STS chain). The SDK's own retryer is disabled (aws.NopRetryer{}) so R2Hub.Retry is the ONLY retry layer and there is no double-retry/billing loop. Credentials are supplied inline via aws.CredentialsProviderFunc so the `credentials` module is never pulled in.

Index

Constants

This section is empty.

Variables

View Source 
var ErrNotImplemented = errors.New("s3 operation not implemented")

ErrNotImplemented signals an S3Client method that has no production wiring yet.

View Source 
var ErrPreconditionFailed = errors.New("object already exists (conditional put failed)")

ErrPreconditionFailed signals that a conditional PutObject (If-None-Match: *) was rejected because the object already exists (R2 error 10031 / HTTP 412). For content-addressed blobs and immutable event keys a 412 is definitionally an idempotent dedup hit, not an error (HUB-09): the same sha256 yields the same ciphertext, and a duplicate event key is the same event.

View Source 
var ErrS3Auth = errors.New("s3 authentication/authorization failed")

ErrS3Auth signals a credential/authorization failure (401/403, SignatureDoesNotMatch, InvalidAccessKeyId, AccessDenied). Terminal — never retried — and carries remediation guidance so a bad or unresolved secret (e.g. an op:// ref pasted where a literal was expected before P6-HUB-02) surfaces as an actionable message instead of a raw SDK error.

View Source 
var ErrS3Throttle = errors.New("s3 throttling (429/503 slow down)")

ErrS3Throttle signals an R2/S3 throttling response (429 TooManyRequests / 503 SlowDown) that is retryable after backoff (HUB-10).

View Source 
var ErrS3Transient = errors.New("s3 transient (500/connection reset)")

ErrS3Transient signals an R2/S3 transient response (InternalError / connection reset) that is retryable after a short backoff (HUB-10).

Functions

This section is empty.

Types

type FolderHub 

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

FolderHub implements dssync.Hub over a shared directory. Construct with NewFolderHub. Safe for concurrent use: every operation serializes on an in-process mutex plus a cross-process lock file (both keyed to this device's local cache, so they serialize same-device processes; cross-device ordering is left to the drive — see the package comment).

func NewFolderHub 

func NewFolderHub(dir, workspaceID, cacheRoot string) (*FolderHub, error)

NewFolderHub prepares a folder-carrier hub rooted at dir (which must be an absolute path). cacheRoot is the local cache PARENT (one subdirectory per resolved folder, so two folders never share a lock or observation floor); the lock file and observed.json live under it, never inside the shared folder. The folder is created (0700) when missing and its symlinks are resolved once — cloud-drive roots are frequently symlinks — so the store, the lock, and the cache hash all key on the real path.

func (*FolderHub) CompactEventsBelow 

func (f *FolderHub) CompactEventsBelow(ctx context.Context, floors dssync.Cursor) (int, error)

CompactEventsBelow deletes cold event objects. Unlike the git carrier there is no history to squash — file deletion IS the reclamation on a plain filesystem — so it is a single locked delegation to R2Hub. The caller holds the advisory sweep lock.

func (*FolderHub) DeleteAck 

func (f *FolderHub) DeleteAck(ctx context.Context, deviceID string) error

func (*FolderHub) DeleteBlob 

func (f *FolderHub) DeleteBlob(ctx context.Context, sha256Hex string) error

func (*FolderHub) DeleteDeviceStream 

func (f *FolderHub) DeleteDeviceStream(ctx context.Context, deviceID string) (int, error)

func (*FolderHub) DeleteSnapshotObject 

func (f *FolderHub) DeleteSnapshotObject(ctx context.Context, sha256Hex string) error

func (*FolderHub) DeleteSweepLock 

func (f *FolderHub) DeleteSweepLock(ctx context.Context) error

func (*FolderHub) GetBlob 

func (f *FolderHub) GetBlob(ctx context.Context, sha256Hex string) (io.ReadCloser, error)

func (*FolderHub) GetRetention 

func (f *FolderHub) GetRetention(ctx context.Context) ([]byte, string, error)

func (*FolderHub) GetSnapshotObject 

func (f *FolderHub) GetSnapshotObject(ctx context.Context, sha256Hex string) ([]byte, error)

func (*FolderHub) GetSweepLock 

func (f *FolderHub) GetSweepLock(ctx context.Context) ([]byte, time.Time, error)

func (*FolderHub) HasEvents 

func (f *FolderHub) HasEvents(ctx context.Context) (bool, error)

HasEvents reports whether any event was ever recorded on this hub (the doctor --remote capability probe, mirroring GitCarrierHub.HasEvents).

func (*FolderHub) ListAcks 

func (f *FolderHub) ListAcks(ctx context.Context) (map[string][]byte, error)

func (*FolderHub) ListBlobs 

func (f *FolderHub) ListBlobs(ctx context.Context) ([]dssync.BlobInfo, error)

func (*FolderHub) ListSnapshotObjects 

func (f *FolderHub) ListSnapshotObjects(ctx context.Context) ([]dssync.BlobInfo, error)

func (*FolderHub) MigrateLegacyEvents 

func (f *FolderHub) MigrateLegacyEvents(ctx context.Context, dryRun bool) (int, int, error)

MigrateLegacyEvents delegates for symmetric reporting; the folder carrier, like the git carrier, never had the retired HLC-keyed layout, so it is a structural no-op. Both the dry run and the real run only read/rewrite object files, so a single lock covers either.

func (*FolderHub) Pull 

func (f *FolderHub) Pull(ctx context.Context, after dssync.Cursor) ([]state.Event, error)

func (*FolderHub) Push 

func (f *FolderHub) Push(ctx context.Context, events []state.Event) error

func (*FolderHub) PutAck 

func (f *FolderHub) PutAck(ctx context.Context, deviceID string, raw []byte) error

func (*FolderHub) PutBlob 

func (f *FolderHub) PutBlob(ctx context.Context, sha256Hex string, r io.Reader) error

func (*FolderHub) PutRetention 

func (f *FolderHub) PutRetention(ctx context.Context, raw []byte, ifMatchETag string) error

func (*FolderHub) PutSnapshotObject 

func (f *FolderHub) PutSnapshotObject(ctx context.Context, sha256Hex string, body []byte) error

func (*FolderHub) PutSweepLock 

func (f *FolderHub) PutSweepLock(ctx context.Context, raw []byte) error

func (*FolderHub) StatBlob 

func (f *FolderHub) StatBlob(ctx context.Context, sha256Hex string) (dssync.BlobInfo, error)

type GitCarrierHub 

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

GitCarrierHub implements dssync.Hub over a private git repository. Construct with NewGitCarrierHub. Safe for concurrent use; all operations serialize on an in-process mutex plus a cross-process lock file, because they share one working clone.

func NewGitCarrierHub 

func NewGitCarrierHub(remote, branch, workspaceID, cacheRoot string) (*GitCarrierHub, error)

NewGitCarrierHub prepares a git-carrier hub for remote/branch. cacheRoot is the local clone cache directory (one subdirectory per remote+branch, so two hubs never share a checkout); it is created on demand. The remote must pass git.ValidateRemote; the branch must be a safe git branch name.

func (*GitCarrierHub) CompactEventsBelow 

func (g *GitCarrierHub) CompactEventsBelow(ctx context.Context, floors dssync.Cursor) (int, error)

CompactEventsBelow deletes cold event objects like R2, then — because file deletion never shrinks a git repository — rewrites the branch to a single parentless commit of the surviving tree and pushes it with --force-with-lease against the head this pass fetched. The caller holds the advisory sweep lock; a concurrent pusher that loses the lease race simply refetches the squashed head and re-applies (its mutations are idempotent). A lost lease HERE refetches and re-runs the deletion against the new head.

func (*GitCarrierHub) DeleteAck 

func (g *GitCarrierHub) DeleteAck(ctx context.Context, deviceID string) error

func (*GitCarrierHub) DeleteBlob 

func (g *GitCarrierHub) DeleteBlob(ctx context.Context, sha256Hex string) error

func (*GitCarrierHub) DeleteDeviceStream 

func (g *GitCarrierHub) DeleteDeviceStream(ctx context.Context, deviceID string) (int, error)

func (*GitCarrierHub) DeleteSnapshotObject 

func (g *GitCarrierHub) DeleteSnapshotObject(ctx context.Context, sha256Hex string) error

func (*GitCarrierHub) DeleteSweepLock 

func (g *GitCarrierHub) DeleteSweepLock(ctx context.Context) error

func (*GitCarrierHub) GetBlob 

func (g *GitCarrierHub) GetBlob(ctx context.Context, sha256Hex string) (io.ReadCloser, error)

func (*GitCarrierHub) GetRetention 

func (g *GitCarrierHub) GetRetention(ctx context.Context) ([]byte, string, error)

func (*GitCarrierHub) GetSnapshotObject 

func (g *GitCarrierHub) GetSnapshotObject(ctx context.Context, sha256Hex string) ([]byte, error)

func (*GitCarrierHub) GetSweepLock 

func (g *GitCarrierHub) GetSweepLock(ctx context.Context) ([]byte, time.Time, error)

func (*GitCarrierHub) HasEvents 

func (g *GitCarrierHub) HasEvents(ctx context.Context) (bool, error)

HasEvents reports whether any event was ever recorded on this hub (the doctor --remote capability probe, mirroring FileHub.HasEvents).

func (*GitCarrierHub) ListAcks 

func (g *GitCarrierHub) ListAcks(ctx context.Context) (map[string][]byte, error)

func (*GitCarrierHub) ListBlobs 

func (g *GitCarrierHub) ListBlobs(ctx context.Context) ([]dssync.BlobInfo, error)

func (*GitCarrierHub) ListSnapshotObjects 

func (g *GitCarrierHub) ListSnapshotObjects(ctx context.Context) ([]dssync.BlobInfo, error)

func (*GitCarrierHub) MigrateLegacyEvents 

func (g *GitCarrierHub) MigrateLegacyEvents(ctx context.Context, dryRun bool) (int, int, error)

MigrateLegacyEvents is a structural no-op: the git carrier never had the retired HLC-keyed layout. Delegated for symmetric reporting. A dry run goes through the READ path — the write loop would seed the carrier marker (and, on an empty carrier, create the branch), violating the CLI's report-without-writing dry-run contract.

func (*GitCarrierHub) Pull 

func (g *GitCarrierHub) Pull(ctx context.Context, after dssync.Cursor) ([]state.Event, error)

func (*GitCarrierHub) Push 

func (g *GitCarrierHub) Push(ctx context.Context, events []state.Event) error

func (*GitCarrierHub) PutAck 

func (g *GitCarrierHub) PutAck(ctx context.Context, deviceID string, raw []byte) error

func (*GitCarrierHub) PutBlob 

func (g *GitCarrierHub) PutBlob(ctx context.Context, sha256Hex string, r io.Reader) error

func (*GitCarrierHub) PutRetention 

func (g *GitCarrierHub) PutRetention(ctx context.Context, raw []byte, ifMatchETag string) error

func (*GitCarrierHub) PutSnapshotObject 

func (g *GitCarrierHub) PutSnapshotObject(ctx context.Context, sha256Hex string, body []byte) error

func (*GitCarrierHub) PutSweepLock 

func (g *GitCarrierHub) PutSweepLock(ctx context.Context, raw []byte) error

func (*GitCarrierHub) StatBlob 

func (g *GitCarrierHub) StatBlob(ctx context.Context, sha256Hex string) (dssync.BlobInfo, error)

type R2Config 

type R2Config struct {
	Endpoint    string // R2 S3 API endpoint (https://<account>.r2.cloudflarestorage.com)
	Bucket      string
	WorkspaceID string
	// CredentialMode is "self-hosted" (bucket-scoped key) or "hosted"
	// (temporary prefix-scoped credentials brokered by the control plane).
	CredentialMode string
	// PrefixScope restricts operations to workspaces/<workspace_id>/. Required
	// in hosted mode so a scoped key cannot access other tenants.
	PrefixScope string
}

R2Config configures the Cloudflare R2 backend credentials and endpoint (HUB-07). Two credential modes are supported:

  • Self-hosted (single owner): a bucket-scoped R2 API token may be used directly. The key is kept on the trusted device and never reaches runners.
  • Hosted/SaaS: the parent R2 key stays only in trusted control-plane code. Devices and runner Machines receive short-lived temporary credentials or presigned URLs scoped to workspaces/<workspace_id>/... with the minimum needed operations. Runner Machines never receive the parent key.

PrefixScope, when set, restricts all operations to a workspace prefix so a scoped credential cannot touch another tenant's objects.

type R2Hub 

type R2Hub struct {
	S3          S3Client
	WorkspaceID string
	// RetentionSeqs is the hub's per-device retention horizon (P5-HUB-03,
	// re-based on the P5-SYNC-01 Seq cursor): for each origin device, the
	// minimum Seq still retained on the hub. A Pull whose cursor would leave a
	// gap below a device's floor (after[dev]+1 < RetentionSeqs[dev]) returns
	// dssync.ErrSnapshotRequired so the caller performs a full-state snapshot
	// exchange instead of silently receiving a partial (post-compaction) event
	// set and diverging. Empty means "no compaction yet" (R2 retains
	// everything). The future compaction marker must be recorded per device
	// (e.g. eventlog/<device>/.retention).
	RetentionSeqs map[string]int64
	// Retry configures R2Hub-level retry, backoff, and error classification for
	// S3 operations (HUB-10). A zero value uses a default policy: throttling
	// (429/503 SlowDown) and transient (500/connection-reset) errors are retried
	// with capped exponential backoff plus full jitter; terminal errors (auth,
	// precondition, not-found, malformed) fail fast. A real aws-sdk-go-v2
	// client wires its own standard retryer; this seam works with any S3Client
	// (including the in-memory conformance double) and is exercised via fault
	// injection before the SDK is wired.
	Retry R2Retry
}

R2Hub is the Cloudflare R2 zero-knowledge Hub backend (HUB-02). It implements dssync.Hub. All content is client-side encrypted before it reaches S3Client.

func (R2Hub) CompactEventsBelow 

func (h R2Hub) CompactEventsBelow(ctx context.Context, floors dssync.Cursor) (int, error)

CompactEventsBelow deletes event objects strictly below each device's floor (Seq < floors[dev]) in both layouts. The caller must have durably published the superseding snapshot + manifest first (confirm-before-delete). In the seq-keyed layout the per-device prefix list is bounded with StartAfter-free enumeration from the stream head, stopping at the floor key; in the legacy layout parsed (device, seq) keys below the floor are deleted and unparseable keys are KEPT (fail safe — mirroring the dual-read's fail-open posture, a parse bug must never delete an event it cannot account for).

func (R2Hub) DeleteAck 

func (h R2Hub) DeleteAck(ctx context.Context, deviceID string) error

DeleteAck removes a device's sync-ack marker (idempotent).

func (R2Hub) DeleteBlob 

func (h R2Hub) DeleteBlob(ctx context.Context, sha256Hex string) error

DeleteBlob removes a content-addressed blob from the hub (SEC-01/HUB-12). A missing blob is not an error (idempotent delete). On revoke, after a blob is rewrapped to the reduced recipient set and its references repointed, the old ciphertext is deleted so the revoked device can no longer fetch it.

func (R2Hub) DeleteDeviceStream 

func (h R2Hub) DeleteDeviceStream(ctx context.Context, deviceID string) (int, error)

DeleteDeviceStream deletes every object under one origin device's event-log prefix (idempotent), reclaiming a revoked device's stream after compaction has folded its state into the snapshot. It returns the object count deleted.

func (R2Hub) DeleteSnapshotObject 

func (h R2Hub) DeleteSnapshotObject(ctx context.Context, sha256Hex string) error

DeleteSnapshotObject removes a superseded snapshot object (idempotent).

func (R2Hub) DeleteSweepLock 

func (h R2Hub) DeleteSweepLock(ctx context.Context) error

DeleteSweepLock removes the sweep-lock object (idempotent).

func (R2Hub) GetBlob 

func (h R2Hub) GetBlob(ctx context.Context, sha256Hex string) (io.ReadCloser, error)

func (R2Hub) GetRetention 

func (h R2Hub) GetRetention(ctx context.Context) ([]byte, string, error)

GetRetention returns the raw retention-manifest bytes plus the object's ETag for CAS (P4-SYNC-02/P6-HUB-04). Absent manifest is ErrRetentionNotFound.

func (R2Hub) GetSnapshotObject 

func (h R2Hub) GetSnapshotObject(ctx context.Context, sha256Hex string) ([]byte, error)

GetSnapshotObject returns a sealed snapshot object's bytes. A missing object wraps dssync.ErrBlobNotFound.

func (R2Hub) GetSweepLock 

func (h R2Hub) GetSweepLock(ctx context.Context) ([]byte, time.Time, error)

GetSweepLock reads the advisory sweep-lock object plus the object's LastModified (from a single-key list) for TTL judgment (P4-HUB-12). A missing lock is dssync.ErrSweepLockNotFound.

func (R2Hub) HasEvents 

func (h R2Hub) HasEvents(ctx context.Context) (bool, error)

HasEvents reports whether this workspace has any events at all on the hub (P4-SEC-07 doctor mismatch check): one retried MaxKeys=1 ListObjectsV2 call per layout (seq-keyed, then legacy). It answers "is this prefix populated" cheaply, without paging the whole event log.

func (R2Hub) ListAcks 

func (h R2Hub) ListAcks(ctx context.Context) (map[string][]byte, error)

ListAcks returns every device's sync-ack marker keyed by device id.

func (R2Hub) ListBlobs 

func (h R2Hub) ListBlobs(ctx context.Context) ([]dssync.BlobInfo, error)

ListBlobs returns metadata for every blob in this workspace's blob prefix (P5-HUB-02), the enumeration primitive for mark-and-sweep hub GC.

func (R2Hub) ListSnapshotObjects 

func (h R2Hub) ListSnapshotObjects(ctx context.Context) ([]dssync.BlobInfo, error)

ListSnapshotObjects returns metadata for every snapshot object in this workspace's snapshots prefix (compaction prunes superseded ones by age).

func (R2Hub) MigrateLegacyEvents 

func (h R2Hub) MigrateLegacyEvents(ctx context.Context, dryRun bool) (migrated, kept int, err error)

MigrateLegacyEvents re-keys the retired HLC-keyed legacy layout 

workspaces/<ws>/events/<hlc pad20>/<device>/<seq>/<id>.json

into the per-device seq layout and deletes the migrated legacy objects (P4-HUB-12). It is idempotent and resumable — the dual-read keeps unmigrated objects live, so a partial run leaves a correct superset — and FAILS OPEN: a key that does not parse, a body that does not decode as a state.Event, or a body whose (DeviceID, Seq) disagree with the key coordinates is reported and KEPT (never deleted), mirroring Pull's fail-open dual-read. Each object is verified by read-back on the new key before the legacy object is deleted, so a mid-migration crash or a backend that returns wrong bytes never loses an event. Re-running against a fully migrated hub reports (0, 0, nil). A dryRun classifies every object (would-migrate vs would-keep) but writes and deletes NOTHING.

func (R2Hub) Pull 

func (h R2Hub) Pull(ctx context.Context, after dssync.Cursor) ([]state.Event, error)

func (R2Hub) Push 

func (h R2Hub) Push(ctx context.Context, events []state.Event) error

func (R2Hub) PutAck 

func (h R2Hub) PutAck(ctx context.Context, deviceID string, raw []byte) error

PutAck writes one device's signed sync-ack marker (P4-SYNC-06). Single writer per key (each device writes only its own ack), so an unconditional PUT is last-writer-wins by design.

func (R2Hub) PutBlob 

func (h R2Hub) PutBlob(ctx context.Context, sha256Hex string, r io.Reader) error

func (R2Hub) PutRetention 

func (h R2Hub) PutRetention(ctx context.Context, raw []byte, ifMatchETag string) error

PutRetention writes the retention manifest with compare-and-swap semantics: ifMatchETag "" is create-only (If-None-Match: *); otherwise If-Match. A lost race in either mode is dssync.ErrRetentionConflict — the caller must re-read, re-derive floors, and retry or refuse.

A conditional PUT retried after an ambiguous failure (the first attempt committed but the response was lost to a transient network/5xx error) would 412 against its OWN success — the object's ETag already changed. So a 412 is disambiguated by a read-back: if the current manifest bytes equal what this call was writing, the write already happened and this is success, not a lost race (post-#65 Codex review, P2).

func (R2Hub) PutSnapshotObject 

func (h R2Hub) PutSnapshotObject(ctx context.Context, sha256Hex string, body []byte) error

PutSnapshotObject stores a sealed snapshot object. Content-addressed: a 412 on the conditional put is definitionally a dedup hit (same sha256 = same bytes), so concurrent compactors producing identical snapshots cannot clobber or fail each other.

func (R2Hub) PutSweepLock 

func (h R2Hub) PutSweepLock(ctx context.Context, raw []byte) error

PutSweepLock writes the sweep-lock object create-only (If-None-Match:*): an existing lock surfaces as dssync.ErrSweepLockHeld (P4-HUB-12).

func (R2Hub) StatBlob 

func (h R2Hub) StatBlob(ctx context.Context, sha256Hex string) (dssync.BlobInfo, error)

StatBlob returns one blob's current metadata via HEAD (P4-HUB-12), the pre-delete revalidation primitive for hub GC. A missing blob wraps dssync.ErrBlobNotFound.

type R2Retry 

type R2Retry struct {
	MaxAttempts   int           // total attempts including the first; 0 => default (3)
	BaseDelay     time.Duration // base delay for transient errors; 0 => 50ms
	ThrottleDelay time.Duration // base delay for throttling errors; 0 => 1s
	Cap           time.Duration // max backoff delay; 0 => 20s
	// Jitter returns a non-negative int64 in [0, n). Defaults to math/rand.Int63n.
	// Tests inject a deterministic source (e.g. always 0) for fast retries.
	Jitter func(n int64) int64
}

R2Retry configures R2Hub-level retry behavior (HUB-10): throttling and transient S3 errors are retried with capped exponential backoff plus full jitter; terminal errors fail fast. A real aws-sdk-go-v2 client wires its own standard retryer (retry.NewStandard + a token-bucket RateLimiter so retries cannot create a runaway billing loop); this seam is the R2Hub-level policy that works with any S3Client and is tested via fault injection.

type S3Adapter 

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

S3Adapter is the production hub.S3Client backed by aws-sdk-go-v2 (P5-HUB-01). It points at one bucket; workspace key-prefix scoping is handled by R2Hub's keying (workspaces/<workspace_id>/...), so the adapter itself is prefix-free.

func NewS3Client 

func NewS3Client(endpoint, region, bucket, accessKeyID, secretAccessKey string) (*S3Adapter, error)

NewS3Client builds a production S3Client against an R2/S3-compatible endpoint (P5-HUB-01). endpoint is the S3 API URL (e.g. https://<account>.r2.cloudflarestorage.com); region defaults to "auto" (R2); bucket is the target bucket. accessKeyID/secretAccessKey are the bucket-scoped static credentials (self-hosted mode). The SDK retryer is disabled so R2Hub.Retry is the single retry layer.

func (*S3Adapter) DeleteObject 

func (a *S3Adapter) DeleteObject(ctx context.Context, key string) error

DeleteObject removes the object at key. A missing object is not an error (idempotent delete) so blob/event GC (HUB-12) and revoke cleanup (SEC-01) can call it unconditionally for superseded ciphertext.

func (*S3Adapter) GetObject 

func (a *S3Adapter) GetObject(ctx context.Context, key string) (data []byte, err error)

GetObject returns the object bytes at key, or an error wrapping dssync.ErrBlobNotFound when the object is absent (HUB-02).

func (*S3Adapter) GetObjectWithETag 

func (a *S3Adapter) GetObjectWithETag(ctx context.Context, key string) (data []byte, etag string, err error)

GetObjectWithETag returns the object bytes at key plus the object's ETag, for compare-and-swap read-modify-write of the retention manifest (P4-SYNC-02/P6-HUB-04).

func (*S3Adapter) ListCommonPrefixes 

func (a *S3Adapter) ListCommonPrefixes(ctx context.Context, prefix, delimiter string) ([]string, error)

ListCommonPrefixes returns the distinct sub-prefixes directly under prefix, grouped at delimiter (P5-SYNC-01 device-stream discovery). Pagination uses the S3 continuation token, NOT the start-after-last-prefix trick: resuming after a common prefix would re-list that prefix's own keys (they sort after the bare prefix) and return it again — duplicate device streams past 1000 devices (post-#59 opus review, Minor).

func (*S3Adapter) ListObjectsV2 

func (a *S3Adapter) ListObjectsV2(ctx context.Context, prefix, startAfter string, maxKeys int) ([]dssync.BlobInfo, string, error)

ListObjectsV2 returns objects under prefix, lexicographically after startAfter, up to maxKeys. When truncated, it returns the last key of the page as nextStartAfter (the memS3 start-after contract — NOT the S3 continuation token) so R2Hub.Pull/ListBlobs page with the same semantics as the in-memory conformance double (HUB-06).

func (*S3Adapter) ObjectExists 

func (a *S3Adapter) ObjectExists(ctx context.Context, key string) (bool, error)

ObjectExists reports whether an object exists at key via HEAD (HUB-02). A missing object is (false, nil), not an error.

func (*S3Adapter) PutObject 

func (a *S3Adapter) PutObject(ctx context.Context, key string, body []byte, ifNoneMatch bool) error

PutObject stores body at key. When ifNoneMatch is true the put is conditional on the object not already existing (If-None-Match: *), making event append and content-addressed blob put idempotent (HUB-06/HUB-09). A collision surfaces as ErrPreconditionFailed, which R2Hub classifies as a dedup no-op.

func (*S3Adapter) PutObjectIfMatch 

func (a *S3Adapter) PutObjectIfMatch(ctx context.Context, key string, body []byte, etag string) error

PutObjectIfMatch stores body at key conditionally on the current object still carrying etag (If-Match — an S3 extension R2 supports on PUT). A lost CAS race surfaces as ErrPreconditionFailed via mapS3Error.

func (*S3Adapter) StatObject 

func (a *S3Adapter) StatObject(ctx context.Context, key string) (dssync.BlobInfo, error)

StatObject returns one object's metadata (Key + LastModified) via HEAD, for hub GC's pre-delete revalidation (P4-HUB-12). A missing object wraps dssync.ErrBlobNotFound (via mapS3Error, which maps *types.NotFound / 404).

type S3Client 

type S3Client interface {
	// PutObject stores data at key. When ifNoneMatch is true the put is
	// conditional on the object not already existing (If-None-Match: *),
	// making event append idempotent (HUB-06).
	PutObject(ctx context.Context, key string, body []byte, ifNoneMatch bool) error
	// GetObject returns the object at key, or an error wrapping
	// dssync.ErrBlobNotFound when absent.
	GetObject(ctx context.Context, key string) ([]byte, error)
	// ObjectExists reports whether an object exists at key.
	ObjectExists(ctx context.Context, key string) (bool, error)
	// DeleteObject removes the object at key. A missing object is not an error
	// (idempotent delete) so blob/event GC (HUB-12) and revoke cleanup (SEC-01)
	// can call it unconditionally for superseded ciphertext.
	DeleteObject(ctx context.Context, key string) error
	// ListObjectsV2 returns objects under prefix, lexicographically after
	// startAfter, up to maxKeys. BlobInfo is reused as a generic key+time pair;
	// for event objects, Key is the full trimmed key as before. When truncated,
	// it returns the next key to continue from.
	ListObjectsV2(ctx context.Context, prefix, startAfter string, maxKeys int) (objs []dssync.BlobInfo, nextStartAfter string, err error)
	// ListCommonPrefixes returns the distinct sub-prefixes directly under
	// prefix, grouped at the given delimiter (ListObjectsV2 CommonPrefixes).
	// P5-SYNC-01 uses it to discover origin-device streams under the eventlog/
	// prefix without listing every object.
	ListCommonPrefixes(ctx context.Context, prefix, delimiter string) ([]string, error)
	// GetObjectWithETag returns the object at key plus its ETag, for
	// compare-and-swap read-modify-write of head objects (the retention
	// manifest, P4-SYNC-02/P6-HUB-04). A missing object wraps
	// dssync.ErrBlobNotFound.
	GetObjectWithETag(ctx context.Context, key string) (data []byte, etag string, err error)
	// PutObjectIfMatch stores data at key conditionally on the current object
	// still carrying etag (If-Match). A lost race surfaces as
	// ErrPreconditionFailed. R2 supports If-Match on PUT as an S3 extension.
	PutObjectIfMatch(ctx context.Context, key string, body []byte, etag string) error
	// StatObject returns one object's metadata (Key + LastModified) via HEAD,
	// for hub GC's pre-delete revalidation (P4-HUB-12). A missing object wraps
	// dssync.ErrBlobNotFound.
	StatObject(ctx context.Context, key string) (dssync.BlobInfo, error)
}

S3Client is the minimal S3-compatible operation set the R2 backend needs (HUB-02). It is abstracted so the keying scheme and Hub contract are testable with an in-memory double. A production implementation wraps the AWS SDK v2 S3 client pointed at an R2 endpoint.

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.