local_cache

package
v0.0.0-...-a1a4906 Latest Latest
Warning

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

 Go to latest Published: Nov 15, 2026 License: Apache-2.0 Imports: 86 Imported by: 0

Documentation

Overview

persistent_cache.go implements PersistentCache, the main entry point for the local-cache subsystem. It coordinates:

  • Object resolution and download orchestration (resolveObject, downloadObject, performDownload).
  • Cache-Control / ETag-based revalidation (revalidateObject).
  • Inline (small-file) vs disk (large-file) storage decisions via decisionWriter.
  • No-store passthrough streaming when the origin forbids caching.
  • Namespace ID assignment and federation token management.
  • Background configuration updates from the director.

Supporting types live in other files: CacheDB (database.go), StorageManager / BlockWriter (storage.go), EvictionManager (eviction.go), ConsistencyChecker (consistency.go), and BlockFetcherV2 (block_fetcher.go).

Index

Constants

View Source 
const (
	// ChunkSize is the notification granularity - notify waiters every 128KB
	ChunkSize = 128 * 1024

	// DefaultPrefetchTimeout is the default time after which a prefetch with no active
	// clients will be cancelled
	DefaultPrefetchTimeout = 20 * time.Second

	// PrefetchSemaphoreReleaseInterval is how often to release and reacquire the prefetch semaphore
	PrefetchSemaphoreReleaseInterval = 5 * time.Second

	// ETAUpdateInterval is how often to update the ETA estimate
	ETAUpdateInterval = 250 * time.Millisecond

	// DefaultInitialRate is the initial assumed download rate (1 MB/s)
	DefaultInitialRate = 1024 * 1024

	// ETAStaleThreshold is how long an ETA can be late before waiters give up
	ETAStaleThreshold = 5 * time.Second
)
View Source 
const (
	// PrefixMeta stores CacheMetadata (headers, validation info, storage mode)
	PrefixMeta = "m:"
	// PrefixState stores Roaring Bitmap tracking downloaded blocks
	PrefixState = "s:"
	// PrefixInline stores encrypted inline data for small objects (< 4KB)
	PrefixInline = "d:"
	// PrefixLRU stores sorted index for eviction candidates: l:<storage_id>:<namespace_id>:<ts>:<instance_hash>
	PrefixLRU = "l:"
	// PrefixUsage stores total bytes used per storage+namespace: u:<storage_id>:<namespace_id>
	PrefixUsage = "u:"
	// PrefixDiskMap stores the mapping of disk IDs to directories
	PrefixDiskMap = "di:"
	// PrefixPurgeFirst stores instance hashes marked for priority eviction
	PrefixPurgeFirst = "pf:"
	// PrefixETag stores the latest ETag for an object: e:<object_hash> -> etag
	PrefixETag = "e:"
	// PrefixNamespace stores namespace prefix -> ID mappings: n:<prefix> -> uint32
	PrefixNamespace = "n:"
	// PrefixAppendIntent records that an AppendWriter is part-way through
	// building an object: aw:<instance_hash> -> msgpack(AppendIntent).
	//
	// A streaming append charges capacity for every chunk it allocates but has
	// no LRU entry until it finishes, so an interrupted one would otherwise be
	// invisible to both eviction and the consistency checker -- it has metadata
	// *and* files, so neither half of RunMetadataScan fires.  This marker is
	// the reclamation handle: NewAppendWriter writes it, Finalize and Abort
	// remove it, and anything left over belongs to a writer that did not
	// survive.  See StorageManager.ReclaimAbandonedAppends.
	PrefixAppendIntent = "aw:"
	// PrefixTierUpload tracks in-progress uploads to tiering targets:
	// up:<instance_hash> -> msgpack(TierUploadIntent).  An intent is written
	// before the first byte reaches S3 and removed only after the object
	// has been relocated (or the partial upload aborted), so a crash never
	// leaves untracked objects in the bucket.
	PrefixTierUpload = "up:"
	// PrefixRedirectHold records when a pre-signed URL was last handed out for
	// an object: ps:<instance_hash> -> 8-byte big-endian UnixNano.
	// Eviction skips objects with a recent redirect timestamp.
	PrefixRedirectHold = "ps:"

	// KeySalt is the single DB key that stores the random salt used when
	// hashing object/instance names.  The salt prevents an attacker with
	// DB access from correlating hashes with known URLs.
	KeySalt = "_salt"

	// KeyStoreMode records whether this database belongs to a cache or to a
	// pstore-backed origin.  The two share this key space, so opening one as
	// the other would silently corrupt it; both check this marker on open.
	KeyStoreMode = "_mode"

	// KeySchemaVersion records the layout of the records in this database:
	// which keys exist, how their values are encoded, and how the hashes that
	// name them are derived.  The value is the decimal ASCII form of a
	// SchemaVersion, so a dump of the database reads plainly.
	//
	// KeyStoreMode says *who* owns the database; this says *what layout* the
	// owner wrote.  See CurrentSchemaVersion and CacheDB.ensureSchemaVersion.
	KeySchemaVersion = "_schema"
)

Key prefixes for BadgerDB.

This block is the authoritative registry of every key namespace used in the cache database; when adding a new namespace, document it here (and add a key constructor next to the others below). Per-object keys must also be removed in deleteObjectInTxn and PurgeStorageID or they will leak on eviction / storage recycling.

View Source 
const (
	// PrefixDirent stores pstore directory entries, keyed by parent path and
	// name: pd:<parent path>\x00<name>
	PrefixDirent = "pd:"
	// PrefixGarbage stores pstore instances and subtrees awaiting reclamation
	PrefixGarbage = "pg:"
)

Prefixes reserved by the pstore origin backend (see docs/pstore-design.md). A pstore shares this key space but never opens a cache database and vice versa (enforced via KeyStoreMode), so the prefixes cannot collide at runtime. They are declared here so that future cache features do not claim them.

View Source 
const (
	// BlockDataSize is the size of data in each encrypted block (before encryption)
	BlockDataSize = 4080
	// AuthTagSize is the size of the AES-GCM authentication tag
	AuthTagSize = 16
	// BlockTotalSize is the total size of an encrypted block on disk
	BlockTotalSize = BlockDataSize + AuthTagSize
	// InlineThreshold is the max size for inline storage (< 4KB stored in DB)
	InlineThreshold = 4096
	// NonceSize is the standard size for AES-GCM nonce
	NonceSize = 12
	// KeySize is the size for AES-256 key
	KeySize = 32
)

Block size constants for encryption and storage

View Source 
const (
	CCFlagNoStore        = ccNoStore
	CCFlagNoCache        = ccNoCache
	CCFlagPrivate        = ccPrivate
	CCFlagMustRevalidate = ccMustRevalidate
)

Cache-Control flag bits — canonical definitions are in cache_control.go (ccNoStore, ccNoCache, ccPrivate, ccMustRevalidate). These aliases are kept for any callers that reference the CCFlag* names directly.

View Source 
const (
	// BackendPosix is a local directory-backed storage target.
	BackendPosix = ""
	// BackendTier is an S3 bucket-backed storage target.
	BackendTier = "s3"
)

Backend types for DiskMapping. The zero value ("") means a local POSIX directory for backward compatibility with existing databases.

View Source 
const DefaultReadBatchBlocks = 16

DefaultReadBatchBlocks is the default number of blocks to read from disk in a single ReadAt syscall before decrypting block-by-block. 16 blocks × 4096 bytes = 64 KiB.

View Source 
const NoCacheGracePeriod = 5 * time.Second

NoCacheGracePeriod is the minimum interval between revalidation attempts when the origin sends Cache-Control: no-cache. Strictly speaking, no-cache means "must revalidate before each use," but revalidation in Pelican requires a full GET (no conditional-request support yet), which is expensive. A short grace period prevents a thundering-herd of redundant downloads when many clients hit the same object concurrently.

View Source 
const OptimalReadSize = DefaultReadBatchBlocks * BlockDataSize

OptimalReadSize is the ideal number of plaintext bytes to request per read call. It equals DefaultReadBatchBlocks × BlockDataSize (16 × 4080 = 65 280 B). When callers align their buffer sizes to a multiple of BlockDataSize (4080), every block decrypts directly into the destination buffer (zero-copy). Misaligned sizes cause the last partial block in each batch to be decrypted into a temporary buffer and copied, adding overhead.

View Source 
const (
	// SaltSize is the number of random bytes prepended to object/instance
	// names before hashing.  32 bytes (256 bits) provides a comfortable
	// security margin.
	SaltSize = 32
)

Variables

View Source 
var ErrInitNoStore = errors.New("object must not be stored (no-store/no-cache/private)")

ErrInitNoStore is a sentinel error returned by initObjectFromStat when the origin's Cache-Control indicates the response must not be stored. The caller should fall back to a full streaming download via downloadObject.

View Source 
var ErrNoStore = errors.New("origin response has Cache-Control: no-store")

ErrNoStore is returned when the origin sends Cache-Control: no-store (or private). The response data is streamed via persistentDownload.noStoreReader (an io.Pipe) directly to the first caller without persisting to the cache.

View Source 
var ErrNoStoreRetry = errors.New("no-store download in progress; retry independently")

ErrNoStoreRetry is returned to waiters that attach to an in-flight no-store download. Because the stream can only be consumed once (by the first caller), subsequent callers must retry independently.

View Source 
var ErrNotCached = errors.New("object not cached")

ErrNotCached is returned when an object is not in the cache

View Source 
var ErrReadOnly = errors.New("the cache database is open read-only")

ErrReadOnly is returned by every mutating CacheDB method when the handle was opened for introspection. Callers can test for it with errors.Is.

View Source 
var ErrTierObjectChanged = errors.New("the tiering target no longer holds the copy of this object that was uploaded")

ErrTierObjectChanged reports that a tiering target no longer holds the copy of an object the cache uploaded -- it was overwritten or replaced.

View Source 
var ErrTierRedirectUnavailable = errors.New("the tiering target cannot issue a redirect URL right now")

ErrTierRedirectUnavailable reports that a backend that can issue redirect URLs cannot issue one at the moment -- it is waiting for a credential to mint them from, say. The caller proxies instead.

Functions

func AppendIntentKey 

func AppendIntentKey(instanceHash InstanceHash) []byte

AppendIntentKey returns the BadgerDB key for an in-flight streaming append.

func BlockOffset 

func BlockOffset(blockNum uint32) int64

BlockOffset returns the byte offset in the file for a given block number

func BlocksInChunk 

func BlocksInChunk(contentLength int64, chunkSizeCode ChunkSizeCode, chunkIndex int) (startBlock, endBlock uint32)

BlocksInChunk returns the range of blocks (startBlock, endBlock inclusive) that belong to a specific chunk.

func CalculateBlockCount 

func CalculateBlockCount(contentLength int64) uint32

CalculateBlockCount returns the number of blocks needed for a given content length

func CalculateChunkCount 

func CalculateChunkCount(contentLength int64, chunkSizeCode ChunkSizeCode) int

CalculateChunkCount returns the number of chunks needed to store an object of the given size with the specified chunk size code. Returns 1 if chunking is disabled (the entire object is one "chunk").

func CalculateFileSize 

func CalculateFileSize(contentLength int64) int64

CalculateFileSize returns the exact on-disk file size for a given content length. The last block is only as large as the remaining data plus its AES-GCM authentication tag — it is not padded to a full BlockTotalSize.

func CheckCacheObjectExists 

func CheckCacheObjectExists(ctx context.Context, socketPath, objectPath string) (bool, error)

CheckCacheObjectExists checks if an object exists in the PersistentCache by making a HEAD request to the cache's unix socket. Returns true if the object exists. NOTE: This will trigger a download if the object is not cached, so it always returns true unless the origin is unreachable. For testing eviction, use CheckCacheObjectIsCached instead.

func CheckCacheObjectIsCached 

func CheckCacheObjectIsCached(ctx context.Context, socketPath, objectPath string) (bool, error)

CheckCacheObjectIsCached checks if an object is actually in the cache (not just accessible). This sends a special header to prevent downloading if the object is not cached. Returns true only if the object is already cached, false if it would need to be downloaded.

func ChecksumAlgorithmName 

func ChecksumAlgorithmName(t ChecksumType) string

ChecksumAlgorithmName returns the RFC 3230 token for a checksum type, as it appears in Want-Digest and Digest headers. Unknown types return "".

func ChunkContainsBlock 

func ChunkContainsBlock(contentLength int64, chunkSizeCode ChunkSizeCode, chunkIndex int, blockNum uint32) bool

ChunkContainsBlock returns true if the specified block belongs to the given chunk.

func ChunkContentLength 

func ChunkContentLength(totalContentLength int64, chunkSizeCode ChunkSizeCode, chunkIndex int) int64

ChunkContentLength returns the content length for a specific chunk. This is the size of data within the chunk (not the encrypted file size).

func ChunkSizeCodeToBytes 

func ChunkSizeCodeToBytes(code ChunkSizeCode) uint64

ChunkSizeCodeToBytes converts a ChunkSizeCode to the actual chunk size in bytes. Returns 0 if chunking is disabled. The returned size is always rounded down to a multiple of BlockDataSize to ensure blocks don't span chunk boundaries.

func ContentOffsetToBlock 

func ContentOffsetToBlock(contentOffset int64) uint32

ContentOffsetToBlock converts a content byte offset to a block number

func ContentOffsetToChunk 

func ContentOffsetToChunk(contentOffset int64, chunkSizeCode ChunkSizeCode) int

ContentOffsetToChunk returns the chunk index for a given content byte offset.

func ContentOffsetWithinBlock 

func ContentOffsetWithinBlock(contentOffset int64) int

ContentOffsetWithinBlock returns the offset within a block for a content offset

func DefaultFreshness 

func DefaultFreshness(lastValidated time.Time) time.Duration

DefaultFreshness returns the jittered freshness lifetime used when the origin doesn't provide explicit Cache-Control headers. The jitter is deterministic per object (seeded from lastValidated) so that repeated calls for the same object return the same duration.

func DiskMappingKey 

func DiskMappingKey(storageID StorageID) []byte

DiskMappingKey returns the BadgerDB key for a disk mapping entry.

func ETagKey 

func ETagKey(objectHash ObjectHash) []byte

ETagKey returns the BadgerDB key for ETag lookup Maps objectHash -> latest ETag for that object

func FormatChecksumValue 

func FormatChecksumValue(c Checksum) string

FormatChecksumValue renders a checksum the way RFC 3230 expects for its algorithm: message digests are base64, CRCs are lowercase hex. Unknown types return "".

Getting this wrong is not cosmetic. A client compares the value it computed against the one the origin reports and fails the transfer on a mismatch, even when the underlying digests are identical.

func FormatChunkSize 

func FormatChunkSize(code ChunkSizeCode) string

FormatChunkSize returns a human-readable string for a ChunkSizeCode.

func FormatContentRange 

func FormatContentRange(start, end, total int64) string

FormatContentRange formats a Content-Range header value

func FormatDigestEntry 

func FormatDigestEntry(c Checksum) string

FormatDigestEntry renders one checksum as an RFC 3230 "name=value" pair, or "" when the algorithm is not recognized or the digest is empty.

Skipping an empty digest is a deliberate change from the cache's previous Digest-header code, which emitted a bare "md5=" for a zero-value Checksum{Type: ChecksumMD5}. Every algorithm here produces a fixed-length digest, so an empty Value never means "the digest of nothing" -- it means the checksum was never populated. Advertising it anyway is worse than omitting it: RFC 3230 gives a client no way to read "md5=" as "unknown", so the Pelican client compares the digest it computed against an empty string and fails an otherwise-good transfer as a checksum mismatch. A missing entry, by contrast, is exactly the "no digest available" signal clients already handle.

func FormatDigestHeader 

func FormatDigestHeader(checksums []Checksum) string

FormatDigestHeader renders a set of checksums as an RFC 3230 Digest header value. Unknown algorithms are skipped.

func GetChunkFileSuffix 

func GetChunkFileSuffix(chunkIndex int) string

GetChunkFileSuffix returns the file suffix for a chunk. Chunk 0 has no suffix (the base file), chunk 1 is "-2", chunk 2 is "-3", etc.

func GetChunkPath 

func GetChunkPath(basePath string, chunkIndex int) string

GetChunkPath returns the full path for a chunk file. basePath is the path for chunk 0 (no suffix).

func GetChunkRange 

func GetChunkRange(contentLength int64, chunkSizeCode ChunkSizeCode, chunkIndex int) (start, end int64)

GetChunkRange returns the start and end byte offsets (inclusive) for a specific chunk. chunkIndex is 0-based.

func GetInstanceStoragePath 

func GetInstanceStoragePath(hash InstanceHash) string

GetInstanceStoragePath returns the 2-level directory path for storing a file Given hash "42561abfe18be...", returns "42/56/1abfe18be..."

func InitIssuerKeyForTests 

func InitIssuerKeyForTests(t testing.TB)

InitIssuerKeyForTests initializes issuer keys for testing. This must be called before creating a CacheDB or EncryptionManager. It generates a new issuer key in a temporary directory and registers a cleanup.

func InlineKey 

func InlineKey(instanceHash InstanceHash) []byte

InlineKey returns the BadgerDB key for inline data

func IsRangeRequest 

func IsRangeRequest(req *http.Request) bool

IsRangeRequest checks if a request contains a Range header

func LRUKey 

func LRUKey(storageID StorageID, namespaceID NamespaceID, timestamp time.Time, instanceHash InstanceHash) []byte

LRUKey returns the BadgerDB key for LRU tracking Format: l:<storage_id>:<namespace_id>:<timestamp_ns>:<instance_hash>

func MetaKey 

func MetaKey(instanceHash InstanceHash) []byte

MetaKey returns the BadgerDB key for metadata

func NamespaceKey 

func NamespaceKey(prefix string) []byte

NamespaceKey returns the BadgerDB key for a namespace prefix mapping

func NewChecksumHasher 

func NewChecksumHasher(t ChecksumType) (hash.Hash, error)

NewChecksumHasher returns a hash for the given checksum type.

func NormalizePelicanURL 

func NormalizePelicanURL(urlStr string) string

NormalizePelicanURL normalizes a Pelican URL for consistent hashing. Handles pelican://, osdf://, and bare paths.

func OffsetInChunk 

func OffsetInChunk(contentOffset int64, chunkSizeCode ChunkSizeCode) int64

OffsetInChunk converts an absolute content offset to an offset within a chunk. Returns the offset relative to the start of the chunk.

func ParseLRUKey 

func ParseLRUKey(key []byte) (storageID StorageID, namespaceID NamespaceID, timestamp time.Time, instanceHash InstanceHash, err error)

ParseLRUKey parses an LRU key and returns storageID, namespaceID, timestamp, and instanceHash

func ParseUsageKey 

func ParseUsageKey(key []byte) (storageID StorageID, namespaceID NamespaceID, err error)

ParseUsageKey extracts the storage ID and namespace ID from a usage key

func PurgeFirstKey 

func PurgeFirstKey(instanceHash InstanceHash) []byte

PurgeFirstKey returns the BadgerDB key for purge first tracking

func RedirectHoldKey 

func RedirectHoldKey(instanceHash InstanceHash) []byte

RedirectHoldKey returns the BadgerDB key recording the last time a pre-signed URL was handed out for an object.

func RemainingFreshness 

func RemainingFreshness(lastValidated time.Time) time.Duration

RemainingFreshness returns how much freshness lifetime is left for an object whose origin did not specify Cache-Control, clamped to zero.

func StateKey 

func StateKey(instanceHash InstanceHash) []byte

StateKey returns the BadgerDB key for block state bitmap

func TierUploadIntentKey 

func TierUploadIntentKey(instanceHash InstanceHash) []byte

TierUploadIntentKey returns the BadgerDB key tracking an in-progress upload of an object to a tiering target.

func UsageKey 

func UsageKey(storageID StorageID, namespaceID NamespaceID) []byte

UsageKey returns the BadgerDB key for namespace usage counter per storage Format: u:<storage_id>:<namespace_id>

func ValidateChunkLocations 

func ValidateChunkLocations(contentLength int64, chunkSizeCode ChunkSizeCode, locations []ChunkLocation) error

ValidateChunkLocations checks that ChunkLocations has the correct number of entries. Returns nil if valid, error otherwise.

Types

type AppendIntent 

type AppendIntent struct {
	StartedAt   time.Time   `msgpack:"started_at"`
	NamespaceID NamespaceID `msgpack:"namespace_id"`
}

AppendIntent is the record written while an AppendWriter is building an object. StartedAt lets a reclamation pass leave very recent appends alone even when it cannot consult the in-process registry of live writers.

type AppendWriter 

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

AppendWriter streams an object of unknown length into chunked storage.

It is not safe for concurrent use; a single writer owns one object version for the duration of the write.

func (*AppendWriter) Abort 

func (w *AppendWriter) Abort() error

Abort discards a partially written object, removing its files and metadata.

func (*AppendWriter) Finalize 

func (w *AppendWriter) Finalize() (*CacheMetadata, error)

Finalize flushes any buffered tail, records the object's true length, trims the over-allocated tail chunk, and returns the completed metadata.

After Finalize the writer is closed and further writes are rejected.

func (*AppendWriter) Write 

func (w *AppendWriter) Write(p []byte) (int, error)

Write appends data to the object. It always consumes the whole slice.

A flush failure is reported with the full byte count, not zero: by then the data is already in w.pending and already counted in w.written, so claiming nothing was accepted would invite an io.Copy-style caller to resend bytes this writer has already accounted for. The writer is unusable after such an error; the caller should Abort.

func (*AppendWriter) Written 

func (w *AppendWriter) Written() int64

Written returns the number of content bytes accepted so far.

type AuthorizationError 

type AuthorizationError struct {
	Reason string
}

AuthorizationError wraps authorizationDenied with a human-readable reason. errors.Is(err, authorizationDenied) still works for existing checks.

func (*AuthorizationError) Error 

func (e *AuthorizationError) Error() string

func (*AuthorizationError) Is 

func (e *AuthorizationError) Is(target error) bool

type Batch 

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

Batch allows batching multiple writes for efficiency

func (*Batch) Cancel 

func (b *Batch) Cancel()

Cancel discards the batch

func (*Batch) Delete 

func (b *Batch) Delete(key []byte) error

Delete adds a delete operation to the batch

func (*Batch) Flush 

func (b *Batch) Flush() error

Flush commits the batch

func (*Batch) Set 

func (b *Batch) Set(key, value []byte) error

Set adds a key-value pair to the batch

type BlockEncryptor 

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

BlockEncryptor handles encryption/decryption of individual blocks

func NewBlockEncryptor 

func NewBlockEncryptor(dek, baseNonce []byte) (*BlockEncryptor, error)

NewBlockEncryptor creates a new block encryptor with the given DEK and base nonce

func (*BlockEncryptor) DecryptBlock 

func (be *BlockEncryptor) DecryptBlock(blockNum uint32, encryptedBlock []byte) ([]byte, error)

DecryptBlock decrypts a block and verifies its authentication tag Input is BlockTotalSize bytes (ciphertext + auth tag) Returns the decrypted data (up to BlockDataSize bytes)

func (*BlockEncryptor) DecryptBlockTo 

func (be *BlockEncryptor) DecryptBlockTo(dst []byte, blockNum uint32, encryptedBlock []byte) ([]byte, error)

DecryptBlockTo decrypts a block directly into the provided destination buffer. dst must have sufficient capacity (at least len(encryptedBlock) - AuthTagSize). The decrypted data is appended to dst and the resulting slice is returned. This avoids allocation when dst has enough capacity.

func (*BlockEncryptor) EncryptBlock 

func (be *BlockEncryptor) EncryptBlock(blockNum uint32, data []byte) ([]byte, error)

EncryptBlock encrypts a block of data and returns data + auth tag The input data must be exactly BlockDataSize bytes (or less for the last block) The output is BlockTotalSize bytes (data + 16 byte auth tag)

func (*BlockEncryptor) EncryptBlockTo 

func (be *BlockEncryptor) EncryptBlockTo(dst []byte, blockNum uint32, data []byte) ([]byte, error)

EncryptBlockTo encrypts a block of data directly into the provided destination buffer. dst must have sufficient capacity (at least len(data) + AuthTagSize). The encrypted data is appended to dst and the resulting slice is returned. This avoids allocation when dst has enough capacity.

type BlockFetcherV2 

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

BlockFetcherV2 handles fetching missing blocks from the origin using the Pelican transfer client. It supports: - Per-128KB notification channels for partial completion - Client registration for data requests - Prefetch with semaphore management - Configurable timeout for prefetch cancellation when no active clients

Each BlockFetcherV2 creates its own TransferClient from the shared TransferEngine. This prevents concurrent doFetch goroutines (e.g. during auto-repair) from stealing results off a shared Results() channel.

func NewBlockFetcherV2 

func NewBlockFetcherV2(
	storage *StorageManager,
	instanceHash InstanceHash, originURL, token string,
	fedToken client.TokenProvider,
	te *client.TransferEngine,
	cfg BlockFetcherV2Config,
) (*BlockFetcherV2, error)

NewBlockFetcherV2 creates a new block fetcher using the Pelican transfer client. It creates its own TransferClient from the given TransferEngine to avoid sharing Results() channels with other callers.

func (*BlockFetcherV2) AdoptTransfer 

func (bf *BlockFetcherV2) AdoptTransfer(
	ctx context.Context,
	cancelTransfer context.CancelFunc,
	tc *client.TransferClient,
	dw *decisionWriter,
	resultChan <-chan *client.TransferResults,
	egrp *errgroup.Group,
	wg *sync.WaitGroup,
	onExit func(err error),
) *fetchOperation

AdoptTransfer takes ownership of an already-in-flight full-object transfer initiated by performDownload and drives it to completion using the fetcher's existing idle-timeout and chunk-notification machinery.

Instead of creating a new BlockWriter, AdoptTransfer wraps the decisionWriter's existing BlockWriter with the fetcher's blockWriter adapter via dw.HandoffBlockWriter. All subsequent data written by the transfer engine flows through the adapter, gaining chunk notification and ETA tracking. Blocks already written before the handoff are harmlessly skipped (the shared bitmap check in writeCurrentBlock handles this).

Parameters:

  • ctx: the transfer's context (cancelled on cache close)
  • cancelTransfer: cancels the transfer itself; called when the object goes idle -- no reader open (see ObjectBlockState.AttachReader) for prefetchTimeout
  • tc: the TransferClient that owns the transfer (closed on exit)
  • dw: the decisionWriter whose BlockWriter will be wrapped
  • resultChan: pre-filtered channel delivering the single matching result
  • egrp: errgroup for goroutine lifecycle management (test cleanup)
  • wg: waitgroup to track goroutine completion (decremented on exit)
  • onExit: called when the adopted transfer exits for any reason (clear downloading flag, close completionDone, etc.)

The method starts a goroutine (managed via egrp) and returns the fetchOperation for chunk notification and ETA queries.

func (*BlockFetcherV2) Close 

func (bf *BlockFetcherV2) Close()

Close shuts down the fetcher's dedicated TransferClient. Must be called when the fetcher is no longer needed.

func (*BlockFetcherV2) CreateFetchCallback 

func (bf *BlockFetcherV2) CreateFetchCallback() func(ctx context.Context, startBlock, endBlock uint32) error

CreateFetchCallback returns a callback function for the RangeReader

func (*BlockFetcherV2) FetchBlocks 

func (bf *BlockFetcherV2) FetchBlocks(ctx context.Context, startBlock, endBlock uint32) error

FetchBlocks fetches the specified range of blocks from the origin. Blocks until all requested blocks are available or an error occurs. Implicitly marks this fetcher as having active client interest.

func (*BlockFetcherV2) FetchBlocksAsync 

func (bf *BlockFetcherV2) FetchBlocksAsync(ctx context.Context, startBlock, endBlock uint32) (*fetchOperation, error)

FetchBlocksAsync fetches blocks asynchronously and returns a channel that will be closed when the fetch completes (either successfully or with error). The returned error is non-nil only if the fetch couldn't be started. Check the fetchOperation for the final error after doneCh is closed. Implicitly marks this fetcher as having active client interest.

func (*BlockFetcherV2) Fill 

func (bf *BlockFetcherV2) Fill(ctx context.Context, startBlock, endBlock uint32) error

Fill fetches blocks [startBlock, endBlock] in the background for the object's readers: it is cancelled once no reader of the object has been open for the prefetch timeout (see idle), and readers wait for its blocks on the shared block state rather than through the fetcher. Like the download a cache miss starts, and unlike a speculative prefetch, it takes no slot of the prefetch semaphore: it runs because a reader is waiting. Fill returns when the fetch ends, with its error.

func (*BlockFetcherV2) GetChunkChannel 

func (bf *BlockFetcherV2) GetChunkChannel(op *fetchOperation, chunkIndex int64) <-chan struct{}

GetChunkChannel returns a channel that will be closed when the specified chunk is complete. This is the safe notification pattern - callers wait for close, no values are sent. Returns nil if the chunk is already complete.

func (*BlockFetcherV2) StartPrefetch 

func (bf *BlockFetcherV2) StartPrefetch(ctx context.Context)

StartPrefetch starts prefetching the entire object in the background. The prefetch will be cancelled if there are no active clients for the prefetch timeout period.

func (*BlockFetcherV2) WaitForChunkWithETA 

func (bf *BlockFetcherV2) WaitForChunkWithETA(ctx context.Context, op *fetchOperation, chunkIndex int64) (completed bool, err error)

WaitForChunkWithETA waits for a chunk to complete, but gives up if the per-chunk ETA becomes stale. Returns true if the chunk completed, false if the ETA became stale (caller should try direct download).

type BlockFetcherV2Config 

type BlockFetcherV2Config struct {
	PrefetchTimeout time.Duration
	// PrefetchSem is a shared semaphore limiting the total number of
	// concurrent prefetches across all fetchers and downloads.  When
	// nil, a per-fetcher semaphore is created with capacity 5.
	PrefetchSem chan struct{}
}

BlockFetcherV2Config holds configuration for the block fetcher

type BlockRange 

type BlockRange struct {
	Start uint32
	End   uint32
}

BlockRange represents a contiguous range of blocks

type BlockSummary 

type BlockSummary struct {
	TotalBlocks      uint32   `json:"total_blocks"`
	DownloadedBlocks uint32   `json:"downloaded_blocks"`
	MissingBlocks    []uint32 `json:"missing_blocks,omitempty"` // Up to first 100 missing blocks
	IsComplete       bool     `json:"is_complete"`
	PercentComplete  float64  `json:"percent_complete"`
}

BlockSummary provides an overview of which blocks have been downloaded.

type BlockWriter 

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

BlockWriter is an io.WriteCloser that encrypts and writes blocks directly to disk storage. It buffers incoming data and writes complete blocks as they are filled. Contiguous encrypted blocks are batched (up to writeBatchBlocks) so that multiple blocks can be written to disk in a single WriteAt call, reducing syscall overhead and improving sequential write throughput. This is used for efficient streaming downloads with block-level encryption.

func (*BlockWriter) Abort 

func (bw *BlockWriter) Abort()

Abort closes the writer after a transfer that failed in a way that condemns its data, such as a checksum mismatch, or before it got under way. Nothing more is published: the blocks still waiting in the write batch and a partial block in the buffer are discarded, an object of unknown size is not finalized at the point the input happened to stop, and the object is never marked complete. The batch must not be flushed, because flushing marks its blocks present and wakes every reader waiting for them -- who would read and serve them before anything had marked the object bad; the caller drops the object next (see BlockFetcherV2.dropIfCondemned and PersistentCache.endAdoptedDownload). Every error path must use Abort or StopEarly rather than Close. Safe to call more than once, and after Close.

func (*BlockWriter) BytesWritten 

func (bw *BlockWriter) BytesWritten() int64

BytesWritten returns the total number of bytes written so far

func (*BlockWriter) Close 

func (bw *BlockWriter) Close() error

Close flushes any remaining data and closes the file

func (*BlockWriter) Flush 

func (bw *BlockWriter) Flush() error

Flush writes any accumulated batch of encrypted blocks to disk and updates the shared block state so that concurrent readers can see them immediately. This is a no-op when the batch is empty.

Callers that need low-latency streaming (e.g. the blockWriter adapter in AdoptTransfer) should call Flush periodically instead of waiting for the batch to reach writeBatchBlocks.

func (*BlockWriter) StopEarly 

func (bw *BlockWriter) StopEarly()

StopEarly closes the writer of a transfer that stopped before the end for a reason that says nothing against the data it delivered (see transferStopKeepsData): the cache cancelled it, or the connection broke. Like Abort, it keeps the whole blocks and discards a partial one -- unless that is the object's final block, complete -- and does not finalize an object of unknown size. Unlike Abort, it then checks, as Close does, whether every block of the object is now present: the transfer may have filled the object's last hole before it stopped, and the object must then be marked complete, or nothing ever would. Safe to call more than once, and after Close or Abort.

func (*BlockWriter) Write 

func (bw *BlockWriter) Write(p []byte) (n int, err error)

Write implements io.Writer. Data is buffered and written to disk in encrypted blocks. Blocks that already exist in the bitmap are skipped.

type CacheDB 

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

CacheDB wraps BadgerDB with cache-specific operations

func NewCacheDB 

func NewCacheDB(ctx context.Context, baseDir string) (*CacheDB, error)

NewCacheDB creates and initializes a new cache database.

Requires issuer keys to be initialized via config.GetIssuerPublicJWKS() or InitIssuerKeyForTests() before calling this function.

func OpenCacheDBReadOnly 

func OpenCacheDBReadOnly(baseDir string) (*CacheDB, error)

OpenCacheDBReadOnly opens an existing cache database for introspection. This is suitable for CLI tools that need to inspect cache contents without modifying anything.

Like NewCacheDB, this requires issuer keys to be available for decrypting the database encryption key.

"Read-only" is enforced by Pelican, not by BadgerDB. BadgerDB's own ReadOnly mode cannot be used here:

  • On Linux and macOS it aborts the process. Opening an *encrypted* database that has flushed at least one SST file read-only drives Table.fetchIndex into y.Check(err) -> log.Fatalf, which prints "err: invalid argument" and exits without returning an error or running a single defer. Every database NewCacheDB creates is encrypted and every cache that has served an object has an SST, so this fired on every real cache.
  • On Windows it is refused outright ("Read-only mode is not supported on Windows").

So the database is opened the ordinary way and mutation is refused in Go instead, via the readOnly flag that checkWritable guards every writing entry point with. This is the same trade pstore.OpenMaintenance makes.

What that costs is BadgerDB's exclusive directory lock, which an ordinary open takes: introspection now fails while a cache is running rather than opening alongside it. That is acceptable -- offline introspection is only reached when the cache is not running, since the CLI routes to the live service otherwise -- and it is arguably better, because holding the lock stops a cache from starting up underneath an introspection in progress. A lock failure is translated into an explanation of exactly that, since "another process is using this Badger database" is not something an operator should have to decode.

What it does not cost is the guarantee the read-only path exists for: this open still never stamps a schema version (VerifySchemaVersion only checks) and never claims a store mode (EnsureStoreMode is not called), so inspecting a database leaves its header exactly as it was found.

func (*CacheDB) AddUsage 

func (cdb *CacheDB) AddUsage(storageID StorageID, namespaceID NamespaceID, delta int64) error

AddUsage atomically adjusts the namespace-scoped usage counter by delta bytes. Internally this uses BadgerDB's MergeOperator so that the write is append-only (no read-modify-write cycle) and cannot conflict with concurrent transactions.

func (*CacheDB) ChargeUsage 

func (cdb *CacheDB) ChargeUsage(storageID StorageID, namespaceID NamespaceID, delta int64) error

ChargeUsage is an alias for AddUsage for backward compatibility.

func (*CacheDB) ClearBlocks 

func (cdb *CacheDB) ClearBlocks(instanceHash InstanceHash, blocks []uint32) error

ClearBlocks removes the specified blocks from the downloaded bitmap so they will be re-fetched on the next read. This is used during auto-repair when corruption is detected.

func (*CacheDB) Close 

func (cdb *CacheDB) Close() error

Close closes the database. All usage MergeOperators are stopped first (blocking until their background goroutines exit) so that accumulated deltas are flushed before the DB is closed.

func (*CacheDB) ComputeActualUsage 

func (cdb *CacheDB) ComputeActualUsage() (map[StorageUsageKey]int64, error)

ComputeActualUsage performs a full scan of the metadata table to compute the real byte-level usage per (StorageID, NamespaceID).

Completed objects contribute their full ContentLength. In-progress objects contribute the bytes implied by their block bitmap.

This is an expensive read-only operation. The consistency checker accumulates usage during its metadata scan instead of calling this; it is retained for ad-hoc diagnostics and tests.

func (*CacheDB) ComputeInlineDataSize 

func (cdb *CacheDB) ComputeInlineDataSize() (int64, error)

ComputeInlineDataSize scans all inline data entries (d: prefix) and sums the stored value sizes. This gives the actual bytes consumed in BadgerDB for inline object data (excluding metadata overhead).

func (*CacheDB) DB 

func (cdb *CacheDB) DB() *badger.DB

DB exposes the underlying BadgerDB handle.

This exists for sibling subsystems that store their own records in this key space and need transactional semantics the typed helpers above do not provide — read-your-writes transactions spanning several keys, and prefix iteration with seeks. The pstore origin backend (see docs/pstore-design.md) is the only such consumer today.

Callers must confine themselves to prefixes they own, and must call EnsureStoreMode first so that a cache and a pstore can never operate on the same database.

This is the one way past the read-only guard: the handle it returns is a writable BadgerDB even when ReadOnly reports true (see OpenCacheDBReadOnly for why the underlying database is not opened read-only). A consumer that honors read-only opens must consult ReadOnly before writing through it -- pstore does, in Store.checkWritable.

func (*CacheDB) DeleteAppendIntent 

func (cdb *CacheDB) DeleteAppendIntent(instanceHash InstanceHash) error

DeleteAppendIntent removes an append-in-flight record.

func (*CacheDB) DeleteBlockState 

func (cdb *CacheDB) DeleteBlockState(instanceHash InstanceHash) error

DeleteBlockState removes block state for a file

func (*CacheDB) DeleteLatestETag 

func (cdb *CacheDB) DeleteLatestETag(objectHash ObjectHash) error

DeleteLatestETag removes the ETag entry for an object

func (*CacheDB) DeleteMetadata 

func (cdb *CacheDB) DeleteMetadata(instanceHash InstanceHash) error

DeleteMetadata removes metadata for a file

func (*CacheDB) DeleteObject 

func (cdb *CacheDB) DeleteObject(instanceHash InstanceHash) error

DeleteObject removes all data for a cached object. Uses metadata to compute exact LRU key for efficient deletion. Also cleans up ETag table, purge-first marker, and adjusts usage counters. Usage is decremented via AddUsage after the transaction commits so that the write cannot conflict.

func (*CacheDB) DeleteTierUploadIntent 

func (cdb *CacheDB) DeleteTierUploadIntent(instanceHash InstanceHash) error

DeleteTierUploadIntent removes an upload intent record.

func (*CacheDB) EnsureStoreMode 

func (cdb *CacheDB) EnsureStoreMode(mode StoreMode) error

EnsureStoreMode records which subsystem owns this database, or verifies that a previously recorded owner matches.

Caches and pstore origins share one key space (see the prefix constants in schema.go), so opening one as the other would silently corrupt it. Every consumer must call this immediately after opening.

A database with no marker is adopted when it is empty. A non-empty database with no marker is necessarily a legacy cache — pstore always writes the marker — so adopting it is allowed only for StoreModeCache.

The other half of the database header, the schema version, is settled by NewCacheDB before this runs: a layout this binary cannot read is refused before anyone asks who owns it, because the ownership marker is itself just another record written under that layout.

func (*CacheDB) EvictByLRU 

func (cdb *CacheDB) EvictByLRU(storageID StorageID, namespaceID NamespaceID, maxObjects int, maxBytes int64, skip func(InstanceHash) bool, claim func(InstanceHash)) ([]evictedObject, int, error)

EvictByLRU evicts objects from a storage+namespace combination, draining purge-first items before walking the regular LRU index — all within a single BadgerDB transaction.

Eviction stops when either maxObjects have been removed or maxBytes of content has been freed — whichever comes first. A value of 0 for either limit means "no limit on that dimension". The method is allowed to go one object over the byte threshold so that progress is always made even when only large objects remain.

skip, when non-nil, is consulted for every candidate; returning true leaves the object in place. It is how a caller protects objects that must not disappear right now -- see StorageManager.EvictByLRU, which uses it to spare objects under a live reader. A skipped object is not counted against maxObjects, because skipping frees nothing and counting it would let eviction report a full batch while returning under target.

claim, when non-nil, is called with each object, inside the transaction, just before it is deleted; StorageManager.EvictByLRU uses it to shut the object's writers out (see instanceFences). It may be called for an object whose deletion then fails, or whose transaction does not commit.

Returns the evicted objects and the number that were skipped.

func (*CacheDB) FindRecyclableStorageID 

func (cdb *CacheDB) FindRecyclableStorageID(mountedDirs map[StorageID]string) (StorageID, error)

FindRecyclableStorageID searches persisted disk mappings for the unmounted storageID with the least usage. mountedDirs maps storageID to directory path for IDs currently assigned to live directories — those are excluded. Returns the storageID and nil on success, or an error if no recyclable ID exists.

func (*CacheDB) GetAllUsage 

func (cdb *CacheDB) GetAllUsage() (map[StorageUsageKey]int64, error)

GetAllUsage returns usage for all storage+namespace combinations

func (*CacheDB) GetAppendIntent 

func (cdb *CacheDB) GetAppendIntent(instanceHash InstanceHash) (*AppendIntent, error)

GetAppendIntent returns the append-in-flight record for an object version, or nil when there is none.

func (*CacheDB) GetBlockState 

func (cdb *CacheDB) GetBlockState(instanceHash InstanceHash) (*roaring.Bitmap, error)

GetBlockState retrieves the bitmap of downloaded blocks for a file.

If the block-state key is absent but the object's metadata indicates a completed download, a fully-populated bitmap is returned. This allows callers to treat completed objects uniformly without requiring a separate completion check. The block-state key is removed on completion to save database space (see BlockWriter.Close).

func (*CacheDB) GetDirUsage 

func (cdb *CacheDB) GetDirUsage(storageID StorageID) (map[NamespaceID]int64, error)

GetDirUsage returns usage for all namespaces within a single storage directory.

func (*CacheDB) GetDownloadedBlockCount 

func (cdb *CacheDB) GetDownloadedBlockCount(instanceHash InstanceHash) (uint64, error)

GetDownloadedBlockCount returns the number of downloaded blocks

func (*CacheDB) GetEncryptionManager 

func (cdb *CacheDB) GetEncryptionManager() *EncryptionManager

GetEncryptionManager returns the encryption manager

func (*CacheDB) GetInlineData 

func (cdb *CacheDB) GetInlineData(instanceHash InstanceHash) ([]byte, error)

GetInlineData retrieves encrypted inline data for a small file

func (*CacheDB) GetLatestETag 

func (cdb *CacheDB) GetLatestETag(objectHash ObjectHash) (string, bool, error)

GetLatestETag retrieves the latest ETag for an object. Returns (etag, found, err). An object cached without an ETag will return ("", true, nil); an object not in the cache returns ("", false, nil).

func (*CacheDB) GetMetadata 

func (cdb *CacheDB) GetMetadata(instanceHash InstanceHash) (*CacheMetadata, error)

GetMetadata retrieves cache metadata for a file

func (*CacheDB) GetTierUploadIntent 

func (cdb *CacheDB) GetTierUploadIntent(instanceHash InstanceHash) (*TierUploadIntent, error)

GetTierUploadIntent returns an object's upload intent, or nil if it has none. An unreadable intent is returned as an error.

func (*CacheDB) GetUsage 

func (cdb *CacheDB) GetUsage(storageID StorageID, namespaceID NamespaceID) (int64, error)

GetUsage retrieves the total bytes used by a storage+namespace combination. If a MergeOperator is active for the key, its Get() method is used to replay all accumulated deltas; otherwise a raw read is performed.

func (*CacheDB) HasMetadata 

func (cdb *CacheDB) HasMetadata(instanceHash InstanceHash) (bool, error)

HasMetadata checks if metadata exists for a file

func (*CacheDB) InstanceHash 

func (cdb *CacheDB) InstanceHash(etag string, objectHash ObjectHash) InstanceHash

InstanceHash computes the salted SHA-256 hash for (etag, objectHash).

func (*CacheDB) IsBlockDownloaded 

func (cdb *CacheDB) IsBlockDownloaded(instanceHash InstanceHash, blockNum uint32) (bool, error)

IsBlockDownloaded checks if a specific block has been downloaded

func (*CacheDB) IsPurgeFirst 

func (cdb *CacheDB) IsPurgeFirst(instanceHash InstanceHash) (bool, error)

IsPurgeFirst checks if a file hash is marked for priority eviction

func (*CacheDB) ListTierUploadIntents 

func (cdb *CacheDB) ListTierUploadIntents(after InstanceHash, limit int) ([]TierUploadIntentEntry, error)

ListTierUploadIntents returns up to limit upload intents whose hashes sort after the given one ("" to start from the beginning), in hash order. Pass the last hash returned to fetch the next page. Intents that fail to unmarshal are skipped with a warning.

func (*CacheDB) LoadDiskMappings 

func (cdb *CacheDB) LoadDiskMappings() ([]DiskMapping, error)

LoadDiskMappings loads all persisted disk mappings.

func (*CacheDB) LoadNamespaceMappings 

func (cdb *CacheDB) LoadNamespaceMappings() (map[string]NamespaceID, NamespaceID, error)

LoadNamespaceMappings loads all persisted namespace mappings and returns them as a map[prefix]->id, along with the highest ID seen (so the caller can resume the counter).

func (*CacheDB) MarkBlocksDownloaded 

func (cdb *CacheDB) MarkBlocksDownloaded(instanceHash InstanceHash, startBlock, endBlock uint32, storageID StorageID, namespaceID NamespaceID, contentLength int64) error

MarkBlocksDownloaded marks specific blocks as downloaded and atomically updates usage statistics based on the number of newly-added blocks. Usage tracking requires metadata to be set for the instanceHash; if metadata is not yet available, the bitmap is still updated but usage tracking is skipped.

func (*CacheDB) MarkPurgeFirst 

func (cdb *CacheDB) MarkPurgeFirst(instanceHash InstanceHash) error

MarkPurgeFirst marks a file hash for priority eviction

func (*CacheDB) MergeBlockStateWithUsage 

func (cdb *CacheDB) MergeBlockStateWithUsage(instanceHash InstanceHash, newBlocks *roaring.Bitmap, storageID StorageID, namespaceID NamespaceID, contentLength int64) error

MergeBlockStateWithUsage atomically merges new blocks into the existing bitmap AND updates usage statistics based on the number of newly-enabled bits.

contentLength controls how the usage delta is calculated:

  • If >= 0, the supplied contentLength, storageID, and namespaceID are used directly, avoiding a metadata DB read.
  • If < 0 (typically -1), the metadata is read within the transaction to obtain the content length, storage ID, and namespace ID.

The method retries on BadgerDB transaction conflicts, which can occur when multiple concurrent block fetchers write to the same object's bitmap.

func (*CacheDB) MergeExistingMetadata 

func (cdb *CacheDB) MergeExistingMetadata(instanceHash InstanceHash, incoming *CacheMetadata) (bool, error)

MergeExistingMetadata merges incoming into an object's record as MergeMetadata does, but only if the record exists; it reports whether it did. It is for a writer that only annotates an object -- the consistency scan recording what it verified -- and must not make a bare record for one deleted meanwhile, which a later miss would take for a cached copy.

func (*CacheDB) MergeMetadata 

func (cdb *CacheDB) MergeMetadata(instanceHash InstanceHash, incoming *CacheMetadata) error

MergeMetadata performs an atomic read-modify-update of the metadata for instanceHash. If no metadata exists yet, incoming is written as-is (initial creation).

Field-level merge rules:

  • Max-time (LastModified, LastValidated, LastAccessTime, Expires, Completed): keep the later of existing vs incoming.
  • Additive (Checksums): union by ChecksumType; if both sides provide the same Type, prefer the OriginVerified entry.
  • Last-writer-wins (ContentType, ContentLength, VaryHeaders, CCFlags, CCMaxAge): incoming replaces existing when the incoming value is non-zero / non-empty.
  • Set-once (ETag, SourceURL, DataKey, StorageID, NamespaceID): may transition from zero-value to a value, but changing a non-zero value to a different non-zero value returns an error. ETag is set-once because it is part of the instance hash; a changed ETag produces a different instance.

The method retries on BadgerDB transaction conflicts, which can occur when multiple concurrent callers merge metadata for the same instance (e.g. concurrent range-on-miss initialization via initObjectFromStat).

When no record exists, MergeMetadata creates one from incoming; see MergeExistingMetadata for a merge that never does.

func (*CacheDB) NewBatch 

func (cdb *CacheDB) NewBatch() *Batch

NewBatch creates a new write batch.

A batch taken from a read-only handle refuses at Set, Delete, and Flush rather than here, because there is no error to return from this call.

func (*CacheDB) ObjectHash 

func (cdb *CacheDB) ObjectHash(pelicanURL string) ObjectHash

ObjectHash computes the salted SHA-256 hash for a pelican URL.

func (*CacheDB) PurgeStorageID 

func (cdb *CacheDB) PurgeStorageID(storageID StorageID) error

PurgeStorageID removes all database entries associated with a storageID: object metadata, block state, inline data, LRU entries, purge-first markers, ETag entries, usage counters, and the disk mapping itself.

This is used during storage ID recycling to reclaim an ID that was previously assigned to a directory that is no longer mounted.

Objects are deleted in batches to avoid exceeding BadgerDB's transaction size limit.

func (*CacheDB) ReadOnly 

func (cdb *CacheDB) ReadOnly() bool

ReadOnly reports whether this handle refuses mutations.

func (*CacheDB) RecordRedirectIssued 

func (cdb *CacheDB) RecordRedirectIssued(instanceHash InstanceHash) error

RecordRedirectIssued stamps the current time on an object's redirect-hold key. Objects with a stamp newer than the configured redirect hold window are skipped by eviction, so a client holding a fresh pre-signed URL never has the object deleted out from under it.

The write is debounced so that a hot object being redirected repeatedly does not stamp the same key on every request. The debounce window is bounded by how much slack the hold has over a URL's lifetime: skipping a write leaves the stamp up to one window old, so the protection a caller can still count on is hold-minus-window, which must remain at least as long as the URL it just handed out. redirectHoldHeadroom is the slack the cache guarantees when it resolves the hold, and a quarter of the hold is the cheaper bound, so the debounce is the smaller of the two.

func (*CacheDB) ReloadSalt 

func (cdb *CacheDB) ReloadSalt() error

ReloadSalt re-reads the hash salt from the database.

The salt is cached at open because it is on the hot path, which is fine until something replaces the database's contents underneath it. Restoring a backup does exactly that: the restored salt is the one every instance hash in that catalog was derived from, so a consumer still holding the salt generated at open computes different hashes and finds none of its objects.

func (*CacheDB) RelocateObject 

func (cdb *CacheDB) RelocateObject(instanceHash InstanceHash, newStorageID StorageID, remote *TierObjectInfo) (*CacheMetadata, error)

RelocateObject atomically moves a completed object's metadata onto the given tiering target. The StorageID is rewritten (bypassing the set-once merge rule — this is the one sanctioned transition) and the LRU index entry is moved from the object's base storage ID to the new one.

Chunked objects are supported: because a tiering target holds the object as a single contiguous blob, relocation flattens the chunk layout (ChunkSizeCode / ChunkLocations are cleared). The returned pre-relocation metadata still carries the original chunk layout so the caller can delete every local chunk file and refund each contributing directory's usage.

The caller is responsible for moving the object *data* and for adjusting usage counters after the transaction commits (AddUsage cannot run inside a transaction). Returns the pre-relocation metadata.

func (*CacheDB) Salt 

func (cdb *CacheDB) Salt() []byte

Salt returns the per-database random salt used for hashing.

The returned slice is shared with every other caller and with the database itself; callers must treat it as read-only. Nothing in the tree writes through it -- both consumers hand it straight to hmac.New, which copies -- and a caller that did would corrupt every hash computed concurrently.

func (*CacheDB) SaveDiskMapping 

func (cdb *CacheDB) SaveDiskMapping(dm DiskMapping) error

SaveDiskMapping persists a single storageID → (UUID, directory) mapping.

func (*CacheDB) ScanAppendIntents 

func (cdb *CacheDB) ScanAppendIntents(fn func(InstanceHash, AppendIntent) error) error

ScanAppendIntents calls fn for every append-in-flight record. Returning an error from fn stops the scan and is returned to the caller.

func (*CacheDB) ScanMetadata 

func (cdb *CacheDB) ScanMetadata(fn func(instanceHash InstanceHash, meta *CacheMetadata) error) error

ScanMetadata iterates over all metadata entries

func (*CacheDB) ScanMetadataFrom 

func (cdb *CacheDB) ScanMetadataFrom(startKey InstanceHash, fn func(instanceHash InstanceHash, meta *CacheMetadata) error) error

ScanMetadataFrom scans metadata starting from the given instanceHash (empty string = start from beginning)

func (*CacheDB) SetAppendIntent 

func (cdb *CacheDB) SetAppendIntent(instanceHash InstanceHash, intent AppendIntent) error

SetAppendIntent records that a streaming append is in flight for the given object version. See PrefixAppendIntent for why the record exists.

func (*CacheDB) SetBlockState 

func (cdb *CacheDB) SetBlockState(instanceHash InstanceHash, bitmap *roaring.Bitmap) error

SetBlockState stores the bitmap of downloaded blocks

func (*CacheDB) SetInlineData 

func (cdb *CacheDB) SetInlineData(instanceHash InstanceHash, encryptedData []byte) error

SetInlineData stores encrypted inline data for a small file. The caller (StoreInline) is responsible for usage accounting via ChargeUsage; this function does NOT adjust usage counters.

func (*CacheDB) SetLatestETag 

func (cdb *CacheDB) SetLatestETag(objectHash ObjectHash, etag string, observedAt time.Time) error

SetLatestETag stores the latest ETag for an object, but only if observedAt is more recent than the already-stored timestamp. This prevents a slow download that finishes late from clobbering a newer ETag written by a more recent request.

func (*CacheDB) SetMetadata 

func (cdb *CacheDB) SetMetadata(instanceHash InstanceHash, meta *CacheMetadata) error

SetMetadata stores cache metadata for a file, unconditionally replacing any previously stored metadata. Use this only for initial creation of a metadata entry (e.g. InitDiskStorage, StoreInline); for subsequent updates prefer MergeMetadata which applies field-level merge semantics.

func (*CacheDB) SetNamespaceMapping 

func (cdb *CacheDB) SetNamespaceMapping(prefix string, id NamespaceID) error

SetNamespaceMapping persists the mapping from a namespace prefix string to a numeric ID. This ensures the IDs survive restarts so that LRU keys and usage counters remain valid.

func (*CacheDB) SetTierUploadIntent 

func (cdb *CacheDB) SetTierUploadIntent(instanceHash InstanceHash, intent *TierUploadIntent) error

SetTierUploadIntent records an in-progress upload to a tiering target. Written before the first byte reaches the bucket so that a crash never leaves an untracked object on the target.

This and DeleteTierUploadIntent perform a single blind Set/Delete with no prior read, so their transaction read-set is empty and BadgerDB's serializable-snapshot conflict detection can never flag them — no retry loop is required (unlike RecordRedirectIssued or RelocateObject, which read before writing).

func (*CacheDB) SetUsage 

func (cdb *CacheDB) SetUsage(storageID StorageID, namespaceID NamespaceID, value int64) error

SetUsage sets the absolute usage counter for a (storageID, namespaceID) pair. Any active MergeOperator for the key is stopped first (flushing pending deltas) and then the value is overwritten.

The entry is written with WithDiscard() so that BadgerDB marks all earlier versions of the key as eligible for garbage collection. Without this flag, the MergeOperator's iterateAndMerge would still see the old compacted entry (which carries bitDiscardEarlierVersions) and sum it into the total, causing the counter to include both the new baseline and the old accumulated value — a compounding overcount that grows with every reconciliation cycle.

The next AddUsage call will lazily create a fresh MergeOperator.

func (*CacheDB) StartGC 

func (cdb *CacheDB) StartGC(ctx context.Context, egrp *errgroup.Group)

StartGC starts the background garbage collection goroutine.

Value-log GC rewrites the value log, so a read-only handle starts nothing.

func (*CacheDB) UnmarkPurgeFirst 

func (cdb *CacheDB) UnmarkPurgeFirst(instanceHash InstanceHash) error

UnmarkPurgeFirst removes the purge first marker for a file hash

func (*CacheDB) UpdateLRU 

func (cdb *CacheDB) UpdateLRU(instanceHash InstanceHash, debounceTime time.Duration) error

UpdateLRU updates the LRU access time for a file Uses debouncing: only updates if last access was more than debounceTime ago This is optimized to avoid iteration by storing the last access time in metadata

func (*CacheDB) VerifySchemaVersion 

func (cdb *CacheDB) VerifySchemaVersion() error

VerifySchemaVersion is the read-only counterpart of ensureSchemaVersion: it refuses a database written under a newer schema but never writes, so it can run inside a read-only BadgerDB handle (see OpenCacheDBReadOnly).

A database with no version key is accepted as CurrentSchemaVersion for the same reason ensureSchemaVersion adopts one, minus the stamp — introspecting a cache must not require write access to it. An older version is likewise accepted: the writable open is where migration happens, and refusing to *look* at a database that a normal start would quietly upgrade helps nobody.

It is exported for the same reason EnsureStoreMode is: a consumer that replaces this database's contents wholesale — restoring a backup catalog into it — has a database whose header keys came from wherever the backup did and no longer describe what was checked at open, and should re-verify.

type CacheDirectives 

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

CacheDirectives holds parsed Cache-Control header values that are relevant to the persistent cache.

Boolean directives are packed into a bitfield; use the accessor methods (NoStore, NoCache, etc.) instead of reading fields directly.

max-age and s-maxage are merged into a single freshness lifetime: when both are present the maximum of the two is kept, because the Pelican local cache does not distinguish shared from private.

func ParseCacheControl 

func ParseCacheControl(header string) CacheDirectives

ParseCacheControl parses a Cache-Control header value into structured directives. The parsing is case-insensitive for directive names, as required by RFC 7234. When both max-age and s-maxage are present, the maximum of the two durations is kept (the Pelican local cache does not distinguish shared from private).

func (*CacheDirectives) Flags 

func (cd *CacheDirectives) Flags() uint8

Flags returns the raw packed bitfield. Callers should prefer the named accessors; this is provided for efficient serialization into CacheMetadata.CCFlags.

func (*CacheDirectives) Freshness 

func (cd *CacheDirectives) Freshness() (time.Duration, bool)

Freshness returns the merged freshness lifetime and whether it was set.

func (*CacheDirectives) HasDirectives 

func (cd *CacheDirectives) HasDirectives() bool

HasDirectives returns true if any Cache-Control directive was parsed.

func (*CacheDirectives) HasFreshness 

func (cd *CacheDirectives) HasFreshness() bool

HasFreshness returns true if the origin provided explicit freshness information (max-age or s-maxage).

func (*CacheDirectives) IsStale 

func (cd *CacheDirectives) IsStale(lastValidated time.Time) bool

IsStale checks whether a cached response is stale given when it was last validated (or originally stored). If no freshness information is available (no max-age / s-maxage / no-cache), applies the default cache policy from params with jitter.

func (*CacheDirectives) IsStaleWithDefaults 

func (cd *CacheDirectives) IsStaleWithDefaults(lastValidated time.Time) bool

IsStaleWithDefaults checks staleness using configured default values and jitter. This is used when the origin doesn't provide explicit Cache-Control headers.

func (*CacheDirectives) MaxAge 

func (cd *CacheDirectives) MaxAge() time.Duration

func (*CacheDirectives) MaxAgeSet 

func (cd *CacheDirectives) MaxAgeSet() bool

func (*CacheDirectives) MustRevalidate 

func (cd *CacheDirectives) MustRevalidate() bool

func (*CacheDirectives) NoCache 

func (cd *CacheDirectives) NoCache() bool

func (*CacheDirectives) NoStore 

func (cd *CacheDirectives) NoStore() bool

func (*CacheDirectives) Private 

func (cd *CacheDirectives) Private() bool

func (*CacheDirectives) ShouldStore 

func (cd *CacheDirectives) ShouldStore() bool

ShouldStore returns true if the response is allowed to be stored in the cache.

type CacheMetadata 

type CacheMetadata struct {
	// Validation fields
	ETag          string     `msgpack:"etag"`         // HTTP ETag header
	LastModified  time.Time  `msgpack:"lm"`           // HTTP Last-Modified header
	Expires       time.Time  `msgpack:"exp"`          // HTTP Expires header
	LastValidated time.Time  `msgpack:"lv"`           // When we last validated with origin
	Completed     time.Time  `msgpack:"c"`            // When download was completed
	Checksums     []Checksum `msgpack:"ck,omitempty"` // Object checksums
	DataVerified  time.Time  `msgpack:"dv,omitempty"` // When the on-disk data was last read back and checksum-verified

	// Identification fields
	ContentType   string   `msgpack:"ct"`            // MIME type
	ContentLength int64    `msgpack:"cl"`            // Total object size in bytes
	VaryHeaders   []string `msgpack:"vh,omitempty"`  // Headers that affect caching
	SourceURL     string   `msgpack:"url,omitempty"` // Original URL including federation

	// Cache-Control directives (efficient packed representation)
	CCFlags  uint8 `msgpack:"ccf,omitempty"`  // Bitset: 0x01=no-store, 0x02=no-cache, 0x04=private, 0x08=must-revalidate
	CCMaxAge int32 `msgpack:"ccma,omitempty"` // Merged max-age/s-maxage freshness lifetime (seconds, 0 = not set, max of both if both specified)

	// Storage fields.
	// StorageID encodes both location type and directory identity:
	//   0         = inline (data stored directly in BadgerDB)
	//   1 .. 255  = disk-backed (directory identified by this ID)
	// For chunked objects, this is the location of chunk 0.
	StorageID StorageID `msgpack:"sid"`
	DataKey   []byte    `msgpack:"key"` // Encrypted DEK (Data Encryption Key)

	// Chunking fields for large objects spread across multiple storage directories.
	// ChunkSizeCode encodes the chunk size (0 = chunking disabled, see chunking.go).
	// ChunkLocations stores the StorageID for chunks 1, 2, ... (chunk 0 uses StorageID above).
	ChunkSizeCode  ChunkSizeCode   `msgpack:"csc,omitempty"` // 0 = disabled, see ChunkSizeCodeToBytes()
	ChunkLocations []ChunkLocation `msgpack:"chl,omitempty"` // Locations for chunks after chunk 0

	// Namespace and storage tracking for fairness-aware eviction
	NamespaceID NamespaceID `msgpack:"ns"` // ID of the namespace prefix

	// LRU tracking
	LastAccessTime time.Time `msgpack:"la"` // Last access time for LRU index

	// Remote describes the copy a tiering target holds, as the target
	// reported it at upload: its entity tag, version and size.  Nil for an
	// object that is not tiered.  Reads are pinned to it and the integrity
	// scan checks the target against it; see TierObjectInfo.
	Remote *TierObjectInfo `msgpack:"rmt,omitempty"`
}

CacheMetadata stores all metadata about a cached object Serialized using MessagePack for efficiency

Merge semantics (see CacheDB.MergeMetadata)

Fields are classified into groups that govern how concurrent updates are reconciled:

  • Max-time: LastModified, LastValidated, LastAccessTime, Expires, Completed, DataVerified — only advance forward (keep the later timestamp).
  • Additive: Checksums — union by algorithm; prefer OriginVerified.
  • Last-writer-wins: ContentType, ContentLength, VaryHeaders, CCFlags, CCMaxAge — the incoming value always replaces the old one.
  • Set-once: ETag, SourceURL, DataKey, StorageID, NamespaceID, ChunkSizeCode, ChunkLocations — may transition from zero-value to set, but changing a non-zero value to a different non-zero value is an error. ETag is set-once because it is part of the instance hash; a changed ETag produces a different instance. Chunking fields are set-once because changing them would invalidate existing chunk files.

func (*CacheMetadata) AllStorageIDs 

func (m *CacheMetadata) AllStorageIDs() []StorageID

AllStorageIDs returns a deduplicated list of all StorageIDs used by this object. For non-chunked objects, returns just the base StorageID.

func (*CacheMetadata) ChunkCount 

func (m *CacheMetadata) ChunkCount() int

ChunkCount returns the number of chunks for this object. Returns 1 for non-chunked objects.

func (*CacheMetadata) ComputeExpires 

func (m *CacheMetadata) ComputeExpires() time.Time

ComputeExpires returns the absolute time at which this object expires. It uses the origin-supplied max-age when available, otherwise the configured default freshness policy. The result is based on LastValidated (or Completed if LastValidated is zero).

func (*CacheMetadata) EnsureExpires 

func (m *CacheMetadata) EnsureExpires()

EnsureExpires sets the Expires field if it is currently zero. Call this after SetCacheControl and after LastValidated is set.

func (*CacheMetadata) GetCacheControlHeader 

func (m *CacheMetadata) GetCacheControlHeader() string

GetCacheControlHeader reconstructs the Cache-Control header string for HTTP responses. It returns the origin's directives verbatim when present; when the origin did not specify any Cache-Control, it returns "" (callers should use ResponseCacheControl instead to get a header that reflects the default policy).

func (*CacheMetadata) GetCacheDirectives 

func (m *CacheMetadata) GetCacheDirectives() CacheDirectives

GetCacheDirectives returns the parsed cache directives

func (*CacheMetadata) GetChunkInfo 

func (m *CacheMetadata) GetChunkInfo() []ChunkInfo

GetChunkInfo returns information about all chunks for this object.

func (*CacheMetadata) GetChunkStorageID 

func (m *CacheMetadata) GetChunkStorageID(chunkIndex int) StorageID

GetChunkStorageID returns the StorageID for a specific chunk (0-indexed). Chunk 0 uses the base StorageID, chunks 1+ use ChunkLocations. Returns StorageIDInline (0) for unallocated chunks (lazy allocation).

func (*CacheMetadata) IsChunkAllocated 

func (m *CacheMetadata) IsChunkAllocated(chunkIndex int) bool

IsChunkAllocated returns true if the specified chunk has a storage ID assigned. For chunked objects, StorageIDInline (0) means the chunk is not yet allocated. For non-chunked objects, always returns true (single storage).

func (*CacheMetadata) IsChunked 

func (m *CacheMetadata) IsChunked() bool

IsChunked returns true when the object is stored across multiple chunk files.

func (*CacheMetadata) IsDisk 

func (m *CacheMetadata) IsDisk() bool

IsDisk returns true when the object data is (or will be) stored on disk. Chunked objects are always disk-based, even when chunk 0 has not yet been allocated.

func (*CacheMetadata) IsInline 

func (m *CacheMetadata) IsInline() bool

IsInline returns true when the object data is stored directly in BadgerDB. A chunked object is never inline, even if its base StorageID is still 0 (unallocated).

func (*CacheMetadata) PerDirectoryBytes 

func (m *CacheMetadata) PerDirectoryBytes() map[StorageID]int64

PerDirectoryBytes returns a map from StorageID to the number of content bytes that live in each storage directory. For non-chunked objects the entire ContentLength is attributed to the base StorageID. For chunked objects the byte count is split according to each chunk's assigned directory. Unallocated chunks (StorageID 0) are skipped.

func (*CacheMetadata) ResponseCacheControl 

func (m *CacheMetadata) ResponseCacheControl() string

ResponseCacheControl returns the Cache-Control header value the cache should send to downstream clients. When the origin specified directives, those are forwarded. When it did not, the cache advertises the remaining freshness lifetime (derived from LocalCache_DefaultMaxAge + jitter) as max-age so that downstream clients can cache the response without re-contacting the cache until revalidation is due.

func (*CacheMetadata) SetCacheControl 

func (m *CacheMetadata) SetCacheControl(header string)

SetCacheControl parses a Cache-Control header and stores the directives efficiently

func (*CacheMetadata) SetChunkStorageID 

func (m *CacheMetadata) SetChunkStorageID(chunkIndex int, storageID StorageID)

SetChunkStorageID sets the StorageID for a specific chunk (0-indexed). For chunk 0, sets the base StorageID. For chunks 1+, sets ChunkLocations. The ChunkLocations slice must be pre-allocated to the correct size.

type CacheMode 

type CacheMode int

CacheMode distinguishes whether the persistent cache is running as a full cache server (CacheType) or as a client-side local cache (LocalCacheType). The mode determines which configuration namespace is consulted for defaults (Cache.* vs LocalCache.*). 

const (
	// CacheModeLocal is the default: use LocalCache.* params.
	CacheModeLocal CacheMode = iota
	// CacheModeServer uses Cache.* params (cache server).
	CacheModeServer
)

type CacheStats 

type CacheStats struct {
	TotalInlineBytes     int64                       `json:"total_inline_bytes"`
	TotalMetadataEntries int64                       `json:"total_metadata_entries"`
	TotalBytesMetadata   int64                       `json:"total_bytes_metadata"`     // Sum of ContentLength from metadata entries
	UsageCounters        map[string]int64            `json:"usage_counters,omitempty"` // Pre-computed usage from u: prefix keys
	StorageBreakdown     map[string]*StorageDirStats `json:"storage_breakdown,omitempty"`
	DirPaths             map[uint8]string            `json:"dir_paths,omitempty"`       // StorageID → directory path
	NamespaceNames       map[uint32]string           `json:"namespace_names,omitempty"` // NamespaceID → prefix string
}

CacheStats contains aggregate size statistics about the cache.

type ChaosInjector 

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

ChaosInjector injects corruption into a running cache's already-open database and storage, for fault-injection ("chaos") testing of the cache's integrity-detection paths.

BadgerDB is single-process, so corruption must be performed in-process by the cache server itself (a CLI cannot open the database while the server holds it). The injector therefore wraps the live handles and does not own them — there is nothing to close.

func NewChaosInjector 

func NewChaosInjector(db *CacheDB, storage *StorageManager) *ChaosInjector

NewChaosInjector wraps a live cache's database and storage manager.

func (*ChaosInjector) CorruptBlock 

func (ci *ChaosInjector) CorruptBlock(objectURL, etag, instanceHash string, blockNum uint32, numBytes int) (*ChaosResult, error)

CorruptBlock flips the first numBytes bytes of the on-disk (encrypted) representation of the given block. This makes the block's AES-GCM authentication tag fail to validate, which the cache detects on the next read of that block or during the periodic data-integrity scan.

blockNum is a global, zero-based block number. numBytes <= 0 defaults to the authentication-tag size (the minimum needed to guarantee detection).

Detection is not necessarily immediate: a block whose plaintext is still in the cache's in-memory caches will continue to read successfully until those entries are evicted; the corruption is caught on the next cold read of the block, the periodic data-integrity scan, or after a restart.

func (*ChaosInjector) TruncateObject 

func (ci *ChaosInjector) TruncateObject(objectURL, etag, instanceHash string, chunkIndex int, dropBytes int64) (*ChaosResult, error)

TruncateObject removes dropBytes bytes from the end of one of the object's on-disk chunk files (the last chunk by default, when chunkIndex < 0). This drops trailing block(s), which the cache detects on a cold read or during the data scan. dropBytes <= 0 defaults to a single encrypted block (BlockTotalSize).

type ChaosResult 

type ChaosResult struct {
	InstanceHash string `json:"instance_hash"`
	SourceURL    string `json:"source_url,omitempty"`
	ETag         string `json:"etag,omitempty"`
	Operation    string `json:"operation"` // "corrupt-block" or "truncate"
	ChunkIndex   int    `json:"chunk_index"`
	FilePath     string `json:"file_path"`
	BlockNum     int64  `json:"block_num,omitempty"` // global block number (corrupt-block)
	DiskOffset   int64  `json:"disk_offset"`
	BytesChanged int    `json:"bytes_changed,omitempty"`
	OldFileSize  int64  `json:"old_file_size"`
	NewFileSize  int64  `json:"new_file_size"`
}

ChaosResult describes a corruption injected into a cached object by the chaos-testing helpers. It is intended for fault-injection testing of the cache's integrity scan and read-time corruption detection.

type Checksum 

type Checksum struct {
	Type            ChecksumType `msgpack:"t"`
	Value           []byte       `msgpack:"v"`
	OriginVerified  bool         `msgpack:"ov"`           // True if checksum came from origin
	VerifyAttempted bool         `msgpack:"va,omitempty"` // True if we tried to get origin checksum
}

Checksum holds a checksum type and its value

func ParseChecksumHeader 

func ParseChecksumHeader(headerValue string, headerType string) *Checksum

ParseChecksumHeader parses a checksum from HTTP headers

type ChecksumInfo 

type ChecksumInfo struct {
	Type           string `json:"type"`
	Value          string `json:"value"` // Hex-encoded
	OriginVerified bool   `json:"origin_verified"`
}

ChecksumInfo describes a stored checksum.

type ChecksumStatus 

type ChecksumStatus struct {
	Type     string `json:"type"`
	Expected string `json:"expected"` // Hex-encoded
	Computed string `json:"computed"` // Hex-encoded (only if verification ran)
	Match    bool   `json:"match"`
}

ChecksumStatus describes the verification status of a single checksum.

type ChecksumType 

type ChecksumType uint8

ChecksumType identifies the type of checksum 

const (
	ChecksumMD5    ChecksumType = 0
	ChecksumSHA1   ChecksumType = 1
	ChecksumSHA256 ChecksumType = 2
	ChecksumCRC32  ChecksumType = 3
	ChecksumCRC32C ChecksumType = 4
)

func ParseChecksumAlgorithm 

func ParseChecksumAlgorithm(name string) (ChecksumType, bool)

ParseChecksumAlgorithm maps the spellings clients use in Want-Digest onto a checksum type. The second result reports whether the name was recognized.

type ChunkInfo 

type ChunkInfo struct {
	Index       int       // 0-based chunk index
	StorageID   StorageID // Storage directory for this chunk
	StartOffset int64     // Content byte offset where this chunk starts
	EndOffset   int64     // Content byte offset where this chunk ends (inclusive)
	Size        int64     // Size of this chunk in bytes
}

ChunkInfo holds computed information about a chunk.

func GetChunkInfo 

func GetChunkInfo(contentLength int64, chunkSizeCode ChunkSizeCode, chunkLocations []ChunkLocation, baseStorageID StorageID) []ChunkInfo

GetChunkInfo returns information about all chunks for an object. chunkLocations should be the ChunkLocations from CacheMetadata (may be nil/empty for non-chunked objects). baseStorageID is the StorageID from CacheMetadata (where chunk 0 is stored).

type ChunkInfoSummary 

type ChunkInfoSummary struct {
	ChunkSizeBytes int64   `json:"chunk_size_bytes"`
	ChunkCount     int     `json:"chunk_count"`
	ChunkLocations []uint8 `json:"chunk_locations"` // StorageID per chunk
}

ChunkInfoSummary describes chunking configuration for large objects.

type ChunkLocation 

type ChunkLocation struct {
	StorageID StorageID `msgpack:"s"` // Which storage directory holds this chunk
}

ChunkLocation stores the storage location of a single chunk. Chunk 0 is always stored at the StorageID in CacheMetadata. Additional chunks (1, 2, ...) are stored in the ChunkLocations slice.

type ChunkNotification 

type ChunkNotification struct {
	ChunkIndex int64     // Which chunk this notification is for
	Completed  bool      // True if this chunk is complete
	Error      error     // Non-nil if an error occurred
	ETA        time.Time // Estimated time of completion (updated atomically)
}

ChunkNotification contains information about a chunk's completion status

type ChunkSizeCode 

type ChunkSizeCode uint8

ChunkSizeCode is a compact uint8 encoding of chunk sizes for storage efficiency. The encoding is non-linear to cover a wide range from 2MB to ~57GB:

  • 0: Chunking disabled (object stored in a single file)
  • 1-6: Doubling: 2^n MB (2, 4, 8, 16, 32, 64 MB)
  • 7-21: 64 MB increments starting at 128 MB (128, 192, ..., 1024 MB)
  • 22-53: 128 MB increments starting at 1152 MB (1152, 1280, ..., 5120 MB)
  • 54-255: 256 MB increments starting at 5376 MB (5376, 5632, ..., ~57 GB)
const (
	// ChunkingDisabled indicates the object is stored in a single file
	ChunkingDisabled ChunkSizeCode = 0
)

func BytesToChunkSizeCode 

func BytesToChunkSizeCode(size uint64) ChunkSizeCode

BytesToChunkSizeCode converts a byte size to the nearest ChunkSizeCode that is >= the requested size (rounds up to ensure chunks can hold the data). Returns ChunkingDisabled (0) if size is 0.

func ParseChunkSize 

func ParseChunkSize(s string) (ChunkSizeCode, error)

ParseChunkSize parses a human-readable chunk size string (e.g., "64MB", "2GB") and returns the corresponding ChunkSizeCode.

type ConsistencyCheckResult 

type ConsistencyCheckResult struct {
	MetadataScanRan    bool   `json:"metadata_scan_ran"`
	DataScanRan        bool   `json:"data_scan_ran"`
	OrphanedFiles      int64  `json:"orphaned_files"`
	OrphanedDBEntries  int64  `json:"orphaned_db_entries"`
	ChecksumMismatches int64  `json:"checksum_mismatches"`
	BytesVerified      int64  `json:"bytes_verified"`
	ObjectsVerified    int64  `json:"objects_verified"`
	MetadataScanErrors int64  `json:"metadata_scan_errors"`
	DataScanErrors     int64  `json:"data_scan_errors"`
	Duration           string `json:"duration"`
	Error              string `json:"error,omitempty"`
}

ConsistencyCheckResult contains the result of a consistency check run.

type ConsistencyChecker 

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

ConsistencyChecker verifies cache consistency between database and disk. For multi-directory configurations, it scans each storage directory independently.

func NewConsistencyChecker 

func NewConsistencyChecker(db *CacheDB, storage *StorageManager, config ConsistencyConfig) *ConsistencyChecker

NewConsistencyChecker creates a new consistency checker.

func (*ConsistencyChecker) GetStats 

func (cc *ConsistencyChecker) GetStats() ConsistencyStats

GetStats returns current statistics

func (*ConsistencyChecker) RunDataScan 

func (cc *ConsistencyChecker) RunDataScan(ctx context.Context, progressCh chan<- ScanProgressEvent) error

RunDataScan performs a full data integrity scan It verifies checksums of stored objects using block-by-block reading

func (*ConsistencyChecker) RunMetadataScan 

func (cc *ConsistencyChecker) RunMetadataScan(ctx context.Context, progressCh chan<- ScanProgressEvent) error

RunMetadataScan performs a metadata consistency scan. It verifies that database entries match files on disk and vice versa.

func (*ConsistencyChecker) RunTierScan 

func (cc *ConsistencyChecker) RunTierScan(ctx context.Context) error

RunTierScan reconciles each tiering target's bucket contents against the metadata store:

  • objects in the bucket with no tiered metadata (and no in-flight upload intent) are orphans and deleted from the bucket;
  • tiered metadata whose remote object is missing (or no longer matches the copy that was uploaded) is removed so the object is re-fetched from the origin on the next request.

Byte-level usage for tiering storage IDs is reconciled by the regular metadata scan (usageDuringScan covers every metadata entry regardless of backend).

func (*ConsistencyChecker) Start 

func (cc *ConsistencyChecker) Start(ctx context.Context, egrp *errgroup.Group)

Start begins the background consistency checking goroutines

func (*ConsistencyChecker) Stop 

func (cc *ConsistencyChecker) Stop()

Stop stops the consistency checker

func (*ConsistencyChecker) VerifyBlockIntegrity 

func (cc *ConsistencyChecker) VerifyBlockIntegrity(instanceHash InstanceHash) ([]uint32, error)

VerifyBlockIntegrity verifies the integrity of individual blocks by delegating to the storage manager's IdentifyCorruptBlocks, which handles crypto setup, file access, and AES-GCM auth-tag verification.

func (*ConsistencyChecker) VerifyObject 

func (cc *ConsistencyChecker) VerifyObject(instanceHash InstanceHash) (bool, error)

VerifyObject verifies a single object's integrity. If checksums are present, it verifies them all in a single pass. If no checksums are present, it still reads all data to verify readability (i.e. that decryption tags are valid and all blocks are accessible). Returns (true, nil) if the object is valid, (false, nil) if corrupt.

type ConsistencyConfig 

type ConsistencyConfig struct {
	// MetadataScanActiveMs limits metadata scan to this many ms per second (default: 100)
	MetadataScanActiveMs int64
	// DataScanBytesPerSec limits data scan to this many bytes per second (default: 100MB)
	DataScanBytesPerSec int64
	// MinAgeForCleanup is the minimum age before an entry/file can be cleaned up (default: 5 minutes, 0 for tests)
	MinAgeForCleanup time.Duration
	// ChecksumTypes specifies which checksums to calculate and verify.
	// When empty, defaults to []ChecksumType{ChecksumSHA256}.
	ChecksumTypes []ChecksumType
	// SkipVerifiedData, when true, makes the data scan verify each object's
	// on-disk data exactly once (recording the checksum and comparing it
	// against the origin's reported value when available) and then skip that
	// object on subsequent scans.  Intended for storage backends that already
	// guarantee at-rest integrity (e.g. ZFS).  When false (default) every scan
	// re-verifies all objects.
	SkipVerifiedData bool
	// ResampleInterval, when SkipVerifiedData is true, makes the scan still
	// re-verify roughly 1 in N already-verified objects each cycle, so there is
	// a nonzero floor of at-rest corruption detection even in "once" mode.
	// 0 disables re-sampling (pure once).
	ResampleInterval int
	// PreserveCorruptObjects, when true, makes the data scan report at-rest
	// corruption without deleting the object.
	//
	// A cache must delete: the object is a copy, deleting it is how the next
	// request repairs it from the origin.  A primary store (pstore) has no
	// upstream, so deleting is the loss rather than the repair -- the block
	// file and its metadata are the only record that the object ever existed,
	// including its path, size, and checksums.  Set this for any store that is
	// the last copy; leave it false for a cache.
	PreserveCorruptObjects bool
	// SkipChecksumBackfill, when true, makes the data scan leave objects that
	// carry no recorded checksums alone rather than computing and storing them.
	//
	// The backfill is a metadata write, so a scan that sets it is not the
	// read-only pass its callers may believe it to be.  For a cache the write
	// is harmless; for a primary store it can persist a checksum derived from
	// data that is already corrupt, or race a write that has not yet recorded
	// its own checksums.
	SkipChecksumBackfill bool
	// SuppressCacheMetrics stops this checker from updating the
	// pelican_cache_*_scan_* collectors above.
	//
	// Those names describe a cache, and this checker has a second consumer:
	// pstore drives it to scan an *origin*.  Left on, an origin's integrity
	// scan is reported as cache activity -- a process serving no cache at all
	// publishes cache series, and one serving both adds the two together in
	// the same graph with no way to separate them.  A consumer that sets this
	// takes on publishing the equivalent figures under its own names; pstore
	// does, as pelican_pstore_data_scan_*.
	//
	// It does not change what the scan does, only what it reports.
	SuppressCacheMetrics bool
}

ConsistencyConfig holds configuration for the consistency checker

type ConsistencyStats 

type ConsistencyStats struct {
	LastMetadataScan   time.Time
	LastDataScan       time.Time
	MetadataScanErrors int64
	DataScanErrors     int64
	OrphanedFiles      int64
	OrphanedDBEntries  int64
	ChecksumMismatches int64
	BytesVerified      int64
	ObjectsVerified    int64
}

ConsistencyStats holds statistics from consistency checks

type DirDiskStat 

type DirDiskStat struct {
	StorageID uint8  `json:"storage_id"`
	Path      string `json:"path"`
	BytesUsed int64  `json:"bytes_used"`
	FileCount int64  `json:"file_count"`
}

DirDiskStat holds per-directory disk usage from walking the filesystem.

type DirEvictionStats 

type DirEvictionStats struct {
	MaxSize   uint64
	HighWater uint64
	LowWater  uint64
}

DirEvictionStats contains per-directory eviction statistics

type DiskMapping 

type DiskMapping struct {
	ID        StorageID `msgpack:"id"`
	UUID      string    `msgpack:"uuid"`
	Directory string    `msgpack:"dir"`
	Backend   string    `msgpack:"be,omitempty"`
}

DiskMapping stores the mapping of a storage ID to its directory path and UUID. The UUID file is dropped in the directory root so that directories can be remounted at different paths and re-associated.

For tiering targets (Backend == BackendTier), Directory holds a display URL (s3://<endpoint-host>/<bucket>/<prefix>) and the UUID is stored in an identity object inside the bucket, so the same re-association logic applies when a bucket is reconfigured under a different endpoint.

type DiskUsageResult 

type DiskUsageResult struct {
	TotalBytesOnDisk int64                   `json:"total_bytes_on_disk"`
	TotalFiles       int64                   `json:"total_files"`
	Directories      map[string]*DirDiskStat `json:"directories,omitempty"`
	Duration         string                  `json:"duration"` // How long the walk took
}

DiskUsageResult contains the result of an expensive disk walk.

type EncryptionManager 

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

EncryptionManager handles all encryption operations for the cache.

The masterKey is loaded (or generated) once during construction and is never mutated afterwards. Two child keys are derived from it via HKDF so that nothing ever uses the masterKey directly:

  • dekWrapKey – wraps/unwraps per-object data encryption keys (DEKs)
  • the DB key – returned by DeriveDBKey for BadgerDB encryption

The mu mutex protects only the on-disk serialization in saveMasterKey (called by UpdateMasterKeyEncryption); reader methods are lock-free because the fields they touch are immutable after construction.

func NewEncryptionManager 

func NewEncryptionManager(baseDir string) (*EncryptionManager, error)

NewEncryptionManager creates a new encryption manager It loads or creates the master key from the specified directory.

Requires issuer keys to be initialized via config.GetIssuerPublicJWKS() or InitIssuerKeyForTests() before calling this function.

func (*EncryptionManager) DecryptDataKey 

func (em *EncryptionManager) DecryptDataKey(encryptedDEK []byte) ([]byte, error)

DecryptDataKey decrypts a DEK using the HKDF-derived wrapping key. The wrapping key is immutable after construction, so no lock is needed.

func (*EncryptionManager) DecryptInline 

func (em *EncryptionManager) DecryptInline(encryptedData, dek, nonce []byte) ([]byte, error)

DecryptInline decrypts inline data (small objects)

func (*EncryptionManager) DeriveDBKey 

func (em *EncryptionManager) DeriveDBKey() ([]byte, error)

DeriveDBKey derives a separate encryption key for BadgerDB using HKDF. This ensures proper key separation: the DEK-wrapping key handles per-object keys, while this derived key encrypts BadgerDB's LSM tree and WAL files (protecting metadata such as ETags, URLs, and timestamps at rest).

masterKey is immutable after construction, so no lock is needed.

func (*EncryptionManager) EncryptDataKey 

func (em *EncryptionManager) EncryptDataKey(dek []byte) ([]byte, error)

EncryptDataKey encrypts a DEK using the HKDF-derived wrapping key. The wrapping key is immutable after construction, so no lock is needed.

func (*EncryptionManager) EncryptInline 

func (em *EncryptionManager) EncryptInline(data, dek, nonce []byte) ([]byte, error)

EncryptInline encrypts data for inline storage (small objects)

func (*EncryptionManager) GenerateDataKey 

func (em *EncryptionManager) GenerateDataKey() ([]byte, error)

GenerateDataKey generates a new random data encryption key (DEK)

func (*EncryptionManager) UpdateMasterKeyEncryption 

func (em *EncryptionManager) UpdateMasterKeyEncryption() error

UpdateMasterKeyEncryption re-encrypts the master key with current issuer keys This should be called when issuer keys change

type EvictionConfig 

type EvictionConfig struct {
	// DirConfigs maps storageID to its eviction limits.
	// Each entry describes one storage directory.
	DirConfigs map[StorageID]EvictionDirConfig
}

EvictionConfig holds configuration for the eviction manager

type EvictionDirConfig 

type EvictionDirConfig struct {
	MaxSize             uint64 // Maximum cache size in bytes for this directory
	HighWaterPercentage int    // Percentage at which eviction starts (0 = default 90)
	LowWaterPercentage  int    // Percentage at which eviction stops  (0 = default 80)
	HighWaterBytes      uint64 // Absolute byte threshold (overrides percentage when > 0)
	LowWaterBytes       uint64 // Absolute byte threshold (overrides percentage when > 0)
	// NoPlacement excludes this storage target from new-object placement
	// (ChooseDiskStorage).  Used for tiering targets, which only receive
	// completed objects tiered by the uploader; they still participate in
	// usage accounting and eviction.
	NoPlacement bool
}

EvictionDirConfig holds per-directory eviction configuration.

type EvictionManager 

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

EvictionManager handles fairness-aware cache eviction. Each storage directory has independent watermarks; eviction is triggered per-directory when a directory exceeds its high-water mark and proceeds until the directory's usage falls to its low-water mark.

func NewEvictionManager 

func NewEvictionManager(db *CacheDB, storage *StorageManager, config EvictionConfig) *EvictionManager

NewEvictionManager creates a new eviction manager

func (*EvictionManager) ChooseDiskStorage 

func (em *EvictionManager) ChooseDiskStorage() StorageID

ChooseDiskStorage selects a storage directory using a pre-computed, shuffled lookup table. The table is rebuilt at most every 100 ms from the in-memory atomic usage estimates. The hot path is a single atomic increment + array index, so many goroutines can call this concurrently with negligible contention.

func (*EvictionManager) DirFree 

func (em *EvictionManager) DirFree(storageID StorageID) int64

DirFree returns the estimated free bytes on a storage target based on the in-memory usage estimate (no DB query). Returns 0 for unknown IDs.

func (*EvictionManager) ForcePurge 

func (em *EvictionManager) ForcePurge() error

ForcePurge forces an immediate purge down to the low water mark. Directories are processed concurrently (see checkAndEvict).

func (*EvictionManager) ForcePurgeToBytes 

func (em *EvictionManager) ForcePurgeToBytes(targetBytes uint64) (uint64, int64, error)

ForcePurgeToBytes forces an immediate purge across all directories until total cache usage drops to targetBytes or below. Unlike ForcePurge, which uses the configured low-water mark, this method accepts an explicit target. Returns (bytes freed, objects evicted, error).

func (*EvictionManager) GetAllNamespaceUsage 

func (em *EvictionManager) GetAllNamespaceUsage() (map[StorageUsageKey]int64, error)

GetAllNamespaceUsage returns usage for all storage+namespace combinations

func (*EvictionManager) GetNamespaceUsage 

func (em *EvictionManager) GetNamespaceUsage(storageID StorageID, namespaceID NamespaceID) (int64, error)

GetNamespaceUsage returns usage for a specific storage+namespace combination

func (*EvictionManager) GetStats 

func (em *EvictionManager) GetStats() EvictionStats

GetStats returns eviction manager statistics

func (*EvictionManager) GetTotalUsage 

func (em *EvictionManager) GetTotalUsage() uint64

GetTotalUsage returns the current total cache usage (sum of per-dir atomics).

func (*EvictionManager) HasSpace 

func (em *EvictionManager) HasSpace(needed uint64) bool

HasSpace returns true if there's room for more data in at least one directory. Uses the in-memory atomic estimates (no DB query).

func (*EvictionManager) MarkPurgeFirst 

func (em *EvictionManager) MarkPurgeFirst(instanceHash InstanceHash) error

MarkPurgeFirst marks an object to be purged first during next eviction

func (*EvictionManager) NoteUsageDecrease 

func (em *EvictionManager) NoteUsageDecrease(storageID StorageID, bytes int64)

NoteUsageDecrease adjusts the in-memory estimate downward after bytes were released from a storage target (e.g. the local copy removed after tiering to a remote target, or a failed upload refunded). The DB-level counters must already have been updated by the caller.

func (*EvictionManager) NoteUsageIncrease 

func (em *EvictionManager) NoteUsageIncrease(storageID StorageID, bytes int64)

NoteUsageIncrease updates the in-memory usage estimate and triggers an eviction check if the specified directory's high-water mark appears to be exceeded. It does NOT write to the persistent database — the caller is responsible for ensuring the DB was already updated (e.g., via MergeBlockStateWithUsage which tracks usage atomically alongside the bitmap merge). The database usage updates handle race conditions where multiple writers may be updating the same blocks simultaneously. This doesn't - hence the estimated usage is going to be potentially higher than the actual usage.

The per-directory atomic counter is the fast path: if the counter is under the high-water mark no database call is made at all. When the counter indicates a possible threshold crossing we consult the database for the authoritative total, correct the counter, and only then decide whether to trigger eviction.

func (*EvictionManager) RecordAccess 

func (em *EvictionManager) RecordAccess(instanceHash InstanceHash) error

RecordAccess records an access to an object, updating LRU

func (*EvictionManager) Start 

func (em *EvictionManager) Start(ctx context.Context, egrp *errgroup.Group)

Start begins the background eviction goroutine

func (*EvictionManager) TriggerEviction 

func (em *EvictionManager) TriggerEviction()

TriggerEviction triggers an eviction check

type EvictionStats 

type EvictionStats struct {
	TotalUsage     uint64
	DirStats       map[StorageID]DirEvictionStats
	NamespaceUsage map[StorageUsageKey]int64
}

EvictionStats contains eviction manager statistics

type HeadResult 

type HeadResult struct {
	ContentLength int64
	Meta          *CacheMetadata // non-nil when the object is cached
	// Checksums carries the object's digests for the RFC 3230 Digest header.
	// Populated from cached metadata on a hit, or from the origin's stat
	// response on a miss, so a HEAD always relays a Digest when one is known.
	Checksums []Checksum
}

HeadResult contains the response metadata for a HEAD request.

type InstanceHash 

type InstanceHash string

InstanceHash is an HMAC-SHA-256 digest that identifies a specific version (ETag) of an object. Using a dedicated type prevents accidental confusion with ObjectHash or arbitrary strings.

func ComputeInstanceHash 

func ComputeInstanceHash(salt []byte, etag string, objectHash ObjectHash) InstanceHash

ComputeInstanceHash computes HMAC-SHA-256(salt, etag + ":" + objectHash). This identifies a specific version of an object. If etag is empty, uses empty string (for objects without ETag support).

func ParseChunkFilename 

func ParseChunkFilename(filename string) (baseHash InstanceHash, chunkIndex int, ok bool)

ParseChunkFilename parses a filename (without path) that may be a chunk file. Returns the base instance hash and chunk index. For non-chunked files (no suffix), returns the filename as-is and chunkIndex 0. For chunk files like "deadbeef...-2", returns the base hash and chunkIndex 1. Returns ok=false if the filename doesn't match expected patterns.

func ParseInstanceHash 

func ParseInstanceHash(s string) (InstanceHash, error)

ParseInstanceHash checks that s has the form of an instance hash -- 64 lowercase hex digits -- before treating it as one. An instance hash names files and remote objects, so one taken from outside the cache (a request, a directory listing, a bucket key) must go through here.

type IntrospectAPIOpen 

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

IntrospectAPIOpen provides read-only introspection access to the cache database. It opens the cache database in read-only mode, suitable for CLI tools that need to inspect cache contents without disturbing a running cache server.

func NewIntrospectAPI 

func NewIntrospectAPI(baseDir string) (*IntrospectAPIOpen, error)

NewIntrospectAPI creates a new introspection API by opening the cache database and storage manager in read-only mode.

The cache must be stopped: OpenCacheDBReadOnly takes BadgerDB's directory lock, and it explains as much when it cannot. `pelican cache introspect` only reaches this path when the cache is not running, routing to the live service otherwise.

func (*IntrospectAPIOpen) Close 

func (api *IntrospectAPIOpen) Close() error

Close releases resources held by the introspection API.

func (*IntrospectAPIOpen) GetCacheStats 

func (api *IntrospectAPIOpen) GetCacheStats() (*CacheStats, error)

GetCacheStats returns aggregate cache size statistics by scanning metadata. This scans all metadata entries and the inline data prefix, but does NOT walk the disk (use GetDiskUsage for that).

func (*IntrospectAPIOpen) GetDiskUsage 

func (api *IntrospectAPIOpen) GetDiskUsage() (*DiskUsageResult, error)

GetDiskUsage walks the storage directories to compute actual disk usage. This is an expensive operation that reads every file's size on disk.

func (*IntrospectAPIOpen) GetObjectDetails 

func (api *IntrospectAPIOpen) GetObjectDetails(instanceHash string) (*ObjectDetails, error)

GetObjectDetails returns detailed metadata for a specific object instance. The instanceHash can be obtained from ListObjectInstances. Alternatively, pass objectURL and etag to look up by those identifiers.

func (*IntrospectAPIOpen) GetObjectDetailsByURL 

func (api *IntrospectAPIOpen) GetObjectDetailsByURL(objectURL, etag string) (*ObjectDetails, error)

GetObjectDetailsByURL looks up object details by URL and optional ETag. If etag is empty, returns details for the latest version.

func (*IntrospectAPIOpen) ListAllObjects 

func (api *IntrospectAPIOpen) ListAllObjects(limit int, pattern string) ([]ObjectInstance, error)

ListAllObjects returns a summary of all cached objects. This can be slow for large caches; consider using pagination in production.

func (*IntrospectAPIOpen) ListObjectInstances 

func (api *IntrospectAPIOpen) ListObjectInstances(objectURL string) ([]ObjectInstance, error)

ListObjectInstances returns all cached instances for a given object URL. The URL should be a pelican:// URL (e.g., pelican://host/path/file.dat). If the URL doesn't have a scheme, it will be normalized with the default federation.

func (*IntrospectAPIOpen) RunConsistencyCheck 

func (api *IntrospectAPIOpen) RunConsistencyCheck(ctx context.Context, metadataScan, dataScan bool) (*ConsistencyCheckResult, error)

RunConsistencyCheck runs a full consistency check (metadata scan + data scan) and returns statistics. This creates a temporary ConsistencyChecker with unlimited rate limiting for the ad-hoc run.

func (*IntrospectAPIOpen) VerifyChecksum 

func (api *IntrospectAPIOpen) VerifyChecksum(instanceHash string) (*VerificationResult, error)

VerifyChecksum triggers a checksum verification for the specified instance. Returns detailed verification results including per-checksum status.

func (*IntrospectAPIOpen) VerifyChecksumByURL 

func (api *IntrospectAPIOpen) VerifyChecksumByURL(objectURL, etag string) (*VerificationResult, error)

VerifyChecksumByURL verifies checksum for an object by URL and optional ETag.

type LocalCache 

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

func NewLocalCache 

func NewLocalCache(ctx context.Context, egrp *errgroup.Group, options ...LocalCacheOption) (lc *LocalCache, err error)

Create a local cache object

Launches background goroutines associated with the cache

func (*LocalCache) Config 

func (lc *LocalCache) Config(egrp *errgroup.Group) (err error)

Try to configure the local cache and launch the reconfigure goroutine

func (*LocalCache) Get 

func (sc *LocalCache) Get(ctx context.Context, path, token string) (io.ReadCloser, error)

Get path from the cache

func (*LocalCache) MarkObjectPurgeFirst 

func (lc *LocalCache) MarkObjectPurgeFirst(objectPath string) (int, error)

MarkObjectPurgeFirst marks the given object path as PURGEFIRST by creating the corresponding sentinel file on disk and updating the in-memory data structures accordingly.

func (*LocalCache) ReconstructCache 

func (lc *LocalCache) ReconstructCache() error

ReconstructCache rebuilds the in-memory data structures of the LocalCache based on the files present in the directory specified by "LocalCache.DataLocation".

func (*LocalCache) Stat 

func (lc *LocalCache) Stat(path, token string) (uint64, error)

type LocalCacheOption 

type LocalCacheOption = option.Interface

func WithDeferConfig 

func WithDeferConfig(deferConfig bool) LocalCacheOption

Create an option to defer the configuration of the local cache

Useful in cases where the cache should be created before the web interface is up -- but the web interface is needed to complete configuration.

type MasterKeyFile 

type MasterKeyFile struct {
	// Keys maps public key fingerprint to encrypted master key
	Keys map[string][]byte `json:"keys"`
}

MasterKeyFile represents the encrypted master key file format The master key is encrypted with each issuer private key

type NamespaceID 

type NamespaceID uint32

NamespaceID identifies a namespace prefix. Each distinct prefix registered in the cache is assigned a monotonically increasing ID for efficient storage and lookup. Using a dedicated type prevents accidental confusion with StorageID or arbitrary uint32 values.

type ObjectBlockState 

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

ObjectBlockState holds the shared, thread-safe block availability state for a single cached object. All RangeReaders serving the same instanceHash share one instance, ensuring that block additions (downloads) and removals (corruption repairs) are immediately visible to every goroutine.

The repairMu mutex serializes repair operations: when multiple readers detect corruption simultaneously, only the first one performs the clear-fetch-verify cycle. Subsequent callers acquire repairMu, re-check block availability, and skip the repair if it has already been done.

func NewObjectBlockState 

func NewObjectBlockState(bitmap *roaring.Bitmap) *ObjectBlockState

NewObjectBlockState wraps an existing bitmap in a thread-safe container.

func (*ObjectBlockState) Add 

func (obs *ObjectBlockState) Add(block uint32)

Add marks a single block as downloaded.

func (*ObjectBlockState) AddRange 

func (obs *ObjectBlockState) AddRange(start, end uint32)

AddRange marks all blocks in [start, end] as downloaded.

func (*ObjectBlockState) AttachReader 

func (obs *ObjectBlockState) AttachReader() (detach func())

AttachReader records that a reader of the object is open and returns the function to call when it closes; calling that more than once is harmless.

The count lives here, on the one state every reader of the object shares, rather than with whichever download or fetcher a reader happened to come through: a reader that arrived through a cache hit while a miss was still filling the object waits on this state, not on the miss's fetcher, and must keep that fill going just the same.

func (*ObjectBlockState) ClearDownloading 

func (obs *ObjectBlockState) ClearDownloading()

ClearDownloading marks the background download as finished and wakes any goroutines waiting in WaitForBlock. A download that failed in a way that condemns its data must condemn the state first (see condemn), so that the readers it wakes see why.

func (*ObjectBlockState) Clone 

func (obs *ObjectBlockState) Clone() *roaring.Bitmap

Clone returns a point-in-time snapshot of the bitmap. The returned bitmap is independent (mutations to it do not affect the shared state).

func (*ObjectBlockState) Condemned 

func (obs *ObjectBlockState) Condemned() error

Condemned returns why the instance was dropped as bad, or nil.

func (*ObjectBlockState) Contains 

func (obs *ObjectBlockState) Contains(block uint32) bool

Contains returns true if the given block is marked as downloaded.

func (*ObjectBlockState) ContainsRange 

func (obs *ObjectBlockState) ContainsRange(start, end uint32) bool

ContainsRange returns true if every block in [start, end] is present. Creates a temporary range bitmap and checks if the intersection has the expected cardinality. This is O(log n) for typical bitmap layouts.

func (*ObjectBlockState) GetCardinality 

func (obs *ObjectBlockState) GetCardinality() uint64

GetCardinality returns the number of blocks that are downloaded.

func (*ObjectBlockState) LockRepair 

func (obs *ObjectBlockState) LockRepair()

LockRepair acquires the per-object repair mutex. Only one goroutine at a time may run the clear-fetch-verify repair cycle.

func (*ObjectBlockState) MissingInRange 

func (obs *ObjectBlockState) MissingInRange(start, end uint32) []uint32

MissingInRange returns the list of blocks in [start, end] that are not present. Uses RoaringBitmap's iterator for efficiency instead of checking each block.

func (*ObjectBlockState) Remove 

func (obs *ObjectBlockState) Remove(block uint32)

Remove marks a single block as not-downloaded.

func (*ObjectBlockState) RemoveMany 

func (obs *ObjectBlockState) RemoveMany(blocks []uint32)

RemoveMany marks the given blocks as not-downloaded.

func (*ObjectBlockState) SetDownloadETA 

func (obs *ObjectBlockState) SetDownloadETA(eta func(block uint32) time.Time)

SetDownloadETA records how to estimate when the whole-object download in progress will have written a block. ClearDownloading forgets it.

func (*ObjectBlockState) SetDownloading 

func (obs *ObjectBlockState) SetDownloading()

SetDownloading marks this object as having a background download in progress. WaitForBlock will block while downloading is true.

func (*ObjectBlockState) UnlockRepair 

func (obs *ObjectBlockState) UnlockRepair()

UnlockRepair releases the per-object repair mutex.

func (*ObjectBlockState) WaitForBlock 

func (obs *ObjectBlockState) WaitForBlock(ctx context.Context, block uint32) bool

WaitForBlock waits until the specified block is available in the bitmap. It returns true if the block is available, false if the context was cancelled, or no background download or fill is due to write the block (any more) and it is not there. This avoids starting duplicate range downloads when a download or fill is already in progress.

The implementation spawns a goroutine to wait on the sync.Cond (which cannot be interrupted) and selects between it and ctx.Done(). On context cancellation, we signal the goroutine via a done channel and broadcast the cond, then wait for the goroutine to acknowledge exit before returning. This guarantees no goroutine leak.

type ObjectDetails 

type ObjectDetails struct {
	ObjectInstance

	// Additional details
	NamespaceID   uint16            `json:"namespace_id"`
	StorageID     uint8             `json:"storage_id"`
	LastValidated time.Time         `json:"last_validated,omitempty"`
	CacheControl  string            `json:"cache_control,omitempty"`
	Checksums     []ChecksumInfo    `json:"checksums,omitempty"`
	BlockSummary  *BlockSummary     `json:"block_summary,omitempty"` // nil for inline storage
	ChunkSummary  *ChunkInfoSummary `json:"chunk_info,omitempty"`    // nil for non-chunked
}

ObjectDetails provides detailed metadata for a cached object instance.

type ObjectHash 

type ObjectHash string

ObjectHash is an HMAC-SHA-256 digest that identifies a logical object (URL) regardless of version or ETag. Using a dedicated type prevents accidental confusion with InstanceHash or arbitrary strings.

func ComputeObjectHash 

func ComputeObjectHash(salt []byte, pelicanURL string) ObjectHash

ComputeObjectHash computes HMAC-SHA-256(salt, normalized URL). This identifies the logical object (URL) regardless of version/ETag. The salt is generated once per cache database and prevents offline correlation of hashes with known URLs.

type ObjectInstance 

type ObjectInstance struct {
	InstanceHash  string    `json:"instance_hash"`
	ETag          string    `json:"etag"`
	SourceURL     string    `json:"source_url"`
	ContentLength int64     `json:"content_length"`
	ContentType   string    `json:"content_type"`
	LastModified  time.Time `json:"last_modified,omitempty"`
	Completed     time.Time `json:"completed,omitempty"`
	LastAccessed  time.Time `json:"last_accessed,omitempty"`
	Expires       time.Time `json:"expires,omitempty"`
	IsLatest      bool      `json:"is_latest"` // True if this is the latest ETag
	IsInline      bool      `json:"is_inline"` // True if stored inline in database
}

ObjectInstance represents a cached instance of an object with a specific ETag.

type ObjectReader 

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

ObjectReader provides a reader interface for cached objects. For disk-backed objects, it caches the BlockEncryptor and ObjectBlockState at construction time so that Read/ReadAt calls do not need to look them up through the TTL caches on every call.

func (*ObjectReader) Close 

func (r *ObjectReader) Close() error

Close releases the reader's reference to the underlying file handle. The actual FD is closed only when all references (including the TTL cache entry) have been released.

func (*ObjectReader) ContentType 

func (r *ObjectReader) ContentType() string

ContentType returns the content type of the object

func (*ObjectReader) ETag 

func (r *ObjectReader) ETag() string

ETag returns the ETag

func (*ObjectReader) LastModified 

func (r *ObjectReader) LastModified() time.Time

LastModified returns the last modified time

func (*ObjectReader) Read 

func (r *ObjectReader) Read(p []byte) (n int, err error)

Read implements io.Reader

func (*ObjectReader) ReadAt 

func (r *ObjectReader) ReadAt(p []byte, off int64) (n int, err error)

ReadAt implements io.ReaderAt

func (*ObjectReader) Seek 

func (r *ObjectReader) Seek(offset int64, whence int) (int64, error)

Seek implements io.Seeker

func (*ObjectReader) Size 

func (r *ObjectReader) Size() int64

Size returns the total size of the object

func (*ObjectReader) WriteTo 

func (r *ObjectReader) WriteTo(w io.Writer) (int64, error)

WriteTo implements io.WriterTo. When io.Copy detects this interface it calls WriteTo directly instead of issuing many small Read calls with a 32 KB default buffer. This lets the ObjectReader control the batch size, reading and decrypting many blocks at once and writing the plaintext to w in large chunks.

type PersistentCache 

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

PersistentCache is the new persistent local cache implementation It uses BadgerDB for metadata and block tracking, and encrypted files on disk

func NewPersistentCache 

func NewPersistentCache(ctx context.Context, egrp *errgroup.Group, cfg PersistentCacheConfig) (*PersistentCache, error)

NewPersistentCache creates a new persistent cache instance

func (*PersistentCache) BackdateObject 

func (pc *PersistentCache) BackdateObject(objectPath string, age time.Duration) error

BackdateObject shifts the Completed timestamp of a cached object backward by the given duration. This is useful for testing the Age response header without waiting for wall-clock time to elapse.

func (*PersistentCache) Close 

func (pc *PersistentCache) Close() error

Close shuts down the persistent cache.

Shutdown order:

  1. Cancel all in-flight downloads (via downloadCancel).
  2. Wait for completeDownload goroutines to finish — each one closes its own TransferClient, so all clients are gone before step 3.
  3. Shut down the transfer engine (no live clients remain).
  4. Stop the consistency checker.
  5. Close the database.

func (*PersistentCache) Config 

func (pc *PersistentCache) Config(egrp *errgroup.Group) error

Config configures the cache and starts periodic updates

func (*PersistentCache) EvictObject 

func (pc *PersistentCache) EvictObject(objectPath, token string) error

EvictObject is the programmatic API for evicting an object by path. Returns nil on success, an error if the object is in use or the eviction fails.

func (*PersistentCache) EvictPrefix 

func (pc *PersistentCache) EvictPrefix(prefix, token string, immediate bool) (int, error)

EvictPrefix evicts all cached objects whose source URL starts with the given path prefix (or matches it exactly). It iterates every metadata entry, selects those with a matching SourceURL, and either deletes them immediately or marks them for priority eviction (purge-first) depending on the immediate flag. Objects that are currently being downloaded are skipped. The returned count reflects objects that were acted on.

func (*PersistentCache) Get 

func (pc *PersistentCache) Get(ctx context.Context, objectPath, token string) (io.ReadCloser, error)

Get retrieves an object from the cache, downloading if necessary. The reader returned reads under ctx, as GetRange's does.

func (*PersistentCache) GetCacheStats 

func (pc *PersistentCache) GetCacheStats() (*CacheStats, error)

introspectStatsHandler returns aggregate cache size statistics. GetCacheStats returns aggregate cache size statistics by scanning metadata and reading usage counters. This is the same data returned by the introspect/stats HTTP endpoint.

func (*PersistentCache) GetMetadata 

func (pc *PersistentCache) GetMetadata(objectPath, token string) (*CacheMetadata, error)

GetMetadata returns the cache metadata for an object if it exists. Returns nil, nil if the object is not cached.

func (*PersistentCache) GetRange 

func (pc *PersistentCache) GetRange(ctx context.Context, objectPath, token, rangeHeader string) (io.ReadCloser, error)

GetRange retrieves a range of an object from the cache. A reader that may wait for blocks, or fetch them, reads under ctx: once ctx ends, a Read waiting for a block gives up.

func (*PersistentCache) GetSeekableReader 

func (pc *PersistentCache) GetSeekableReader(ctx context.Context, objectPath, bearerToken string, rangeOnly bool) (*SeekableReader, *CacheMetadata, error)

GetSeekableReader returns a seekable reader for the full object with on-demand block fetching. This is designed for use with http.ServeContent which handles Range requests internally.

When rangeOnly is true and the object is not yet cached, GetSeekableReader uses a lightweight HEAD request to initialize on-disk storage instead of starting a full sequential download. This allows BlockFetcherV2 to fetch only the blocks the caller actually reads, avoiding a potentially expensive full transfer. Callers should set rangeOnly when they know the request is for a sub-range of the object (e.g. an HTTP Range request).

The reader reads under ctx; a caller whose ctx must outlive the reader's client -- because it also drives the download -- sets the client's with SetReadContext.

func (*PersistentCache) GetStats 

func (pc *PersistentCache) GetStats() PersistentCacheStats

GetStats returns cache statistics

func (*PersistentCache) HeadObject 

func (pc *PersistentCache) HeadObject(objectPath, token string) (*HeadResult, error)

HeadObject returns metadata for an object without triggering a download. If the object is cached, the full CacheMetadata is returned. If not cached, the origin is queried via HEAD (DoStat) for the size.

func (*PersistentCache) IsFullyCached 

func (pc *PersistentCache) IsFullyCached(ctx context.Context, objectPath, token string) bool

IsFullyCached returns true if the object at the given path is fully downloaded and present in the cache (metadata exists, ContentLength >= 0, and the download has been marked complete).

When the cached entry carries Cache-Control directives and is stale, a revalidation request is made to the origin. If revalidation confirms the same ETag (or fails gracefully, serving stale), the object is still considered fully cached. If the origin provides a new version or responds with no-store, the method returns false so the caller falls through to a full download.

func (*PersistentCache) KeyChangeCallback 

func (pc *PersistentCache) KeyChangeCallback() func(ctx context.Context) error

KeyChangeCallback returns a callback function that re-encrypts the master key when issuer keys change. This is used with LaunchIssuerKeysDirRefresh to ensure the cache data remains accessible as long as any issuer key is available. The callback updates the masterkey.json file with the master key encrypted under all current issuer keys.

func (*PersistentCache) LaunchListener 

func (pc *PersistentCache) LaunchListener(ctx context.Context, egrp *errgroup.Group) (err error)

LaunchListener launches the unix socket listener for the persistent cache

func (*PersistentCache) MarkPurgeFirst 

func (pc *PersistentCache) MarkPurgeFirst(objectPath string) error

MarkPurgeFirst marks an object to be purged first during next eviction

func (*PersistentCache) Purge 

func (pc *PersistentCache) Purge() error

Purge triggers the eviction manager to purge old entries

func (*PersistentCache) PurgeToBytes 

func (pc *PersistentCache) PurgeToBytes(targetBytes uint64) (uint64, int64, error)

PurgeToBytes triggers the eviction manager to purge down to the specified total usage target. Returns (bytes freed, objects evicted, error).

func (*PersistentCache) Register 

func (pc *PersistentCache) Register(ctx context.Context, router *gin.RouterGroup)

Register registers the control & monitoring routines with Gin

func (*PersistentCache) RegisterCacheHandlers 

func (pc *PersistentCache) RegisterCacheHandlers(engine *gin.Engine, directorEnabled bool) error

RegisterCacheHandlers registers HTTP handlers for the persistent cache on the Gin engine. This is used for the XRootD-free cache implementation where the cache serves content directly via the web server instead of using XRootD.

Unlike LaunchListener (which creates a Unix socket), this registers handlers on the existing Gin web server, allowing it to serve cache requests on the standard HTTP/HTTPS ports.

The /api/v1.0/cache/data/:discovery/*path routes are always registered regardless of whether a director is in use. When running standalone (directorEnabled=false), a NoRoute fallback is also registered so that bare object paths (without the /api prefix) are served. The /api namespace is always reserved and never served by the NoRoute fallback.

func (*PersistentCache) SetFedToken 

func (pc *PersistentCache) SetFedToken(tok string)

SetFedToken stores the federation token in memory. It is called by cache.LaunchFedTokManager (via the onTokenUpdate callback) whenever the token is created or refreshed, eliminating the need to read the token back from disk on every origin request.

func (*PersistentCache) Stat 

func (pc *PersistentCache) Stat(objectPath, token string) (uint64, error)

Stat returns the size of an object, querying the origin if not cached

func (*PersistentCache) StatCachedOnly 

func (pc *PersistentCache) StatCachedOnly(objectPath, token string) (uint64, error)

StatCachedOnly returns the size of an object only if it's cached, whole. Returns 0, ErrNotCached if the object is not in the cache or only part of it is.

type PersistentCacheConfig 

type PersistentCacheConfig struct {
	// Mode selects which configuration namespace to use for defaults.
	// CacheModeServer reads Cache.StorageLocation, Cache.HighWaterMark, etc.
	// CacheModeLocal (the zero value) reads LocalCache.DataLocation,
	// LocalCache.HighWaterMarkPercentage, etc.
	Mode CacheMode

	// BaseDir is the root directory for the cache.  The BadgerDB database
	// lives directly under BaseDir.  If StorageDirs is empty, a single
	// storage directory is created under BaseDir/objects.
	BaseDir string

	// StorageDirs configures one or more storage directories.
	// Each entry describes a directory path and its size limits.
	// When empty, a single directory under BaseDir is used with
	// MaxSize / HighWaterMarkPercentage / LowWaterMarkPercentage
	// as its limits (for backward compatibility).
	StorageDirs []StorageDirConfig

	// ExpectFedToken says a federation token manager will deliver tokens
	// through SetFedToken. Only then does a cache miss that arrives before
	// the first token wait briefly for it; a cache with no manager (a
	// site-local cache, or one built directly by a test) goes to the
	// origin without a token right away.
	ExpectFedToken bool

	// Legacy single-directory fields — used only when StorageDirs is empty.
	MaxSize                 uint64
	HighWaterMarkPercentage int
	LowWaterMarkPercentage  int

	// InlineStorageMaxBytes sets the maximum size for objects stored
	// inline in BadgerDB.  Objects at or below this threshold are stored
	// inline; larger objects go to disk.  0 means use the default (4096).
	InlineStorageMaxBytes int

	DefaultFederation string

	// DeferConfig delays the initial director namespace fetch until
	// Config() is called explicitly.  The server launcher sets this to
	// true because the director may not be reachable when the cache is
	// first constructed (e.g. cache starts before director discovery).
	DeferConfig bool
}

PersistentCacheConfig holds configuration for the persistent cache

type PersistentCacheStats 

type PersistentCacheStats struct {
	TotalUsage       uint64
	DirStats         map[StorageID]DirEvictionStats
	NamespaceUsage   map[string]int64
	ConsistencyStats ConsistencyStats
}

PersistentCacheStats holds cache statistics

type PrestageManager 

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

PrestageManager manages per-identity worker pools for prestage operations.

func NewPrestageManager 

func NewPrestageManager(pc *PersistentCache) *PrestageManager

NewPrestageManager creates a prestage manager for the given cache.

func (*PrestageManager) Submit 

func (pm *PrestageManager) Submit(ident string, req *prestageRequest) bool

Submit queues a prestage request for the given identity. Returns false if the queue is full (caller should return 429).

type RangeReader 

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

RangeReader implements io.ReadSeeker over a byte range of a cached object. It is the primary read-path abstraction that ties the block-level storage layer to HTTP response serving.

Callers obtain a RangeReader via PersistentCache.GetRange (for explicit byte-range requests) or via PersistentCache.GetSeekableReader (wrapped in a SeekableReader so that http.ServeContent can handle Range negotiation).

For normal (cached / cacheable) responses the RangeReader reads encrypted blocks from the StorageManager, fetching missing blocks on demand via the fetchBlocks callback wired to a BlockFetcherV2 instance. It also performs a single round of transparent auto-repair when AES-GCM authentication detects on-disk corruption.

For responses marked Cache-Control: no-store, the origin's body is piped directly through the noStoreReader field; the block-storage path is bypassed entirely.

func NewRangeReader 

func NewRangeReader(
	storage *StorageManager,
	instanceHash InstanceHash,
	start, end int64,
	fetchBlocks func(ctx context.Context, startBlock, endBlock uint32) error,
) (*RangeReader, error)

NewRangeReader creates a reader for a range request fetchBlocks is called when blocks need to be fetched from origin

func (*RangeReader) Close 

func (rr *RangeReader) Close() error

Close closes the range reader

func (*RangeReader) ContentLength 

func (rr *RangeReader) ContentLength() int64

ContentLength returns the length of the range

func (*RangeReader) ContentRange 

func (rr *RangeReader) ContentRange() string

ContentRange returns the Content-Range header value

func (*RangeReader) LimitFill 

func (rr *RangeReader) LimitFill(ranges []RangeRequest)

LimitFill bounds the background fills this reader may start (see ensureBlocks) to the given byte ranges of the object: a fill stays within the range the reader is reading, and none starts outside every range (so no ranges at all means no fills). The cache's HTTP handler reads every object through a reader of the whole object and lets http.ServeContent seek it to each requested range in turn, so it states the ranges here: a request for a few bytes of a large, partly cached object must not start a fill of everything after them, and a multi-range request -- a vector read -- fills each of its ranges with one request.

func (*RangeReader) Read 

func (rr *RangeReader) Read(p []byte) (n int, err error)

Read implements io.Reader with on-demand block fetching, under the context set with SetReadContext.

func (*RangeReader) ReadContext 

func (rr *RangeReader) ReadContext(ctx context.Context, p []byte) (n int, err error)

ReadContext reads with context for cancellation

func (*RangeReader) Seek 

func (rr *RangeReader) Seek(offset int64, whence int) (int64, error)

Seek implements io.Seeker (limited to within the range)

func (*RangeReader) SetReadContext 

func (rr *RangeReader) SetReadContext(ctx context.Context)

SetReadContext sets the context that Read reads under: a read waiting for a block that a download or fill has yet to write, or fetching one, gives up when it ends. A caller that reads through the io.Reader interface -- http.ServeContent, io.Copy -- has no other way to pass one, and without it a reader whose client has gone stays blocked, and attached to the object, for as long as the transfer it waits on takes: with a stalled origin, until that transfer times out, keeping the transfer from ever going idle in the meantime.

The streams a reader may read instead of the cached blocks honour ctx too. A no-store reader's read blocks on the pipe from the origin's transfer, so when ctx ends the pipe is closed with ctx's cause (see noStoreBody.CloseWithError): the read returns, and the transfer is stopped rather than run on with nothing to read it. A reader proxying an object from a tiering target passes ctx on to its stream (see tierObjectStream.setReadContext), so a read stalled on the target gives up too. Call it before reading.

func (*RangeReader) WaitForCompletion 

func (rr *RangeReader) WaitForCompletion(ctx context.Context) error

WaitForCompletion returns the terminal error of the backing download (if any). Intended to be called by the serving path *after* the body has finished streaming, so that a post-download verification failure -- e.g. the origin's reported digest disagreeing with the bytes received -- can be surfaced to the client through the X-Transfer-Status trailer rather than being silently masked by an otherwise-successful read of the cached blocks.

Behavior depends on what the reader was asked to read:

  • Full-object reads block until the download finishes (subject to ctx). The client is already waiting for the whole body, so the small extra wait for the post-body digest comparison is acceptable and necessary to detect verification failures.

  • Partial-range reads do NOT block. The client only asked for a slice (which may be a few KB of a multi-GB object); the backing download may still have far more body to fetch. We only surface an error if verification has *already* completed -- never make the range response wait for the rest of the download.

Returns nil when there is no backing download, the download completed successfully, ctx is cancelled while waiting, or (in the partial-range case) verification hasn't completed yet.

type RangeRequest 

type RangeRequest struct {
	Start int64
	End   int64 // -1 means "to end of file"
}

RangeRequest represents a parsed HTTP Range header

func ParseRangeHeader 

func ParseRangeHeader(rangeHeader string, contentLength int64) ([]RangeRequest, error)

ParseRangeHeader parses an HTTP Range header Supports: "bytes=start-end", "bytes=start-", "bytes=-suffix"

type ScanProgressEvent 

type ScanProgressEvent struct {
	// Phase is the scan phase: "metadata" or "data".
	Phase string `json:"phase"`

	// PercentComplete is the estimated completion (0–100).
	PercentComplete float64 `json:"percent_complete"`

	// DBEntriesScanned is the number of DB entries processed so far (metadata only).
	DBEntriesScanned int64 `json:"db_entries_scanned,omitempty"`

	// FilesScanned is the number of filesystem entries processed (metadata only).
	FilesScanned int64 `json:"files_scanned,omitempty"`

	// ObjectsVerified is the number of objects checksummed so far (data only).
	ObjectsVerified int64 `json:"objects_verified,omitempty"`

	// BytesVerified is the total bytes read for checksumming (data only).
	BytesVerified int64 `json:"bytes_verified,omitempty"`

	// Message is a human-readable status line.
	Message string `json:"message,omitempty"`
}

ScanProgressEvent is sent as an SSE event to report scan progress. The Event field (used as SSE event type) is one of:

  • "metadata_progress": periodic update during the metadata scan
  • "metadata_done": the metadata scan has finished
  • "data_progress": periodic update during the data scan
  • "data_done": the data scan has finished
  • "error": an error occurred
  • "done": all scans complete; the Data field is a ConsistencyCheckResult

type SchemaVersion 

type SchemaVersion uint32

SchemaVersion identifies one revision of the on-disk layout of a block store database (both the cache and the pstore origin backend share it, since they share the key space). 

const (
	// CurrentSchemaVersion is the layout this binary reads and writes.
	//
	// Bump it whenever a change makes records written by an older binary
	// unreadable — or, worse, silently misread — by a newer one: a new or
	// renamed key prefix, a different value encoding, or a change to how
	// object/instance hashes are derived — normalizeURL, which decides which
	// spellings of a URL name one object, is exactly that sort of key
	// derivation, and a change there has no other way to announce itself.
	// Purely additive changes that older binaries can ignore do not need a
	// bump.
	//
	// Whoever bumps it is responsible for teaching migrateSchema how to bring
	// a database written under every still-supported older version forward.
	//
	// Version 2 added the S3 storage-target layout: the up: and ps: prefixes
	// and DiskMapping.Backend.  Nothing needs rewriting to read it -- both
	// prefixes are additive and an absent Backend means BackendPosix -- but
	// the bump is not cosmetic.  It is what stops a version-1 binary from
	// opening the database: that binary's FindRecyclableStorageID predates
	// the "never recycle a non-POSIX mapping" rule, so it would hand an S3
	// target's storage ID to a local directory and PurgeStorageID would take
	// every tiered object's metadata with it.  Refusing the database is the
	// mechanism that turns a silent, unrecoverable data loss into a
	// startup error telling the operator to upgrade again.
	CurrentSchemaVersion SchemaVersion = 2
)

type SeekableReader 

type SeekableReader struct {
	*RangeReader
}

SeekableReader is a reader that supports seeking and on-demand block fetching. It implements io.ReadSeekCloser for use with http.ServeContent.

For no-store streaming responses, the reader is NOT seekable (IsNoStore returns true). In that case the handler must use io.Copy instead of http.ServeContent.

func (*SeekableReader) IsNoStore 

func (sr *SeekableReader) IsNoStore() bool

IsNoStore returns true when this reader wraps a streaming no-store response. The caller must NOT use http.ServeContent (which requires seeking); instead it should stream the response with io.Copy and set headers manually.

type StorageDirConfig 

type StorageDirConfig struct {
	// Path is the directory that will hold an "objects/" subdirectory
	// and, for the first directory, the database.
	Path string
	// MaxSize is the maximum number of bytes stored on this directory.
	// If 0, auto-detected from the filesystem at startup.
	MaxSize uint64
	// HighWaterMarkPercentage overrides the global high-water mark for this
	// directory.  0 means use the global default.
	HighWaterMarkPercentage int
	// LowWaterMarkPercentage overrides the global low-water mark for this
	// directory.  0 means use the global default.
	LowWaterMarkPercentage int
}

StorageDirConfig describes one disk-backed storage directory. Multiple directories can be configured to spread data across devices; each directory has its own maximum size and optional watermark overrides.

func ParseStorageDirsConfig 

func ParseStorageDirsConfig() ([]StorageDirConfig, error)

ParseStorageDirsConfig reads the LocalCache.StorageDirs setting from Viper and returns parsed StorageDirConfig values. It accepts two formats for backward compatibility:

  1. A list of strings (paths only): LocalCache: StorageDirs: - /mnt/cache1 - /mnt/cache2

  2. A list of objects with per-directory configuration: LocalCache: StorageDirs: - Path: /mnt/cache1 MaxSize: 500GB HighWaterMarkPercentage: 95 LowWaterMarkPercentage: 85 - Path: /mnt/cache2 MaxSize: 2TB

Returns nil (not an error) when the key is unset or empty.

func ParseStorageDirsValue 

func ParseStorageDirsValue(raw any, name string) ([]StorageDirConfig, error)

ParseStorageDirsValue parses an already-fetched StorageDirs setting. It exists so another subsystem configuring the same block store -- the pstore origin backend, via Origin.PStoreStorageDirs -- accepts exactly the same two formats without duplicating the parsing.

name identifies the setting in error messages.

type StorageDirInfo 

type StorageDirInfo struct {
	StorageID  StorageID
	ObjectsDir string
}

StorageDirInfo describes a configured storage directory at runtime.

type StorageDirStats 

type StorageDirStats struct {
	StorageID   uint8 `json:"storage_id"`
	ObjectCount int64 `json:"object_count"`
	TotalBytes  int64 `json:"total_bytes"`   // Sum of ContentLength for objects in this dir
	InlineCount int64 `json:"inline_count"`  // Number of inline objects
	InlineBytes int64 `json:"inline_bytes"`  // Sum of ContentLength for inline objects
	OnDiskCount int64 `json:"on_disk_count"` // Number of on-disk objects
	OnDiskBytes int64 `json:"on_disk_bytes"` // Actual bytes on disk (content + per-block MAC overhead)
}

StorageDirStats holds per-storage-directory statistics.

type StorageID 

type StorageID uint8

StorageID identifies a storage location. 0 means inline (data in BadgerDB), 1–255 mean disk-backed storage in the directory mapped to that ID. Using a dedicated type prevents accidental confusion with NamespaceID or arbitrary uint8 values. 

const (
	// StorageIDInline is the storage ID for inline data stored in BadgerDB
	StorageIDInline StorageID = 0
	// StorageIDFirstDisk is the storage ID for the first configured disk directory.
	// Additional directories use StorageIDFirstDisk+1, +2, etc.
	StorageIDFirstDisk StorageID = 1
)

type StorageManager 

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

func NewStorageManager 

func NewStorageManager(db *CacheDB, dirs []string, inlineMax int, egrp *errgroup.Group) (*StorageManager, error)

NewStorageManager creates a new storage manager with UUID-based directory identity. It performs the following steps:

  1. Load existing storageID → (UUID, path) mappings from the database.
  2. Scan the supplied directory paths for UUID files.
  3. Match discovered UUIDs against known mappings, updating paths as needed (so that sysadmins can remount directories at different locations).
  4. Assign new storage IDs to directories that have no UUID yet and drop a new UUID file in each.
  5. Persist updated mappings to the database.

dirs is the ordered set of configured directory paths. inlineMax sets the maximum inline object size (0 = use default InlineThreshold).

func NewStorageManagerReadOnly 

func NewStorageManagerReadOnly(baseDir string, db *CacheDB) (*StorageManager, error)

NewStorageManagerReadOnly creates a storage manager for read-only introspection. This is a lightweight variant suitable for CLI tools that only need to read metadata and block states, not perform downloads or writes.

It carries no read-only flag of its own, and needs none: it never touches the database except through db's typed methods, so a db from OpenCacheDBReadOnly already refuses every write made through it. Callers that hand it a writable db (pstore's maintenance open, which needs `fsck --repair` to work) are asserting their own preconditions instead.

func (*StorageManager) AllocateChunk 

func (sm *StorageManager) AllocateChunk(
	ctx context.Context,
	instanceHash InstanceHash,
	meta *CacheMetadata,
	chunkIndex int,
) (updated *CacheMetadata, err error)

AllocateChunk allocates a storage directory for a chunk and creates the chunk file. This must be called before writing to a chunk that has StorageID = 0 (unallocated).

The storage directory is chosen via the pluggable chooseDir function, which defaults to simple round-robin but is replaced in production with EvictionManager.ChooseDiskStorage (weighted by free space).

Returns the updated CacheMetadata (which should be used for subsequent operations).

func (*StorageManager) Close 

func (sm *StorageManager) Close()

Close stops TTL cache eviction goroutines and releases cached resources, the tiering targets' among them.

Only the caches that were actually started are stopped. ttlcache's Stop is an unbuffered channel send that its Start loop receives, so stopping a cache whose loop was never spawned blocks forever -- which is what happened to every manager built by NewStorageManagerReadOnly, making the read-only introspection paths hang on exit.

func (*StorageManager) Delete 

func (sm *StorageManager) Delete(instanceHash InstanceHash) error

Delete removes an object from storage.

The record, the in-memory state and the local files go with the instance's writers and creators shut out (see instanceFences): no writer of the instance writes anything of it afterwards, and a creation of the instance anew -- a fetch of the same version -- happens wholly before or wholly after, so it never loses its new files to this deletion. Only the database transaction and the unlinking of the files run inside; a copy on a tiering target is deleted after, over the network.

func (*StorageManager) DeleteIfUnpinned 

func (sm *StorageManager) DeleteIfUnpinned(instanceHash InstanceHash) (bool, error)

DeleteIfUnpinned deletes an object version unless a reader holds it.

Returns false without deleting when the version is pinned, leaving the caller to retry later. Use this wherever the object being deleted might plausibly be one somebody is reading.

func (*StorageManager) DirCount 

func (sm *StorageManager) DirCount() int

DirCount returns the number of configured storage directories. This is used to determine if chunking should be enabled.

func (*StorageManager) DirIDs 

func (sm *StorageManager) DirIDs() []StorageID

DirIDs returns the list of configured storage directory IDs in sorted order.

func (*StorageManager) EvictByLRU 

func (sm *StorageManager) EvictByLRU(storageID StorageID, namespaceID NamespaceID, maxObjects int, maxBytes int64) ([]evictedObject, uint64, int, error)

EvictByLRU walks the LRU index for a given storage+namespace and evicts the oldest objects until either maxObjects have been removed or maxBytes of content has been freed — whichever comes first. A value of 0 for either limit means "no limit on that dimension". The method is allowed to go one object over the byte threshold to prevent starvation.

Objects under a live reader are spared. Deleting one does not merely slow that reader down: refCountedFile keeps its descriptor open across the unlink, but the reader also needs the object's metadata, data key, and block state, and eviction removes all three -- so the transfer fails. The overlap is small by construction, since the head of the LRU index is the least recently used object and a pinned object is being read right now, but UpdateLRU debounces by ten minutes: a slow reader streaming a large object through a fast-churning cache can drift toward the head, which is exactly the case worth protecting.

A spared object stays at the head of the index and is reconsidered on the next pass, by which time its reader has usually finished.

All DB mutations happen atomically; filesystem deletes follow afterward. Returns the evicted objects, total bytes freed, and how many were spared.

func (*StorageManager) GetDirs 

func (sm *StorageManager) GetDirs() map[StorageID]string

GetDirs returns the configured storage directories (storageID → objects dir).

func (*StorageManager) GetMetadata 

func (sm *StorageManager) GetMetadata(instanceHash InstanceHash) (*CacheMetadata, error)

GetMetadata retrieves metadata for an object

func (*StorageManager) GetMissingBlocks 

func (sm *StorageManager) GetMissingBlocks(instanceHash InstanceHash) ([]BlockRange, error)

GetMissingBlocks returns the ranges of blocks that haven't been downloaded yet

func (*StorageManager) GetObjectSize 

func (sm *StorageManager) GetObjectSize(instanceHash InstanceHash) (int64, error)

GetObjectSize returns the content length of a cached object

func (*StorageManager) GetSharedBlockState 

func (sm *StorageManager) GetSharedBlockState(instanceHash InstanceHash) (*ObjectBlockState, error)

GetSharedBlockState returns the shared, thread-safe block state for the given instanceHash. The state is loaded from the persistent database on first access and cached in a TTL cache that evicts idle entries after blockStateTTL. Every call touches the entry's TTL so actively-used states remain resident. All callers for the same instanceHash receive the same *ObjectBlockState.

func (*StorageManager) HasObject 

func (sm *StorageManager) HasObject(instanceHash InstanceHash) (bool, error)

HasObject checks if an object exists in storage

func (*StorageManager) IdentifyCorruptBlocks 

func (sm *StorageManager) IdentifyCorruptBlocks(instanceHash InstanceHash, startBlock, endBlock uint32) ([]uint32, error)

IdentifyCorruptBlocks probes each block in [startBlock, endBlock] on disk and returns the block numbers whose AES-GCM authentication tag fails or whose on-disk data is too short. Blocks that are not in the downloaded bitmap are skipped. A missing or unopenable file returns all requested blocks.

func (*StorageManager) InitDiskStorage 

func (sm *StorageManager) InitDiskStorage(ctx context.Context, instanceHash InstanceHash, contentLength int64, storageID StorageID, namespaceID NamespaceID) (meta *CacheMetadata, err error)

InitDiskStorage initializes disk storage for a large object in the specified storage directory. The full contentLength is charged to the usage counter at creation time so that the accounting matches the filesystem's pre-allocation (Truncate). Returns the metadata with encryption keys set up.

func (*StorageManager) InitLazyChunkedStorage 

func (sm *StorageManager) InitLazyChunkedStorage(
	ctx context.Context,
	instanceHash InstanceHash,
	contentLength int64,
	chunkSizeCode ChunkSizeCode,
) (meta *CacheMetadata, err error)

InitLazyChunkedStorage initializes metadata for a chunked object WITHOUT creating chunk files. Files are created lazily when AllocateChunk is called (typically on first write to each chunk). This supports byte-range downloads where chunks may be written out of order.

All chunk StorageIDs are initialized to 0 (unallocated). Use AllocateChunk to assign a storage directory and create the file for each chunk before writing.

Parameters:

  • instanceHash: unique identifier for this object version
  • contentLength: total object size in bytes
  • chunkSizeCode: chunk size encoding (must be non-zero for chunked storage)

func (*StorageManager) InlineMaxBytes 

func (sm *StorageManager) InlineMaxBytes() int

InlineMaxBytes returns the configured maximum inline object size.

func (*StorageManager) InvalidateSharedBlockState 

func (sm *StorageManager) InvalidateSharedBlockState(instanceHash InstanceHash)

InvalidateSharedBlockState removes the cached block state for an instanceHash, forcing the next GetSharedBlockState call to reload from the database. This should be called when an object is deleted or evicted.

func (*StorageManager) IsComplete 

func (sm *StorageManager) IsComplete(instanceHash InstanceHash) (bool, error)

IsComplete checks if all blocks have been downloaded

func (*StorageManager) IsObjectPinned 

func (sm *StorageManager) IsObjectPinned(instanceHash InstanceHash) bool

IsObjectPinned reports whether a reader currently holds the version.

func (*StorageManager) IsTiered 

func (sm *StorageManager) IsTiered(id StorageID) bool

IsTiered reports whether the given storage ID is a remote tiering target.

func (*StorageManager) MergeMetadata 

func (sm *StorageManager) MergeMetadata(instanceHash InstanceHash, meta *CacheMetadata) error

MergeMetadata performs an atomic read-modify-update of object metadata

func (*StorageManager) NewAppendWriter 

func (sm *StorageManager) NewAppendWriter(
	ctx context.Context,
	instanceHash InstanceHash,
	namespaceID NamespaceID,
	chunkSizeCode ChunkSizeCode,
) (*AppendWriter, error)

NewAppendWriter initializes chunked storage for an object whose length is not yet known and returns a writer positioned at offset 0.

chunkSizeCode selects the chunk size; ChunkingDisabled is rejected because growth is expressed in whole chunks.

func (*StorageManager) NewBlockWriter 

func (sm *StorageManager) NewBlockWriter(instanceHash InstanceHash, startBlock uint32, existingBitmap *roaring.Bitmap, onComplete func()) (*BlockWriter, error)

NewBlockWriter creates a new block writer for streaming encrypted writes to disk. The existingBitmap parameter allows resuming partial downloads by skipping already-downloaded blocks. startBlock specifies which block number the first incoming byte corresponds to (use 0 to start from the beginning).

func (*StorageManager) NewObjectReader 

func (sm *StorageManager) NewObjectReader(instanceHash InstanceHash) (*ObjectReader, error)

NewObjectReader creates a reader for a cached object

func (*StorageManager) PinObject 

func (sm *StorageManager) PinObject(instanceHash InstanceHash) func()

PinObject holds an object version open, preventing DeleteIfUnpinned from reclaiming it, and returns the release function.

NewObjectReader does this automatically; call it directly only when holding a version across something other than an ObjectReader.

func (*StorageManager) ReadBlocks 

func (sm *StorageManager) ReadBlocks(instanceHash InstanceHash, startOffset int64, length int) ([]byte, error)

ReadBlocks reads and decrypts blocks from disk storage. It uses the shared ObjectBlockState to check block availability.

func (*StorageManager) ReadBlocksInto 

func (sm *StorageManager) ReadBlocksInto(dst []byte, instanceHash InstanceHash, startOffset int64) (int, error)

ReadBlocksInto reads and decrypts blocks from disk storage directly into the caller-provided dst buffer. It returns the number of bytes written to dst. This avoids the result allocation and copy that ReadBlocks performs. dst must be large enough to hold the requested data.

func (*StorageManager) ReadInline 

func (sm *StorageManager) ReadInline(instanceHash InstanceHash) ([]byte, error)

ReadInline reads small data from inline storage

func (*StorageManager) ReclaimAbandonedAppends 

func (sm *StorageManager) ReclaimAbandonedAppends(minAge time.Duration) (int, error)

ReclaimAbandonedAppends removes objects left behind by streaming appends that never reached Finalize or Abort -- the process died, or a consumer dropped the writer without cleaning up.

Such an object is invisible to every other reclamation path: it has no LRU entry, so eviction never sees it; its usage charge is recomputed from live metadata, so the consistency checker re-affirms rather than releases it; and it has both metadata and files, so neither half of the metadata scan fires. The append-in-flight record written by NewAppendWriter is the only handle.

A record is reclaimed only when no writer in this process owns it (the registry is populated before the record is written, so a live append is never mistaken for an orphan) and it is at least minAge old. Pass 0 for minAge to reclaim purely on the registry.

A record whose object is already Completed is dropped without touching the object: that is a Finalize that succeeded and then failed to tidy up.

Returns the number of objects deleted.

func (*StorageManager) RegisterTierTargets 

func (sm *StorageManager) RegisterTierTargets(ctx context.Context, configs []TierTargetConfig) (_ map[StorageID]TierTargetConfig, err error)

RegisterTierTargets resolves identities for the configured tiering targets and assigns each a storage ID, mirroring the UUID-based directory association performed by NewStorageManager. Must be called during single-threaded initialization, before any concurrent access.

Returns the storageID → config mapping for the registered targets.

func (*StorageManager) SetChooseDir 

func (sm *StorageManager) SetChooseDir(fn func() StorageID)

SetChooseDir replaces the storage-directory selection function.

The default is plain round-robin, which ignores how full each directory is. The cache substitutes a free-space-weighted chooser from its eviction manager; a pstore, which has no eviction manager, substitutes one driven by its own capacity limits. Without this a store with directories of different sizes round-robins straight into a full one.

Must be called before any concurrent access begins.

func (*StorageManager) SetMetadata 

func (sm *StorageManager) SetMetadata(instanceHash InstanceHash, meta *CacheMetadata) error

SetMetadata stores metadata for an object (full replace, use for initial creation only)

func (*StorageManager) SetUnpinObserver 

func (sm *StorageManager) SetUnpinObserver(fn func(InstanceHash))

SetUnpinObserver registers fn to be called whenever an object version's last reader releases it. fn runs on the releasing goroutine, so it must return promptly and must not block; there is one observer at a time.

It exists for work that has to wait for a version to be unread -- the tiering uploader's deferred release of a local copy -- and would otherwise have to poll for it.

func (*StorageManager) StoreInline 

func (sm *StorageManager) StoreInline(ctx context.Context, instanceHash InstanceHash, meta *CacheMetadata, data []byte) (err error)

StoreInline stores small data inline in BadgerDB

func (*StorageManager) WriteBlocks 

func (sm *StorageManager) WriteBlocks(instanceHash InstanceHash, startOffset int64, data []byte) error

WriteBlocks writes encrypted blocks to disk storage This is the main write path for large objects

type StorageUsageKey 

type StorageUsageKey struct {
	StorageID   StorageID
	NamespaceID NamespaceID
}

StorageUsageKey combines storage ID and namespace ID for usage tracking

type StoreMode 

type StoreMode string

StoreMode identifies which subsystem owns a database. 

const (
	// StoreModeCache marks a database owned by the local cache.
	StoreModeCache StoreMode = "cache"
	// StoreModePStore marks a database owned by a pstore origin backend.
	StoreModePStore StoreMode = "pstore"
)

type TierBackend 

type TierBackend interface {
	// Put stores size bytes read from body under key, and reports the
	// object as the backend then holds it.  It reports an error if fewer
	// than size bytes arrive, or if the stored object is not size bytes
	// long, and makes a best effort to abort a partial write rather than
	// commit a short object.
	Put(ctx context.Context, key, contentType string, size int64, body io.Reader) (TierObjectInfo, error)

	// OpenRange returns a reader for the object starting at offset.  When
	// expect is non-nil, the read is pinned to that copy of the object --
	// by version where the backend keeps versions, otherwise by entity tag
	// -- and fails with ErrTierObjectChanged if the backend no longer holds
	// it.  A backend that can do neither ignores expect.
	OpenRange(ctx context.Context, key string, offset int64, expect *TierObjectInfo) (io.ReadCloser, error)

	// Stat reports an object as the backend currently holds it.  A missing
	// object is (TierObjectInfo{}, false, nil), not an error.
	Stat(ctx context.Context, key string) (info TierObjectInfo, exists bool, err error)

	// Delete removes an object.  Deleting a missing object is not an error.
	Delete(ctx context.Context, key string) error

	// List walks every object under the backend's prefix in ascending key
	// order, which is what lets the consistency sweep merge-join the
	// listing against hash-ordered metadata instead of buffering it.
	// A size the backend cannot report (an object still being written,
	// say) is passed as -1.
	List(ctx context.Context, fn func(key string, size int64, modified time.Time) error) error

	// DisplayURL is a human-readable identity for logs and for the
	// DiskMapping.Directory column.  It must not contain credentials.
	DisplayURL() string

	// Close releases any client resources.
	Close() error
}

TierBackend is the remote storage a cache tiers completed objects to.

It is deliberately a flat keyed blob store rather than a filesystem: the cache addresses objects by instance hash, needs ranged reads to serve them, and needs a key-ordered listing to reconcile the backend against its metadata. That is a much smaller contract than server_utils.OriginBackend (a webdav.FileSystem), and a different one -- an origin backend serves a POSIX-ish namespace and has no notion of handing a client a URL.

Keys are supplied by the caller and are backend-agnostic; see tierTarget.objectKey for the layout and why it is ordered.

type TierObjectInfo 

type TierObjectInfo struct {
	Size int64 `msgpack:"s"`
	// ETag is the backend's entity tag, opaque to the cache.  It is
	// compared only for equality: for S3 it is an MD5 of the content for a
	// simple upload but not for a multipart one, so it is not a checksum.
	ETag string `msgpack:"e,omitempty"`
	// Version is the backend's version identifier, set only where the
	// target keeps versions (an S3 bucket with versioning enabled).
	Version string `msgpack:"v,omitempty"`
	// ModTime is when the backend says the object was written.
	ModTime time.Time `msgpack:"m,omitempty"`
}

TierObjectInfo describes an object as a tiering target stored it. The cache records it when an object is tiered, and uses it to tell whether the target still holds the same bytes.

That matters because a tiered object is stored in plaintext and, unlike a local block, is not authenticated when read back: without a record, a same-length substitution in the bucket was undetectable. Any write to an object changes its entity tag, and where the bucket keeps versions, the version identifier names the exact bytes uploaded -- so reads can be pinned to them even after the key is overwritten.

type TierRedirector 

type TierRedirector interface {
	// RedirectURL returns a self-authenticating URL for a GET of key, valid
	// for expiry.  The URL must not require the caller to present any
	// Pelican credential.  When expect names a version, the URL should
	// fetch that version, so a client sent there receives the bytes the
	// cache uploaded even if the key has since been overwritten.
	RedirectURL(ctx context.Context, key string, expiry time.Duration, expect *TierObjectInfo) (string, error)
}

TierRedirector is implemented by backends that can hand a client a URL it can fetch directly, carrying its own authorization. This is the capability that lets the cache redirect instead of proxying; a backend that cannot do it (or is not configured to) simply does not implement this, and the cache serves the bytes itself.

type TierStaleUploadReaper 

type TierStaleUploadReaper interface {
	// ReapStaleUploads aborts incomplete uploads older than maxAge and
	// returns how many were aborted.
	ReapStaleUploads(ctx context.Context, maxAge time.Duration) (int, error)
}

TierStaleUploadReaper is implemented by backends with a notion of an incomplete upload that survives a crash and consumes space invisibly -- S3's multipart uploads being the motivating case. Backends where a failed write leaves nothing behind do not implement it.

type TierTargetConfig 

type TierTargetConfig struct {
	// ProviderURL names the backend in gocloud.dev/blob form --
	// "s3://bucket", "gs://bucket", "azblob://container", "mem://".  When
	// set it takes precedence over the S3-specific fields.
	ProviderURL string
	// ServiceUrl is the S3 endpoint (e.g. https://s3.us-east-1.amazonaws.com).
	ServiceUrl string
	// Region is the S3 region; defaults to us-east-1 when empty.
	Region string
	// Bucket is the bucket name.
	Bucket string
	// Prefix is an optional key prefix under which cache objects live.
	Prefix string
	// UrlStyle is "path" (default) or "virtual".
	UrlStyle string
	// AccessKeyfile / SecretKeyfile point at files holding static
	// credentials.  Both must be set together; when empty the ambient
	// credential chain is used.
	AccessKeyfile string
	SecretKeyfile string
	// WebDavUrl names a WebDAV collection (https://door.example.org:2880/data/pelican)
	// to tier to instead of an object store.  It is a separate key rather
	// than a ProviderURL scheme because a provider URL is opened through
	// gocloud.dev/blob, which has no WebDAV driver, and because it takes
	// different credentials.
	WebDavUrl string
	// TokenFile names a file holding the bearer token presented to the
	// WebDAV server.  It is re-read for every request, so a token rotated
	// in place is picked up.
	TokenFile string
	// DisableMacaroons stops the cache from asking a WebDAV target (dCache)
	// for macaroons, so its objects are always proxied.
	DisableMacaroons bool
	// MaxSize is the maximum bytes of cache data stored on the target.
	// Required -- remote capacity cannot be auto-detected.
	MaxSize uint64
	// Watermark overrides (percent of MaxSize); 0 means use the global default.
	HighWaterMarkPercentage int
	LowWaterMarkPercentage  int
	// AdoptExisting lets the cache take over a target whose identity object
	// it does not recognise -- one another cache wrote, or this cache wrote
	// before its database was lost.  Without it such a target is refused:
	// the cache assumes it owns everything under the prefix, so adopting a
	// target another cache is still using would have each delete the
	// other's objects.
	AdoptExisting bool
}

TierTargetConfig describes one remote storage target the cache tiers completed objects to.

A target is named by a gocloud.dev/blob provider URL, by the explicit S3 fields for S3-compatible services, or by a WebDAV collection URL. Nothing above this struct is S3-specific: the fields are kept because naming an S3 endpoint, region and credential files separately is how the origin's exports are configured too, and operators should not have to learn a second spelling.

func ParseTierTargetsConfig 

func ParseTierTargetsConfig() ([]TierTargetConfig, error)

ParseTierTargetsConfig reads the Cache.TieringTargets setting and returns the parsed target configurations. Returns nil (not an error) when the key is unset or empty.

func (*TierTargetConfig) DisplayURL 

func (c *TierTargetConfig) DisplayURL() string

DisplayURL returns a human-readable identity string for the target. It is persisted in the DiskMapping.Directory field, written to logs and used as a metric label, so it must never carry credentials. A provider URL that embeds them is refused when the configuration is parsed; the redaction here keeps that true for any value that reaches it some other way.

func (*TierTargetConfig) TransportScheme 

func (c *TierTargetConfig) TransportScheme() string

TransportScheme returns the URL scheme object data travels over, defaulting to https. For the S3 fields that is the endpoint's scheme; for a provider URL it is the scheme of an explicit endpoint override when one is given (s3://bucket?endpoint=http://...), since the cloud providers' own endpoints are https.

func (*TierTargetConfig) UsesVirtualHostStyle 

func (c *TierTargetConfig) UsesVirtualHostStyle() bool

UsesVirtualHostStyle reports whether S3 virtual-host addressing was asked for. Path style is the default because most S3-compatible services and custom endpoints require it.

type TierTokenSource 

type TierTokenSource interface {
	// Token returns the current token, or "" to send no Authorization header.
	Token(ctx context.Context) (string, error)
}

TierTokenSource supplies the bearer token a tiering backend presents to its storage service.

It is an interface, rather than a token file path threaded through the backend, so that a token acquired some other way -- an OAuth client credentials flow, say -- can be plugged in without the backend knowing. A source is consulted on every request, so it must be cheap and must do its own caching and refreshing.

type TierUploadIntent 

type TierUploadIntent struct {
	// TargetStorageID is the tiering target being uploaded to.
	TargetStorageID StorageID `msgpack:"tid"`
	// OriginalStorageID is the local (POSIX) storage the object's base
	// (chunk 0) lives on; used by crash recovery to clean up the local
	// file(s) when the upload committed but local deletion did not happen.
	OriginalStorageID StorageID `msgpack:"oid"`
	// OriginalChunkSizeCode and OriginalChunkLocations capture the object's
	// pre-relocation chunk layout so crash recovery can delete every local
	// chunk file after relocation flattened the metadata to a single S3
	// object.  Both are zero/empty for non-chunked objects.
	OriginalChunkSizeCode  ChunkSizeCode   `msgpack:"ocsc,omitempty"`
	OriginalChunkLocations []ChunkLocation `msgpack:"ocl,omitempty"`
	// Key is the full object key inside the bucket (including prefix).
	Key string `msgpack:"key"`
	// Size is the object ContentLength at upload time.
	Size int64 `msgpack:"sz"`
	// NamespaceID for usage accounting.
	NamespaceID NamespaceID `msgpack:"ns"`
	// StartedAt is when the upload began.
	StartedAt time.Time `msgpack:"st"`
	// RelocatedAt is set once the relocation transaction has committed, i.e.
	// once the remote copy -- not the local one -- is the object's
	// authoritative storage.  It is what lets crash recovery tell the two
	// halves of the lifecycle apart when the metadata record is gone: before
	// relocation the remote bytes and their capacity charge belong to this
	// intent and must be reclaimed, while afterwards they belong to the
	// object, so whatever deleted it has already accounted for them and only
	// the leftover local copy needs cleaning up.  Zero means "not yet".
	RelocatedAt time.Time `msgpack:"rat,omitempty"`
}

TierUploadIntent records an in-progress upload of a completed object to an tiering target. Serialized with msgpack under up:<instance_hash>.

type TierUploadIntentEntry 

type TierUploadIntentEntry struct {
	Hash   InstanceHash
	Intent *TierUploadIntent
}

TierUploadIntentEntry is one intent returned by ListTierUploadIntents.

type VerificationResult 

type VerificationResult struct {
	InstanceHash   string           `json:"instance_hash"`
	Valid          bool             `json:"valid"`
	Error          string           `json:"error,omitempty"`
	ChecksumStatus []ChecksumStatus `json:"checksum_status,omitempty"`
}

VerificationResult contains the result of a checksum verification.

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.
⌂ 换网址