engine

Documentation

Overview

Package engine provides the BlockStore orchestrator that composes local store, remote store, and syncer into the blockstore.Store interface.

The orchestrator lives in a sub-package (not the root blockstore package) to avoid import cycles: blockstore/local and blockstore/sync both import the root blockstore package for types and interfaces, so the root package cannot import them back.

Index

• type BlockStore
  • func New(cfg Config) (*BlockStore, error)
  • func (bs *BlockStore) Close() error
  • func (bs *BlockStore) CopyPayload(ctx context.Context, srcPayloadID, dstPayloadID string) (int, error)
  • func (bs *BlockStore) Delete(ctx context.Context, payloadID string) error
  • func (bs *BlockStore) DrainAllUploads(ctx context.Context) error
  • func (bs *BlockStore) EvictLocal(ctx context.Context, payloadID string) error
  • func (bs *BlockStore) EvictReadBuffer() int
  • func (bs *BlockStore) Exists(ctx context.Context, payloadID string) (bool, error)
  • func (bs *BlockStore) Flush(ctx context.Context, payloadID string) (*blockstore.FlushResult, error)
  • func (bs *BlockStore) GetSize(ctx context.Context, payloadID string) (uint64, error)
  • func (bs *BlockStore) GetStats() BlockStoreStats
  • func (bs *BlockStore) HasRemoteStore() bool
  • func (bs *BlockStore) HealthCheck(ctx context.Context) error
  • func (bs *BlockStore) Healthcheck(ctx context.Context) health.Report
  • func (bs *BlockStore) ListFiles() []string
  • func (bs *BlockStore) LocalStats() local.Stats
  • func (bs *BlockStore) ReadAt(ctx context.Context, payloadID string, data []byte, offset uint64) (int, error)
  • func (bs *BlockStore) ReadAtWithCOWSource(ctx context.Context, payloadID, cowSource string, data []byte, offset uint64) (int, error)
  • func (bs *BlockStore) RemoteForTesting() remote.RemoteStore
  • func (bs *BlockStore) SetEvictionEnabled(enabled bool)
  • func (bs *BlockStore) SetRetentionPolicy(policy blockstore.RetentionPolicy, ttl time.Duration)
  • func (bs *BlockStore) Start(ctx context.Context) error
  • func (bs *BlockStore) Stats() (*blockstore.Stats, error)
  • func (bs *BlockStore) Truncate(ctx context.Context, payloadID string, newSize uint64) error
  • func (bs *BlockStore) WriteAt(ctx context.Context, payloadID string, data []byte, offset uint64) error
• type BlockStoreStats
• type Config

Types

type BlockStore

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

BlockStore is the central orchestrator for block storage. It composes a local store, optional remote store, and syncer into the blockstore.Store interface. All protocol adapters and runtime code use BlockStore for I/O.

Read operations check the read buffer first, then the local store, falling back to remote download via the syncer on miss. Write operations go directly to the local store and invalidate the read buffer; the syncer handles background upload to remote.

func New

func New(cfg Config) (*BlockStore, error)

New creates a new BlockStore from the given configuration. Local store and syncer are required; remote may be nil for local-only mode.

func (*BlockStore) Close

func (bs *BlockStore) Close() error

Close releases resources held by the store. Closes prefetcher first (stops workers), then read buffer, then syncer (drains uploads), local store, and remote store.

func (*BlockStore) CopyPayload

func (bs *BlockStore) CopyPayload(ctx context.Context, srcPayloadID, dstPayloadID string) (int, error)

CopyPayload duplicates all blocks from srcPayloadID to dstPayloadID.

For each source block, data is read from the local store (falling back to remote download on miss) and written to the destination's local store. If a remote store is configured, blocks are also copied server-side (e.g., S3 CopyObject) to avoid redundant uploads.

Returns the number of blocks copied and any error encountered.

func (*BlockStore) Delete

func (bs *BlockStore) Delete(ctx context.Context, payloadID string) error

Delete removes all data for a payload from local store and remote store. Invalidates all read buffer entries for the file and resets prefetcher state.

func (*BlockStore) DrainAllUploads

func (bs *BlockStore) DrainAllUploads(ctx context.Context) error

DrainAllUploads waits for all pending uploads to complete.

func (*BlockStore) EvictLocal

func (bs *BlockStore) EvictLocal(ctx context.Context, payloadID string) error

EvictLocal removes all local data (memory and disk) for a file.

func (*BlockStore) EvictReadBuffer

func (bs *BlockStore) EvictReadBuffer() int

EvictReadBuffer clears all entries from the read buffer. Returns the number of entries that were cleared.

func (*BlockStore) Exists

func (bs *BlockStore) Exists(ctx context.Context, payloadID string) (bool, error)

Exists checks whether a payload exists. Checks local store first, falls back to syncer (remote).

func (*BlockStore) Flush

func (bs *BlockStore) Flush(ctx context.Context, payloadID string) (*blockstore.FlushResult, error)

Flush ensures all dirty data for a payload is persisted. After flush, auto-promotes block data into the read buffer if the file fits within the budget (data is in OS page cache, so the read is essentially free).

func (*BlockStore) GetSize

func (bs *BlockStore) GetSize(ctx context.Context, payloadID string) (uint64, error)

GetSize returns the stored size of a payload. Checks local store first, falls back to syncer (remote).

func (*BlockStore) GetStats

func (bs *BlockStore) GetStats() BlockStoreStats

GetStats returns comprehensive block store statistics.

func (*BlockStore) HasRemoteStore

func (bs *BlockStore) HasRemoteStore() bool

HasRemoteStore returns true if this BlockStore has a remote store configured.

func (*BlockStore) HealthCheck

func (bs *BlockStore) HealthCheck(ctx context.Context) error

HealthCheck verifies the store is operational by checking the syncer health (which in turn checks the remote store).

Legacy error-returning probe. New callers should prefer Healthcheck (lowercase 'c') which returns a structured health.Report derived from both the local and remote stores and satisfies health.Checker.

func (*BlockStore) Healthcheck

func (bs *BlockStore) Healthcheck(ctx context.Context) health.Report

Healthcheck returns the engine's overall health, computed as the worst-of of its underlying local and remote stores. The result satisfies health.Checker so the API layer can wrap the engine in a health.CachedChecker for /status routes.

Derivation rules (worst-of):

• If the local store reports unhealthy → engine is unhealthy (we can't even serve cached blocks).
• If a remote store is configured and reports unhealthy → engine is degraded (local reads still work, but new uploads will queue and the system is operating in offline-write mode).
• Otherwise → healthy.

The combined message preserves the worst-status component's message so operators can see exactly which subsystem is at fault.

func (*BlockStore) ListFiles

func (bs *BlockStore) ListFiles() []string

ListFiles returns the payloadIDs of all files tracked in the local store.

func (*BlockStore) LocalStats

func (bs *BlockStore) LocalStats() local.Stats

LocalStats returns a snapshot of local store statistics.

func (*BlockStore) ReadAt

func (bs *BlockStore) ReadAt(ctx context.Context, payloadID string, data []byte, offset uint64) (int, error)

ReadAt reads data from storage at the given offset into dest. Checks read buffer first, then local store, falling back to remote download on miss.

func (*BlockStore) ReadAtWithCOWSource

func (bs *BlockStore) ReadAtWithCOWSource(ctx context.Context, payloadID, cowSource string, data []byte, offset uint64) (int, error)

ReadAtWithCOWSource reads data with copy-on-write source fallback. If data is not found in the primary payloadID, it falls back to cowSource.

func (*BlockStore) RemoteForTesting

func (bs *BlockStore) RemoteForTesting() remote.RemoteStore

RemoteForTesting returns the remote store for cross-package test verification (e.g., shared remote store identity). Do not use in production code.

func (*BlockStore) SetEvictionEnabled

func (bs *BlockStore) SetEvictionEnabled(enabled bool)

SetEvictionEnabled controls whether the local store can evict blocks to free disk space. Delegates to the local store's SetEvictionEnabled method.

func (*BlockStore) SetRetentionPolicy

func (bs *BlockStore) SetRetentionPolicy(policy blockstore.RetentionPolicy, ttl time.Duration)

SetRetentionPolicy updates the retention policy on the underlying local store. Delegates to the local store's SetRetentionPolicy method.

func (*BlockStore) Start

func (bs *BlockStore) Start(ctx context.Context) error

Start initializes the store and starts background goroutines. Recovery runs on the local store first (if supported), then the syncer and local store background goroutines are started. Finally, the prefetcher is created if both the read buffer and prefetch workers are configured.

func (*BlockStore) Stats

func (bs *BlockStore) Stats() (*blockstore.Stats, error)

Stats returns storage statistics from the local store.

func (*BlockStore) Truncate

func (bs *BlockStore) Truncate(ctx context.Context, payloadID string, newSize uint64) error

Truncate changes the size of a payload in both local store and remote store. Invalidates read buffer entries above the new size and resets prefetcher state.

func (*BlockStore) WriteAt

func (bs *BlockStore) WriteAt(ctx context.Context, payloadID string, data []byte, offset uint64) error

WriteAt writes data to storage at the given offset. Writes go directly to the local store; the syncer handles background upload. Read buffer entries for affected blocks are invalidated and prefetcher is reset.

type BlockStoreStats

type BlockStoreStats struct {
	FileCount    int `json:"file_count"`
	BlocksDirty  int `json:"blocks_dirty"`
	BlocksLocal  int `json:"blocks_local"`
	BlocksRemote int `json:"blocks_remote"`
	BlocksTotal  int `json:"blocks_total"`

	LocalDiskUsed int64 `json:"local_disk_used"`
	LocalDiskMax  int64 `json:"local_disk_max"`
	LocalMemUsed  int64 `json:"local_mem_used"`
	LocalMemMax   int64 `json:"local_mem_max"`

	ReadBufferEntries int   `json:"read_buffer_entries"`
	ReadBufferUsed    int64 `json:"read_buffer_used"`
	ReadBufferMax     int64 `json:"read_buffer_max"`

	HasRemote      bool `json:"has_remote"`
	PendingSyncs   int  `json:"pending_syncs"`
	PendingUploads int  `json:"pending_uploads"`
	CompletedSyncs int  `json:"completed_syncs"`
	FailedSyncs    int  `json:"failed_syncs"`

	RemoteHealthy       bool    `json:"remote_healthy"`
	EvictionSuspended   bool    `json:"eviction_suspended"`
	OutageDurationSecs  float64 `json:"outage_duration_seconds"`
	OfflineReadsBlocked int64   `json:"offline_reads_blocked"`
}

BlockStoreStats holds comprehensive block store statistics for a BlockStore.

type Config

type Config struct {
	// Local is the on-node block store (required).
	Local local.LocalStore

	// Remote is the durable backend store (nil for local-only mode).
	Remote remote.RemoteStore

	// Syncer handles async local-to-remote transfers (required).
	Syncer *blocksync.Syncer

	// FileBlockStore provides block metadata for block store statistics.
	// When set, GetStats() populates BlocksLocal/BlocksRemote/BlocksTotal.
	FileBlockStore blockstore.FileBlockStore

	// ReadBufferBytes is the memory budget for the read buffer per share.
	// 0 disables the read buffer. Passed directly to readbuffer.New as byte budget.
	ReadBufferBytes int64

	// PrefetchWorkers is the number of goroutines for sequential prefetch.
	// 0 disables prefetching.
	PrefetchWorkers int
}

Config holds the components that make up a BlockStore.