chunkstore

type BatchItem ¶

type BatchItem struct {
	ID    uint64
	Data  []byte
	Class StoreClass
}

BatchItem is one slice in a WriteBatch.

type BatchReadStats ¶

type BatchReadStats = snapshot.BatchReadStats

BatchReader is an optional Store capability: read many slices in ONE call so the backing can coalesce fetches of slices that share a casstore pack (far fewer backing/S3 GETs than one Read per slice — the cold sequential-read fix for model loading). concurrency bounds in-flight backing (pack) fetches (<=0 → backing default). It is BEST-EFFORT: a slice that can't be read is omitted from the result rather than failing the batch (the demand path re-fetches per slice and surfaces real errors). The sequential-read prefetcher uses it when the backing implements it. BatchReadStats aliases casstore's stats so cache/readahead can consume it through the chunkstore package without importing casstore directly.

type BatchReader ¶

type BatchReader interface {
	ReadBatch(ctx context.Context, ids []uint64, concurrency int) (map[uint64][]byte, BatchReadStats, error)
}

type BatchWriter ¶

type BatchWriter interface {
	WriteBatch(ctx context.Context, items []BatchItem) error
}

BatchWriter is an optional Store capability: write many slices in ONE packed operation so their chunks share casstore packs (far fewer object/S3 PUTs than one Write per slice). The disk-cache write-back drain uses it when the backing store implements it — the fix for the -remote per-128-KB-slice PUT storm.

type CasStore ¶

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

CasStore implements Store over a casstore ChunkedStore. The ChunkedStore should be configured with a GlobalIndex so identical content across distinct slice ids deduplicates.

func New(cs *snapshot.ChunkedStore) *CasStore

func (s *CasStore) Exists(ctx context.Context, id uint64) (bool, error)

func (s *CasStore) Read(ctx context.Context, id uint64) ([]byte, error)

func (s *CasStore) ReadAt(ctx context.Context, id uint64, off int64, p []byte) (int, error)

func (s *CasStore) ReadBatch(ctx context.Context, ids []uint64, concurrency int) (map[uint64][]byte, snapshot.BatchReadStats, error)

func (s *CasStore) RegisterSlice(ctx context.Context, id uint64, chunks []snapshot.ChunkLocation, class StoreClass) error

func (s *CasStore) Remove(ctx context.Context, id uint64) error

func (s *CasStore) SliceChunks(ctx context.Context, id uint64) ([]snapshot.ChunkLocation, error)

func (s *CasStore) WalkSliceIDs(ctx context.Context, fn func(id uint64, createdAt time.Time) error) error

func (s *CasStore) Write(ctx context.Context, id uint64, data []byte, class StoreClass) error

func (s *CasStore) WriteBatch(ctx context.Context, items []BatchItem) error

type Local ¶

type Local struct {
	Store  *CasStore
	GC     *snapshot.ChunkedGC
	Dedup  *snapshot.MemoryDedupStore
	Chunks blobstore.Storage
	// Compactor is the underlying casstore store, exposed so operator tooling
	// (mlfs-admin compact) can consolidate under-filled/small packs.
	Compactor *snapshot.ChunkedStore
}

Local bundles a CasStore with the casstore pieces behind it, for dev, tests, and single-node deployments: the GC (caller schedules it), the dedup index, and the chunk blob store (for inspecting physical blob counts).

func NewLocal(dir, domain string, packTarget int, opts ...LocalOption) (*Local, error)

type LocalOption ¶

type LocalOption func(*localOpts)

LocalOption tunes NewLocal.

func WithCompression(p snapshot.CompressionPolicy) LocalOption

func WithMeter(m metric.Meter) LocalOption

func WithPackCompression(m snapshot.PackCompressionMode) LocalOption

type Remote ¶

type Remote struct {
	Store *CasStore
}

Remote bundles a CasStore wired over the converged direct-S3 node data path: a global dedup index served by blobgw over NATS, presigned-URL S3 transfers, and an injected manifest store. Unlike Local it carries NO GC, dedup, or chunk-store handles: garbage collection is blobgw-owned (it holds the tenant credentials and the authoritative index), so the node never constructs one. Downstream consumers (cache→fileio→fuse) take *CasStore through the chunkstore.Store interface and are untouched by the remote vs local choice.

func NewRemote(cfg RemoteConfig) (*Remote, error)

type RemoteConfig ¶

type RemoteConfig struct {
	Tenant          string
	PackTargetBytes int
	Requester       remoteindex.Requester
	HTTP            *s3http.Client
	Manifests       snapshot.SnapshotStore
	Logger          *slog.Logger
	PresignTTL      time.Duration

	// Compression selects the casstore pack-blob compression policy. A nil
	// policy uses the remote default: UNCOMPRESSED (see NewRemote) — the correct
	// choice for the converged design's primary workload (large, largely
	// incompressible model weights such as .pt/.safetensors), because it
	// preserves mmap/zero-copy device loads per ADR-001 §6. Set it to
	// snapshot.NewContentTypePolicy(snapshot.CompressZstd) to opt compressible
	// workloads (e.g. workclaw scratch) back into zstd.
	//
	// Compressing no longer costs ranged reads: per-chunk pack framing (the
	// PackCompression default) keeps compressed packs range-readable, so the old
	// "compressed ⇒ whole-pack fetch" tradeoff is gone. Zero-copy mmap remains the
	// reason the model/tensor class stays uncompressed.
	Compression snapshot.CompressionPolicy

	// PackCompression selects the framing for packs that ARE compressed:
	// per-chunk (zero value, range-readable) or whole-pack (rollback). It has no
	// effect under the uncompressed default above — an uncompressed pack is framed
	// identically either way.
	PackCompression snapshot.PackCompressionMode

	// Meter, when non-nil, wraps the direct-S3 blob store with the casstore
	// backing-store OTEL adapter (obs.Wrap, backend="s3"), so cold-read and
	// write-back S3 GET/PUT latency, errors, and bytes are recorded — the RED view
	// that is otherwise dark on the converged path. The tenant is attached as
	// mlfs.domain so the series joins this daemon's identity stream.
	Meter metric.Meter
}

RemoteConfig configures NewRemote.

Tenant is the casstore dedup domain (the blob-ID prefix and dedup isolation boundary) and the NATS subject tenant; it must be a valid NATS subject token (see ctlproto.ValidateTenant). Requester is the shared NATS transport that feeds BOTH the remote dedup index and the presign client. HTTP is the node-side S3 transfer client. Manifests is the snapshot manifest store, injected by the caller (cmd wires the PG manifeststore; tests may pass a local store) so this constructor stays storage-agnostic and PG-free. Logger is optional (passed to the GlobalIndex for best-effort dedup diagnostics). PresignTTL, when >0, overrides the requested presigned-URL validity.

type Store ¶

type Store interface {
	// Write stores the bytes of slice id under compression class class.
	// Idempotent for identical content (casstore dedups the underlying chunks;
	// the class only selects the physical codec, never the chunk identity).
	Write(ctx context.Context, id uint64, data []byte, class StoreClass) error
	// Read returns the full bytes of slice id.
	Read(ctx context.Context, id uint64) ([]byte, error)
	// ReadAt reads len(p) bytes of slice id starting at off, returning the
	// number of bytes read (io.ReadFull semantics).
	ReadAt(ctx context.Context, id uint64, off int64, p []byte) (int, error)
	// Exists reports whether slice id has been written.
	Exists(ctx context.Context, id uint64) (bool, error)
	// Remove deletes slice id's manifest; its chunks are reclaimed by GC once
	// no slice references them.
	Remove(ctx context.Context, id uint64) error
}

Store reads and writes slice data. Implementations are safe for concurrent use. Slice ids are allocated by the metadata engine (NewSlice) and are opaque, globally-unique handles here.

type StoreClass ¶

type StoreClass uint8

StoreClass is a slice's compression class: it decides whether the backing casstore compresses the slice's pack or stores it uncompressed. mlfs sets it per file (by file type, at create) so a model/tensor file stays uncompressed (range-readable + mmap/zero-copy device loads) while generic data compresses — on the SAME tenant, instead of relying on a global CompressNone switch. The class is carried with the staged slice (crash-safe) so a recovery drain packs it correctly with no inode context. The zero value is ClassDefault.

const (
	// ClassDefault leaves the codec to the backing's CompressionPolicy (generic
	// content compresses; the policy still stores recognized already-compressed /
	// tensor content types uncompressed). The unchanged pre-classification
	// behavior.
	ClassDefault StoreClass = iota
	// ClassUncompressed forces the slice's pack to be stored uncompressed
	// regardless of policy — the model/tensor class that must stay byte-for-byte
	// on disk for ranged reads and mmap/zero-copy device loads.
	ClassUncompressed
)