trovestore

package
v0.0.2 Latest Latest
Warning

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

 Go to latest Published: Jun 21, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package trovestore implements fabriq's core/blob.Store over the Trove byte engine used as a LIBRARY (trove.Open + the driver registry). It imports only the trove library packages — never trove/extension or its metadata store — so Trove holds no catalog and can never be a source of truth.

Index

Constants

This section is empty.

Variables

View Source 
var ErrUnsupported = errors.New("fabriq: trove blob: capability not supported by driver")

ErrUnsupported is returned by a capability method when the underlying driver does not provide that capability (Capabilities() reports it false).

Functions

This section is empty.

Types

type Adapter 

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

Adapter is a blob.Store backed by a single Trove bucket.

func New 

func New(t *trove.Trove, bucket string) *Adapter

New wraps an already-opened trove.Trove. The caller owns the Trove lifecycle (it was built from a driver + options in the forgeext provider).

func Open 

func Open(ctx context.Context, cfg Config) (*Adapter, error)

Open builds a Trove instance from the config's DSN (via the driver registry), ensures the default bucket exists, and returns an Adapter. The Adapter owns the Trove handle; call Close to release it.

func (*Adapter) AbortMultipart 

func (a *Adapter) AbortMultipart(ctx context.Context, key, uploadID string) error

AbortMultipart cancels an in-progress multipart upload. Returns ErrUnsupported when the underlying driver does not implement MultipartDriver.

func (*Adapter) Capabilities 

func (a *Adapter) Capabilities() blob.Caps

Capabilities reports what the underlying driver supports by type-asserting against the Trove driver capability interfaces.

func (*Adapter) Close 

func (a *Adapter) Close(ctx context.Context) error

Close releases the underlying Trove handle. Safe to call on a nil Adapter.

func (*Adapter) CompleteMultipart 

func (a *Adapter) CompleteMultipart(ctx context.Context, key, uploadID string, parts []blob.PartInfo) (blob.ObjectInfo, error)

CompleteMultipart finalises a multipart upload. Returns ErrUnsupported when the underlying driver does not implement MultipartDriver.

func (*Adapter) Copy 

func (a *Adapter) Copy(ctx context.Context, srcKey, dstKey string) (blob.ObjectInfo, error)

Copy copies an object within the same bucket.

func (*Adapter) Delete 

func (a *Adapter) Delete(ctx context.Context, key string) error

Delete removes an object.

func (*Adapter) Driver 

func (a *Adapter) Driver() trovedriver.Driver

Driver returns the underlying trove driver so callers (e.g. Open) can construct a CASStore without importing trove/driver directly.

func (*Adapter) Get 

func (a *Adapter) Get(ctx context.Context, key string) (io.ReadCloser, blob.ObjectInfo, error)

Get retrieves an object body and its metadata.

func (*Adapter) GetRange 

func (a *Adapter) GetRange(ctx context.Context, key string, offset, length int64) (io.ReadCloser, error)

GetRange retrieves a byte range of an object. Returns ErrUnsupported when the underlying driver does not implement RangeDriver.

func (*Adapter) Head 

func (a *Adapter) Head(ctx context.Context, key string) (blob.ObjectInfo, error)

Head returns object metadata without content.

func (*Adapter) InitiateMultipart 

func (a *Adapter) InitiateMultipart(ctx context.Context, key string, o blob.PutOpts) (string, error)

InitiateMultipart starts a multipart upload and returns the upload ID. Returns ErrUnsupported when the underlying driver does not implement MultipartDriver.

func (*Adapter) List 

func (a *Adapter) List(ctx context.Context, prefix string) ([]blob.ObjectInfo, error)

List returns all objects whose keys share the given prefix.

func (*Adapter) PresignGet 

func (a *Adapter) PresignGet(ctx context.Context, key string, ttl time.Duration) (string, error)

PresignGet returns a pre-signed GET URL for the given key. Returns ErrUnsupported when the underlying driver does not implement PresignDriver.

func (*Adapter) PresignPut 

func (a *Adapter) PresignPut(ctx context.Context, key string, ttl time.Duration) (string, error)

PresignPut returns a pre-signed PUT URL for the given key. Returns ErrUnsupported when the underlying driver does not implement PresignDriver.

func (*Adapter) Put 

func (a *Adapter) Put(ctx context.Context, key string, r io.Reader, o blob.PutOpts) (blob.ObjectInfo, error)

Put stores an object and returns its metadata.

func (*Adapter) UploadPart 

func (a *Adapter) UploadPart(ctx context.Context, key, uploadID string, part int, r io.Reader) (blob.PartInfo, error)

UploadPart uploads a single part of a multipart upload. Returns ErrUnsupported when the underlying driver does not implement MultipartDriver.

type BlobReconciler 

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

BlobReconciler heals blob_cas drift and bounds byte storage for ONE tenant per Reconcile call (the caller stamps the tenant into ctx via tenant.WithTenant). It recomputes ref counts from the live blob_objects catalog (the command-authoritative truth), garbage-collects unreferenced/unpinned entries past a grace window, flags catalog rows whose bytes are missing, and deletes orphan bytes that no ledger row references.

It lives in adapters/trove so it can reach CASStore's driver directly while keeping the trove import out of core/open.go (depguard).

func NewBlobReconciler 

func NewBlobReconciler(store *CASStore, run TenantTxRunner, grace time.Duration) *BlobReconciler

NewBlobReconciler builds a reconciler over the per-tenant CAS store, running ledger SQL through run (tenant-stamped) and protecting entries younger than grace from GC.

func (*BlobReconciler) Reconcile 

func (r *BlobReconciler) Reconcile(ctx context.Context, repair bool) (Report, error)

Reconcile runs all three checks for the tenant in ctx. With repair=false it reports drift without mutating anything (dry-run).

type CASIndex 

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

CASIndex is a per-tenant cas.Index backed by the blob_cas table. Every method runs inside TenantTxRaw so FORCE RLS isolates rows to the calling tenant; no tenant predicate is needed in the WHERE clauses.

func NewCASIndex 

func NewCASIndex(run TenantTxRunner) *CASIndex

NewCASIndex returns a CASIndex that runs queries through run.

func (*CASIndex) DecrementRef 

func (c *CASIndex) DecrementRef(ctx context.Context, hash string) error

DecrementRef subtracts 1 from the ref_count for hash. Returns trovecas.ErrNotFound when no row matched (matches trove's reference MemoryIndex semantics).

func (*CASIndex) Delete 

func (c *CASIndex) Delete(ctx context.Context, hash string) error

Delete removes the entry for hash from this tenant's namespace. Returns trovecas.ErrNotFound when no row matched (matches trove's reference MemoryIndex semantics).

func (*CASIndex) Get 

func (c *CASIndex) Get(ctx context.Context, hash string) (*trovecas.Entry, error)

Get returns the entry for hash, or trovecas.ErrNotFound if absent.

func (*CASIndex) IncrementRef 

func (c *CASIndex) IncrementRef(ctx context.Context, hash string) error

IncrementRef adds 1 to the ref_count for hash. Returns trovecas.ErrNotFound when no row matched (matches trove's reference MemoryIndex semantics).

func (*CASIndex) ListUnpinned 

func (c *CASIndex) ListUnpinned(ctx context.Context) ([]*trovecas.Entry, error)

ListUnpinned returns all entries for this tenant that have ref_count = 0 and pinned = false (eligible for garbage collection). RLS already scopes the query to the calling tenant — no explicit tenant predicate needed.

func (*CASIndex) Pin 

func (c *CASIndex) Pin(ctx context.Context, hash string) error

Pin marks the entry as pinned, preventing garbage collection. Returns trovecas.ErrNotFound when no row matched (matches trove's reference MemoryIndex semantics).

func (*CASIndex) Put 

func (c *CASIndex) Put(ctx context.Context, e *trovecas.Entry) error

Put stores or upserts an entry. If the hash already exists for this tenant, the stored ref_count is incremented (per cas.Index contract).

func (*CASIndex) Unpin 

func (c *CASIndex) Unpin(ctx context.Context, hash string) error

Unpin clears the pinned flag, making the entry eligible for GC. Returns trovecas.ErrNotFound when no row matched (matches trove's reference MemoryIndex semantics).

type CASStore 

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

CASStore is a per-tenant content-addressable store. Each tenant's bytes live in their OWN bucket (bucketFor(base, tenantID)), so per-tenant GC of one tenant's unreferenced object can never touch another tenant's bytes — even when both stored identical content (same hash). The trove/cas import stays confined to this adapters/ package; open.go and core import only this type.

func NewCASStore 

func NewCASStore(drv trovedriver.Driver, idx trovecas.Index, bucket string) *CASStore

NewCASStore constructs a per-tenant CASStore over drv, indexing entries in idx and deriving each tenant's bucket from base. The signature is unchanged from Phase 3 (open.go passes the same three arguments); the third argument is now the bucket BASE used for per-tenant derivation, not a shared bucket.

Signature note: cas.New takes a driver.Driver (not *trove.Trove). Pass the raw driver obtained from memdriver.New() or troveAdapter.Driver().

func (*CASStore) Retrieve 

func (s *CASStore) Retrieve(ctx context.Context, hash string) (io.ReadCloser, error)

Retrieve returns the bytes stored under hash for the calling tenant as an io.ReadCloser. The caller is responsible for closing the reader. Returns an error if the hash is not found in this tenant's bucket.

func (*CASStore) Store 

func (s *CASStore) Store(ctx context.Context, r io.Reader) (hash string, size int64, err error)

Store writes content from r to the calling tenant's CAS bucket. Identical content stored before by THIS tenant increments ref_count (ON CONFLICT in CASIndex.Put); the returned hash is stable across duplicate writes. The trove *driver.ObjectInfo is unwrapped internally; callers receive only fabriq types.

type Config 

type Config struct {
	// StorageDriver is the backend DSN, e.g. "file:///data/blobs" or "mem://".
	StorageDriver string `json:"storageDriver" yaml:"storageDriver"`
	// DefaultBucket is the bucket all keys live under (created on Open).
	DefaultBucket string `json:"defaultBucket" yaml:"defaultBucket"`
}

Config describes how to build a Trove-backed blob.Store from a DSN.

type Report 

type Report struct {
	RefsCorrected  int      // ledger rows whose ref_count disagreed with truth
	GCCount        int      // unreferenced+unpinned entries garbage-collected
	BytesFreed     int64    // bytes reclaimed by GC
	Broken         []string // referenced hashes whose bytes are missing
	OrphansDeleted int      // stored objects with no ledger row, removed
}

Report summarizes one tenant's reconcile pass.

type TenantTxRunner 

type TenantTxRunner interface {
	TenantTxRaw(ctx context.Context, fn func(tx *pgdriver.PgTx) error) error
}

TenantTxRunner is a seam that opens a tenant-stamped raw SQL transaction. *postgres.Adapter satisfies this interface.

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.