blob

package
v0.116.3 Latest Latest
Warning

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

 Go to latest Published: Jul 9, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package blob is the native blob store described in docs/multi-device-storage.md §2.4 and §4.2. v1 phase 3 slice 1 ships a filesystem-only implementation: Get / Head / Put / Delete / List with atomic publish (temp + fsync + rename + parent dir fsync) plus content digest verification on Put. The blob_refs DB cache and the HTTP transport are deliberately not wired here — slice 2 (DB integration) and slice 3 (HTTP handler) layer on top of this package.

Index

Constants

View Source 
const DefaultScrubInterval = 24 * time.Hour

DefaultScrubInterval is how often Scrubber.loop runs ScrubOnce. 24 h: scrub is heavy (re-hashes every blob body) and the failure modes it catches (silent bitrot, partial-write torn pages) are rare enough that hourly cadence would waste CPU. Tests pass a shorter interval via opts.

Variables

View Source 
var ErrDurabilityDegraded = errors.New("blob: durability degraded (fsync dir failed)")

ErrDurabilityDegraded signals that blob.Store.Put committed the body to disk AND the blob_refs row, but the parent directory fsync failed — a power-loss within seconds of the call could resurrect the prior leaf in the dir entry. The body/row pair is internally consistent, so a caller orchestrating a §3.7 device-switch can treat the write as logically successful (don't abort the handoff) while still surfacing the degradation to operators / metrics. errors.Is the sentinel out of the wrap chain to distinguish from a hard-failure error.

View Source 
var ErrETagMismatch = errors.New("blob: etag mismatch")

ErrETagMismatch is returned by Put / Delete when an If-Match was supplied and the current etag does not match. Mirrors the store's ErrETagMismatch so callers get a consistent type across DB and blob optimistic concurrency.

View Source 
var ErrExpectedSHA256Mismatch = errors.New("blob: sha256 mismatch")

ErrExpectedSHA256Mismatch is returned by Put when PutOptions. ExpectedSHA256 was set and the streamed body hashed to a different digest. atomicWrite wraps this with the actual / expected pair via fmt.Errorf("...: %w", ...) so callers can `errors.Is` cleanly while still getting an informative message in logs.

View Source 
var ErrHandoffPending = errors.New("blob: row is mid-handoff (handoff_pending=1)")

ErrHandoffPending mirrors store.ErrHandoffPending: a device-switch (§3.7) has marked this blob's row as mid-handoff and runtime writes that would actually change the body are refused. The underlying ref store returns the sentinel; blob.Store.Put re-exports it under this package so callers can errors.Is without importing internal/store.

View Source 
var ErrInvalidPath = errors.New("blob: invalid path")

ErrInvalidPath signals a path that violates the blob namespace rules (traversal, absolute, reserved name, reserved suffix). Callers that surface to HTTP map this to 400.

View Source 
var ErrInvalidScope = errors.New("blob: invalid scope")

ErrInvalidScope is returned when a caller passes a Scope outside the three valid values.

View Source 
var ErrNotFound = errors.New("blob: not found")

ErrNotFound is returned by Get / Head / Delete / Verify when no blob exists at the requested (scope, path).

View Source 
var ErrRefNotFound = errors.New("blob: ref not found")

ErrRefNotFound mirrors store.ErrNotFound; refs implementations MUST return this exact value (or wrap it via fmt.Errorf with %w) so the blob layer can sniff it via errors.Is regardless of the concrete store backend.

View Source 
var ErrScopeContainmentBroken = errors.New("blob: scope containment broken")

ErrScopeContainmentBroken signals a runtime defect in the on-disk blob tree: the scope dir or one of its parent components is a symlink, or an intermediate component is a regular file where a directory belongs. Distinct from ErrInvalidPath, which is a logical-path validation failure (NFC, reserved chars).

Migration callers MUST abort on this error rather than warn-and- skip — silently skipping it would let the importer mark the domain "imported" while the v1 blob tree is still structurally unsafe to write into. HTTP callers map it to 500 (server-side defect, not client input).

Functions

func BuildURI 

func BuildURI(scope Scope, path string) string

BuildURI produces the canonical blob_refs primary key for a (scope, path) pair. Each path segment is percent-encoded per RFC 3986 path-segment rules so URIs are unambiguous when round- tripped through HTTP / log lines (a literal `#` would otherwise be parsed as a fragment, a space would be displayed as `+` by some readers, and non-ASCII NFC characters could be re-encoded inconsistently). The `/` separators between segments are preserved as-is — a percent-encoded `/` would defeat the prefix range-scan that ListBlobRefs depends on.

Types

type DeleteOptions 

type DeleteOptions struct {
	IfMatch string
}

DeleteOptions mirrors PutOptions's IfMatch semantics for removals.

type Object 

type Object struct {
	Scope   Scope
	Path    string // logical path under scope, "/"-separated
	Size    int64
	ModTime int64  // unix milliseconds
	SHA256  string // hex; "" when unknown
	ETag    string // strong etag = "sha256:<hex>"; "" when unknown
}

Object is the metadata view of a blob. SHA256 / ETag come from the blob_refs cache when the Store was constructed With Refs; otherwise (slice 1 fs-only mode) they are populated only by Put / Verify.

type Option 

type Option func(*Store)

Option configures a Store at construction time.

func WithHomePeer 

func WithHomePeer(peer string) Option

WithHomePeer sets the peer id stamped onto blob_refs.home_peer for every Put. Must be set whenever WithRefs is — without it StoreRefs.Put rejects the call (the column is NOT NULL). Tests that use noopRefs may leave this empty.

func WithRefs 

func WithRefs(refs RefStore) Option

WithRefs wires a RefStore (typically blob.NewStoreRefs(*store.Store, homePeer)). When set, Head / Get populate ETag / SHA256 directly from the cache, IfMatch becomes a single SELECT, and Put / Delete keep the row in lock-step with the on-disk file. List is not touched in slice 2 — it still walks the filesystem and leaves SHA256 / ETag empty on each entry; callers who need digests for a listing iterate Head per result. Slice 3+ may switch List to a blob_refs query if profiling shows large fs walks dominate.

type PutOptions 

type PutOptions struct {
	IfMatch        string
	ExpectedSHA256 string
	Mode           os.FileMode
	// BypassHandoffPending lets a §3.7 pull (target-side blob
	// fetch) write a row whose existing handoff_pending flag
	// would otherwise refuse the update. The agent runtime
	// MUST NOT set this — it's reserved for the orchestrator's
	// pull path, which is itself the authority that clears the
	// flag. Reflected in the underlying RefStore via a type-
	// asserted bypass-capable interface; backends that don't
	// implement it fall back to the normal Put (which is
	// already correct for cluster-fresh target rows).
	BypassHandoffPending bool
}

PutOptions tunes a Put. ExpectedSHA256 lets the caller assert what the digest should be — if the streamed body hashes to anything else the temp file is unlinked and the publish aborts before rename. IfMatch performs optimistic concurrency against the existing blob's digest (slice-1 etag = "sha256:<hex>"); a non-matching value returns ErrETagMismatch without touching the on-disk file.

type Ref 

type Ref struct {
	URI            string
	Scope          string
	HomePeer       string
	Size           int64
	SHA256         string
	HandoffPending bool
	// UpdatedAt mirrors blob_refs.updated_at (unix millis).
	// blob.Store.Put captures the post-write timestamp so the
	// rollback path's OCC tuple is (sha256, updated_at) — a
	// concurrent same-digest mutation (scrub bump) advances
	// updated_at and our restore correctly refuses. Zero when
	// the backend doesn't track timestamps (slice-1 noopRefs).
	UpdatedAt int64
}

Ref is the metadata cache entry for one blob URI. It mirrors the columns the blob package actually reads from store.BlobRefRecord. HandoffPending was added so Store.Put can preflight the §3.7 guard before writing the body to disk — without it a refused row update would leave the new body orphaned at the rename site.

type RefSnapshot 

type RefSnapshot any

RefSnapshot is an opaque pre-write snapshot of one blob_refs row. blob.Store grabs one before any write that might need to be rolled back; the refSnapshotter implementation owns the concrete shape (StoreRefs stashes a full *store.BlobRefRecord so managed fields — refcount, pin_policy, last_seen_ok, marked_for_gc_at, handoff_pending — are preserved on restore).

type RefStore 

type RefStore interface {
	// Get returns the row keyed by uri or ErrRefNotFound when absent.
	// Implementations MUST translate their internal not-found error
	// to ErrRefNotFound so the blob layer can branch on it without
	// importing the underlying store.
	Get(ctx context.Context, uri string) (*Ref, error)

	// Put inserts or replaces the row. Returns the new
	// updated_at unix-millis stamp so the blob layer's
	// rollback path can gate Restore / DeleteIfMatches on the
	// (sha256, updated_at) tuple WITHOUT a follow-up Get that
	// would race a concurrent writer. Backends that don't
	// track timestamps return 0 — the blob layer's OCC then
	// degrades to the disabled branch (best-effort cleanup).
	Put(ctx context.Context, ref *Ref) (updatedAt int64, err error)

	// Delete removes the row. Idempotent: callers may invoke even
	// when the URI was never present, so implementations must NOT
	// return ErrRefNotFound for missing rows.
	Delete(ctx context.Context, uri string) error
}

RefStore is the minimal contract internal/blob needs from the DB. store.BlobRefStore (defined in this package) adapts *store.Store to it; tests that don't care about the cache pass nil to New() and get the noop implementation, which behaves like slice 1 (Head / List / Get leave SHA256 / ETag empty, IfMatch / Verify read the file).

type Scope 

type Scope string

Scope partitions blobs by placement / sharing policy. The three values map 1-1 to the on-disk subtrees of the kojo config dir: 

<root>/global/   — shareable across peers; the holder serves reads
                   on demand (live_read read-through), not replicated
<root>/local/    — peer-local only (FTS index, transient temp)
<root>/machine/  — never leaves the host (credentials, machine secrets)

MISTAKE: `cas` (content-addressed storage). It was added to the blob_refs CHECK in migrations/0001_initial.sql on the speculation that large files (models / datasets) would one day be partially replicated by content hash. That feature was never built and is not planned — the multi-device design settled on "the holder serves reads on demand" (proxy / live_read), so nothing is content-addressed and nothing is partially replicated. Admitting a value for a non-existent code path was the error: it implied a capability the stack does not have.

It is intentionally NOT a member of this enum and Valid() rejects it, so every write path (BuildURI → Valid) refuses scope='cas'. No row with scope='cas' can ever be created; the CHECK literal is therefore dead and unreachable.

It is left in the CHECK rather than removed because removing it buys nothing and costs real risk: SQLite cannot ALTER a CHECK, so dropping the literal needs a full blob_refs table-rebuild migration (create a twin table, copy every row, drop, rename, recreate indexes) on a core table — pure churn for a value already proven unreachable in code. 

const (
	ScopeGlobal  Scope = "global"
	ScopeLocal   Scope = "local"
	ScopeMachine Scope = "machine"
)

func ParseURI 

func ParseURI(uri string) (Scope, string, error)

ParseURI parses `kojo://<scope>/<path>` back into the typed scope + path expected by Store.Open / Head. Mirrors BuildURI: each path segment between `/` separators is url.PathEscape- encoded by the producer, so PathUnescape is applied here per segment. Returns an error on missing scheme, missing scope, empty path, or invalid percent-encoding.

Exposed alongside BuildURI so the device-switch handoff fetch (server-side) can use the same canonical decoder the scrubber uses.

func (Scope) Valid 

func (s Scope) Valid() bool

Valid reports whether s is one of the three accepted scopes.

type ScrubResult 

type ScrubResult struct {
	Total    int
	OK       int
	Missing  int
	Mismatch int
	Errors   int
}

ScrubOnce is exposed for tests and admin callers. Returns the per-outcome counts.

type Scrubber 

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

Scrubber periodically re-hashes blob bodies against blob_refs rows. One per process; not concurrency-safe across multiple Start invocations (the sync.Once on stopCh would block them).

func NewScrubber 

func NewScrubber(st *store.Store, blobs *Store, logger *slog.Logger, opts ScrubberOpts) *Scrubber

NewScrubber wires the deps. Returns nil if either store is nil so the caller doesn't have to worry about test setups that skip the blob layer.

func (*Scrubber) ScrubOnce 

func (s *Scrubber) ScrubOnce(ctx context.Context) ScrubResult

func (*Scrubber) Start 

func (s *Scrubber) Start(ctx context.Context)

Start launches the background goroutine. Runs one scrub immediately so a freshly-restored snapshot doesn't have to wait a full interval for the first verification. Stops on Stop().

Derives an internal cancellable context from the caller's so Stop can interrupt a multi-minute hash of a large blob without the caller having to thread its own cancellation through. The caller's ctx is still honoured — a parent cancel propagates.

func (*Scrubber) Stop 

func (s *Scrubber) Stop()

Stop signals the loop to exit and cancels in-flight work. Bounded: copyCtx in scrubOne polls ctx.Done so a running hash returns ctx.Err() within one read-buffer iteration after the cancel fires. wg.Wait still blocks until the goroutine exits, but the cancellation makes that exit prompt.

type ScrubberOpts 

type ScrubberOpts struct {
	// Interval between full scrubs. Zero uses DefaultScrubInterval.
	Interval time.Duration
}

ScrubberOpts overrides defaults.

type Store 

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

Store is the entry point. Construct one per kojo data root; goroutine safe because each operation re-resolves paths and uses its own file descriptors. Per-(scope, path) write serialization is provided so the IfMatch + write pair is not trivially racy with another writer holding the same etag.

func New 

func New(root string, opts ...Option) *Store

func (*Store) BaseDir 

func (s *Store) BaseDir() string

New builds a Store anchored at root (typically configdir.Path()). The caller is responsible for creating root before the first Put; individual scope subdirs are auto-created by Put as needed. Without WithRefs the store runs in slice-1 fs-only mode. BaseDir returns the absolute root path the Store writes blob bodies under. Exported for callers (notably the integrity scrubber) that need to write quarantine files *outside* the scope sub-trees so the corrupt body cannot be served back via Open / List. Returns "" when called on a nil receiver.

func (*Store) Delete 

func (s *Store) Delete(scope Scope, p string, opts DeleteOptions) error

Delete removes the blob. ErrNotFound on miss. opts.IfMatch is honored the same way Put honors it.

func (*Store) FSPath added in v0.101.2 

func (s *Store) FSPath(scope Scope, p string) (string, error)

FSPath resolves a blob logical path to its absolute on-disk path without opening it. Returns ErrInvalidScope / ErrInvalidPath for a malformed request; it does NOT check that the file exists (callers that need that should Stat the result). Intended for read-only consumers — e.g. the thumbnail generator — that take a filesystem path rather than an io.Reader.

func (*Store) Get 

func (s *Store) Get(scope Scope, p string) (io.ReadCloser, *Object, error)

Get opens the blob for streaming reads. The returned Object carries Size / ModTime; SHA256 / ETag are populated from the blob_refs cache when WithRefs is set (slice 2) and otherwise left empty (slice 1 callers can call Verify to compute). Callers MUST Close the returned reader.

func (*Store) Head 

func (s *Store) Head(scope Scope, p string) (*Object, error)

Head returns Size / ModTime without opening the body. SHA256 / ETag are populated from blob_refs when WithRefs is set; otherwise empty (slice-1 fs-only mode).

func (*Store) List 

func (s *Store) List(scope Scope, prefix string) ([]Object, error)

List walks the scope's subtree and returns every regular file whose logical path begins with `prefix`. Reserved names / prefixes / suffixes are skipped — they cannot legally exist via Put, but a human-edited blob dir might still contain them and we'd rather hide them than surprise downstream code with paths it would reject on a Put round-trip.

Output is sorted by logical path (filepath.Walk gives lexicographic order naturally). Size / ModTime are populated; SHA256 / ETag are left empty.

func (*Store) Move added in v0.101.6 

func (s *Store) Move(scope Scope, src, dst string) error

Move renames a blob from src to dst within the same scope. The on-disk file is os.Renamed (atomic on the same filesystem — always true here since both paths resolve under the same blob root). The blob_refs row is copied to the new URI and the old one deleted.

Fails if dst already exists. On ref-update failure the rename is rolled back so the blob stays at its original path.

func (*Store) Open 

func (s *Store) Open(scope Scope, p string) (*os.File, *Object, error)

Open is identical to Get but returns *os.File so callers can drive http.ServeContent (Range, conditional GET) without an interface assertion. Callers MUST Close the returned file. The Object is populated the same way as Get.

func (*Store) Put 

func (s *Store) Put(scope Scope, p string, src io.Reader, opts PutOptions) (*Object, error)

Put publishes src under (scope, path). On success the returned Object has SHA256 / ETag filled in (computed in a single pass over the input stream). If opts.IfMatch is set, the current blob's sha256 is computed and compared; a mismatch returns ErrETagMismatch without touching the file. If opts.ExpectedSHA256 is set and the streamed body hashes differently, the publish aborts before rename and the on-disk file is left unchanged.

func (*Store) Verify 

func (s *Store) Verify(scope Scope, p string) (*Object, error)

Verify computes the sha256 of the on-disk blob and returns the fully-populated Object (Size / ModTime / SHA256 / ETag). Slice 1 has no digest cache so this reads the whole file; slice 2 will short-circuit through blob_refs. Used by repair / scrub paths and by Put's IfMatch handling.

type StoreRefs 

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

StoreRefs adapts *store.Store to the RefStore interface so the blob package can stay decoupled from the SQL schema details. The adapter translates store.ErrNotFound into the blob-local ErrRefNotFound and pre-builds the `kojo://<scope>/<path>` URI that blob_refs uses as its primary key.

func NewStoreRefs 

func NewStoreRefs(st *store.Store, homePeer string) *StoreRefs

NewStoreRefs builds a RefStore-compatible adapter. homePeer is the peer id this daemon advertises in blob_refs.home_peer; for slice 2 any non-empty string works (the field is consumed by Phase 4 device switch, not by anything in this slice).

func (*StoreRefs) Delete 

func (r *StoreRefs) Delete(ctx context.Context, uri string) error

func (*StoreRefs) DeleteIfMatches 

func (r *StoreRefs) DeleteIfMatches(ctx context.Context, uri, sha256 string, updatedAt int64) (bool, error)

DeleteIfMatches satisfies the conditionalDeleter interface so blob.Store can roll back a "we created this row" Put without nuking a row a 3rd-party writer materialised in the commit- failure window. Returns (deleted, err); the blob layer treats deleted=false as "leave it alone".

func (*StoreRefs) Get 

func (r *StoreRefs) Get(ctx context.Context, uri string) (*Ref, error)

func (*StoreRefs) Put 

func (r *StoreRefs) Put(ctx context.Context, ref *Ref) (int64, error)

func (*StoreRefs) PutBypass 

func (r *StoreRefs) PutBypass(ctx context.Context, ref *Ref) (int64, error)

PutBypass satisfies the bypassPutter interface so blob.Store can route §3.7 pull writes through the AllowHandoffPending branch. Normal callers MUST use Put — the bypass is reserved for the orchestrator's pull, which already authoritatively owns the row's pending state via the begin/complete transitions.

func (*StoreRefs) Restore 

func (r *StoreRefs) Restore(ctx context.Context, snap RefSnapshot, expectedCurrentSHA256 string, expectedCurrentUpdatedAt int64) error

Restore writes a Snapshot back to blob_refs verbatim. The expectedCurrent OCC tuple the blob layer passes (typically the (sha256, updated_at) of the row right after the just-failed refs.Put) gates the UPDATE — if another writer advanced the row past that state in the commit-failure window, store. RestoreBlobRef returns ErrRestoreSuperseded and we leave their state alone. The store-level helper uses a full-column UPDATE so managed fields (refcount, pin_policy, last_seen_ok, marked_for_gc_at, handoff_pending) survive the round trip.

func (*StoreRefs) Snapshot 

func (r *StoreRefs) Snapshot(ctx context.Context, uri string) (RefSnapshot, error)

Snapshot captures the full blob_refs row for uri so blob.Store can roll back a failed write byte-for-byte (every managed field — refcount, pin_policy, last_seen_ok, marked_for_gc_at, handoff_pending — survives the round trip). Returns ErrRefNotFound when no row exists; the caller treats that as "rollback means delete the new row".

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.