Documentation
¶
Index ¶
- Constants
- func GetOldAccessBucketPrefixes(olderThan time.Duration) []string
- func GetOldestAccessBucketPrefix() []byte
- type Cleaner
- type KeyValue
- type SegmentDeleteStats
- type Storage
- func (s *Storage) CleanerStats() (cleaned, evicted int64)
- func (s *Storage) Close()
- func (s *Storage) DeleteIfVersion(key string, expected uint64) (retErr error)
- func (s *Storage) DeleteKey(key string) error
- func (s *Storage) FlushAccessUpdates()
- func (s *Storage) Get(key string, start, end int64) (io.Reader, bool, error)
- func (s *Storage) GetDeleteIndexStats(segmentPath string) (deletedEntries, deletedBytes int64, err error)
- func (s *Storage) GetWithVersion(key string) (io.Reader, uint64, bool, error)
- func (s *Storage) IsClosed() bool
- func (s *Storage) ListKeyValuesWithPagination(userPrefix string, startKey string, limit int) ([]KeyValue, string, bool, error)
- func (s *Storage) ListKeys(userPrefix string) ([]string, error)
- func (s *Storage) ListKeysWithPagination(userPrefix string, startKey string, limit int) ([]string, string, bool, error)
- func (s *Storage) ListSegmentDeleteStats() ([]SegmentDeleteStats, error)
- func (s *Storage) Put(key string, body io.Reader, ttl int) error
- func (s *Storage) PutIfVersion(key string, body io.Reader, ttl int, expected uint64) (newVersion uint64, retErr error)
- func (s *Storage) RemoveDeleteIndex(segmentPath string) error
- func (s *Storage) SetAccessTime(key string, accessTime int64)
- func (s *Storage) TotalSize() int64
- type StorageConfig
- type TimeBucketedAccessIndex
Constants ¶
const ( // Default disk path DefaultDiskPath = "/var/cache" // Default TTL (seconds) when no key-level TTL is set DefaultTTL = 0 // Default inline threshold (bytes) for small objects that are inlined in RocksDB DefaultInlineThreshold = 64 * 1024 // 64KB // Default compact threshold (bytes) for objects that are compacted to segments DefaultCompactThreshold = 64 * 1024 * 1024 // 64MB // Default segment size (bytes) DefaultSegmentSize = 256 * 1024 * 1024 // 256MB // Default file descriptor cache size (entries) DefaultFdCacheSize = 10000 // Default max disk usage (bytes) DefaultMaxDiskUsage = 0 // Default compaction threads DefaultCompactionThreads = 1 // Default to single thread for backwards compatibility // DefaultCompactionBytesPerSecond reserves most of a typical capped volume // for serving reads while background compaction drains its backlog. It is // the SERVER FLAG default only (see server/config.go): the storage layer // treats a non-positive CompactionBytesPerSecond as unthrottled, matching // the 0-disables convention of MaxDiskUsage, so directly constructed and // embedded storage keeps its historical unthrottled behavior unless a // budget is set explicitly. DefaultCompactionBytesPerSecond int64 = 16 * 1024 * 1024 // 16 MiB/s // Default TTL cleanup interval DefaultTTLCleanupInterval = 1 * time.Minute // DefaultFenceRetention is how long a CAS delete keeps ordering later // put-if-absent writes (issue #267): its tombstone stays as the key's fence // for this long before the TTL sweep removes it. A sizing knob — one small // row per CAS-deleted key for the window — not an enable switch; it need // only outlive the longest populate that could have observed absence // before the delete. DefaultFenceRetention = 6 * time.Hour // Default access update buffer size, batch interval and delay DefaultAccessUpdateBufferSize = 100000 DefaultAccessUpdateInterval = 1 * time.Second DefaultAccessUpdateDelay = 5 * time.Minute // Default queue config DefaultDeleteBatchSize = 1000 // Number of deletions to process per batch // Default segment recompaction settings DefaultRecompactionDisabled = false DefaultFragmentationThreshold = 0.5 // Recompact when dead space exceeds 50% DefaultMinSegmentAgeForRecompaction = 2 * time.Hour // Don't recompact segments younger than 2 hours (ensures they're cold) DefaultMinSegmentsBeforeRecompaction = 2 // Minimum number of segments required for recompaction DefaultRecompactionInterval = 1 * time.Minute // Interval between segment recompaction runs // Default delete queue settings DeleteProcessInterval = time.Second // Interval between batch processing DeletePruneAge = 24 * time.Hour // Age after which entries are pruned DeleteRetryDelay = 30 * time.Second // Backoff before retrying a failed deletion (bounds retry churn for stuck files) // Default number of parallel workers for startup file recovery DefaultRecoveryWorkers = files.MaxWorkers // Default RocksDB configuration DefaultMetadataCacheSize = metadata.DefaultRocksDBBlockCacheSize DefaultMetadataBackgroundJobs = metadata.DefaultRocksDBMaxBackgroundJobs // Eviction policies for StorageConfig.EvictionPolicy (only relevant when // MaxDiskUsage > 0). Each policy maintains its own eviction index: // // lru - evict least-recently-accessed first, using the time-bucketed // access index; a read re-buckets the entry, protecting recently // -read data. // fifo - evict oldest-written first, using the dedicated !fifo/ index; // entries are written once at Put and never touched by reads, so a // read of old data does not protect it from eviction. EvictionPolicyLRU = "lru" EvictionPolicyFIFO = "fifo" // DefaultEvictionPolicy preserves the historical behavior (LRU). DefaultEvictionPolicy = EvictionPolicyLRU )
const ( // BucketDuration defines the time window for each bucket (1 hour) BucketDuration = time.Hour )
const MaxListValueSize int64 = 1 << 20 // 1 MiB
MaxListValueSize caps the per-value bytes returned by List-with-values. A value larger than this is not read from disk; the entry is returned with the value omitted (and its size reported) so a List over large objects cannot buffer an object-sized allocation or exceed the gRPC message limit (#165).
Variables ¶
This section is empty.
Functions ¶
func GetOldAccessBucketPrefixes ¶
GetOldAccessBucketPrefixes returns a list of bucket prefixes older than the given duration that can be deleted during maintenance
func GetOldestAccessBucketPrefix ¶
func GetOldestAccessBucketPrefix() []byte
GetOldestBucket returns the prefix for iterating the oldest bucket This starts from the beginning of the access index
Types ¶
type Cleaner ¶
type Cleaner struct {
// contains filtered or unexported fields
}
Cleaner is responsible for background TTL cleanup and LRU eviction
func NewCleaner ¶
NewCleaner creates a new Cleaner for background TTL cleanup and LRU eviction
func (*Cleaner) Close ¶
func (c *Cleaner) Close()
Close stops the background cleanup loop and waits for it to exit
func (*Cleaner) Start ¶
func (c *Cleaner) Start()
Start launches the background cleanup goroutine It performs an initial size calculation synchronously to establish accurate baseline before any concurrent operations can modify the size
func (*Cleaner) TotalSize ¶ added in v1.8.0
TotalSize returns the current tracked logical cache size in bytes (sum of stored object lengths), maintained live on every write and eviction.
func (*Cleaner) UpdateSize ¶
UpdateSize updates the tracked total size when keys are added/removed
func (*Cleaner) WaitForInitialization ¶
func (c *Cleaner) WaitForInitialization()
WaitForInitialization waits until the cleaner has completed its initial size calculation
type KeyValue ¶
type KeyValue struct {
Key string
Value []byte
// ValueLength is the size of the value in bytes, set even when the value was
// omitted for exceeding MaxListValueSize.
ValueLength int64
// ValueOmitted is true when the value was not read because it exceeds
// MaxListValueSize; Value is nil in that case.
ValueOmitted bool
}
KeyValue holds a key and its associated value bytes.
type SegmentDeleteStats ¶
SegmentDeleteStats holds deletion statistics for a segment
type Storage ¶
type Storage struct {
// contains filtered or unexported fields
}
Storage wraps all RocksDB access and related logic It provides methods to store, retrieve, delete, and list keys
Value encoding format: For small objects (in-memory):
"S|" + [8-byte big-endian expiry] + '|' + [data bytes]
For large objects (spilled to disk):
"L|" + [8-byte big-endian expiry] + '|' + [file path as bytes]
If no TTL is set, expiry and separator are omitted:
"S|" + [data bytes] or "L|" + [file path as bytes]
The separator '|' after the expiry ensures robust parsing, even if the data or file path contains '|'.
On read, expiry is checked (if present) and expired keys are deleted and not returned.
func NewStorageWithConfig ¶
func NewStorageWithConfig(config *StorageConfig) (*Storage, error)
NewStorageWithConfig creates a new isolated Storage instance with the given config.
func (*Storage) CleanerStats ¶
CleanerStats returns statistics from the background cleaner
func (*Storage) Close ¶
func (s *Storage) Close()
Close closes this storage instance and releases all resources. This is safe to call on isolated instances created with NewStorageWithConfig.
func (*Storage) DeleteIfVersion ¶ added in v1.12.0
DeleteIfVersion deletes the key only if it admits expected: a live key needs an exact version match; an absent or dead key admits 0 (delete-if-absent) or a token at or after its fence. The delete lands as a REF-LESS already-expired tombstone stamped with this delete's version (the merge operator drops the backing references so an in-flight compaction operand cannot resurrect it). That stamp is the key's FENCE (issue #267): until the cleaner ages it out (fence retention), GetWithVersion reports it as the absent key's token and a put carrying an older observation is rejected. Delete-if-absent on a missing or dead key therefore is NOT a no-op: it records the delete as a fence, or moves an existing fence forward, so a populate that observed absence before this delete loses to it.
Because the tombstone carries no references, this method reclaims the replaced value's backing bytes itself on a confirmed win and drops the key's eviction-index entries; the cleaner only removes the tiny leftover row once the fence has aged out. This frees the (potentially 256 MB) backing file immediately rather than one cleanup interval later.
func (*Storage) FlushAccessUpdates ¶
func (s *Storage) FlushAccessUpdates()
FlushAccessUpdates forces all pending access time updates to be written immediately This is mainly useful for testing to ensure deterministic behavior
func (*Storage) Get ¶
Get retrieves the value for the given key from the database and returns an io.Reader for streaming Supports byte-range requests via start and end parameters (0 means no limit)
func (*Storage) GetDeleteIndexStats ¶
func (s *Storage) GetDeleteIndexStats(segmentPath string) (deletedEntries, deletedBytes int64, err error)
GetDeleteIndexStats returns the delete index statistics for a segment
func (*Storage) GetWithVersion ¶ added in v1.12.0
GetWithVersion returns the value reader together with the key's current CAS version (issue #254). It is a CAS-path operation that does not touch the plain Get read path. The value reader and the version come from a SINGLE metadata read, so the pair is one atomic snapshot (a second read via Get could pair one generation's data with another's version).
An absent key reports found == false together with an OBSERVATION TOKEN (issue #267), never 0: the stamp of the CAS delete that removed it while that fence is retained, otherwise a fresh stamp meaning "absent as of now". Pass the token back as expected to PutIfVersion to order the write against any delete: a delete stamped after the observation rejects it. Pass 0 instead for the unordered put-if-absent. A live pre-versioning (plain-written) row reports merge.VersionLegacy; mixing plain writes and CAS on one key is unsupported.
func (*Storage) IsClosed ¶
IsClosed returns true if this storage instance has been closed. This can be used to check if it's safe to call other Storage methods.
func (*Storage) ListKeyValuesWithPagination ¶
func (s *Storage) ListKeyValuesWithPagination(userPrefix string, startKey string, limit int) ([]KeyValue, string, bool, error)
ListKeyValuesWithPagination returns paginated, sorted key-value pairs from RocksDB. For inline values the data is returned directly; for file/segment values the data is read from disk. Returns: (entries, lastKey, hasMore, error).
func (*Storage) ListKeys ¶
ListKeys returns all non-expired keys in the RocksDB instance that match the given prefix Note: Expired keys are skipped but not deleted - deletion is handled by the background cleaner
func (*Storage) ListKeysWithPagination ¶
func (s *Storage) ListKeysWithPagination(userPrefix string, startKey string, limit int) ([]string, string, bool, error)
ListKeysWithPagination returns paginated, sorted keys from RocksDB Returns: (keys, lastKey, hasMore, error) - keys: Up to 'limit' keys starting after 'startKey' - lastKey: The last key in this page (for continuation) - hasMore: True if more keys exist beyond this page
func (*Storage) ListSegmentDeleteStats ¶
func (s *Storage) ListSegmentDeleteStats() ([]SegmentDeleteStats, error)
ListSegmentDeleteStats returns deletion statistics for all segments in the delete index
func (*Storage) PutIfVersion ¶ added in v1.12.0
func (s *Storage) PutIfVersion(key string, body io.Reader, ttl int, expected uint64) (newVersion uint64, retErr error)
PutIfVersion writes the value only if the key admits expected: a live key needs an exact version match; an absent key admits 0 (put-if-absent) or an observation token from GetWithVersion, which a CAS delete stamped after that observation rejects (issue #267). On success it returns the new version; on a lost race it returns *storageErrors.VersionMismatchError carrying the current version — for a fenced key, the fence stamp to refetch and retry with. Plain writes keep last-write-wins semantics and are unaffected.
func (*Storage) RemoveDeleteIndex ¶
RemoveDeleteIndex removes the delete index entry for a segment (used when segment is removed)
func (*Storage) SetAccessTime ¶
SetAccessTime sets a specific access time (Unix seconds) for a key. This is mainly useful for testing to create predictable LRU scenarios.
func (*Storage) TotalSize ¶ added in v1.8.0
TotalSize returns the current logical cache size in bytes: the sum of stored object lengths across all keys. It is maintained live on every write and eviction, so embedders can publish their own gauge without a full rescan. Returns 0 if the cleaner is not initialized.
type StorageConfig ¶
type StorageConfig struct {
DiskPath string // Directory for on-disk cache data
TTL int // Default TTL when no key-level TTL is set (seconds)
InlineThreshold int // Threshold for small objects that are inlined in RocksDB (bytes)
CompactThreshold int64 // Objects less than this size are compacted to segments (bytes)
SegmentSize int64 // Segment size (bytes)
FdCacheSize int // Size of the file descriptor cache
MaxDiskUsage int64 // Maximum disk usage in bytes (0 = unlimited)
CompactionThreads int // Number of compaction threads
FragThreshold float64 // Fragmentation threshold for segment recompaction (0.0-1.0)
MinSegmentAge time.Duration // Minimum age for segment recompaction
MinSegments int // Minimum number of segments for recompaction
DisableRecompaction bool // Disable automatic segment recompaction
RecompactionInterval time.Duration // Interval between segment recompaction runs
CleanupInterval time.Duration // Cleanup interval
FenceRetention time.Duration // How long a CAS-delete fence outlives the delete (0 = DefaultFenceRetention)
AccessUpdateDelay time.Duration // Access update delay
RecoveryWorkers int // Number of parallel workers for startup file recovery (<= 0 = default)
DeleteBatchSize int // Number of file deletions processed per deletion-queue batch (<= 0 = default)
EvictionPolicy string // Eviction order when MaxDiskUsage > 0: "lru" (default) or "fifo"
CompactionBytesPerSecond int64 // Shared file/recompaction payload-byte limit (<= 0 = unthrottled)
// RocksDB-specific configuration
MetadataCacheSize int64 // RocksDB Block cache size in bytes (0 = use default)
MetadataBackgroundJobs int // Max concurrent RocksDB background jobs, compactions+flushes (0 = use default)
}
StorageConfig holds all configuration parameters for initializing storage
type TimeBucketedAccessIndex ¶
type TimeBucketedAccessIndex struct {
}
TimeBucketedAccessIndex manages access times using time-bucketed keys for LRU eviction without loading all keys into memory
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package benchio provides optional benchmark-only payload I/O controls.
|
Package benchio provides optional benchmark-only payload I/O controls. |
|
Package merge implements a multiplexing merge operator for RocksDB that supports different merge strategies based on key types.
|
Package merge implements a multiplexing merge operator for RocksDB that supports different merge strategies based on key types. |