crema

README

crema ☕️

A Go cache library with probabilistic revalidation and optional singleflight loading. It smooths refreshes near TTL expiry while deduplicating concurrent loads.

Features

• Smooth probabilistic revalidation near expiry
• Built-in singleflight loader (can be disabled)
• Zero external dependencies in the core module
• Pluggable storage (CacheProvider) and storage codecs (CacheStorageCodec)

Core functionality is covered by a high level of automated tests.

Revalidation Algorithm

Within the revalidation window, the cache reloads with probability

p(t)=1-e^{-kt}

where t is the remaining time. The steepness k is set so that $p(t)=0.999$ at the configured window boundary, smoothing spikes near expiry.

This design is inspired by the following references:

• Cache Stampede: Avoiding Hot Spots in Distributed Caching Systems
• Sometimes I Cache | The Cloudflare Blog

Installation

go get github.com/abema/crema

Go 1.22 or newer is recommended.

Quick Start

provider := newMemoryProvider[int]()
codec := crema.NoopCacheStorageCodec[int]{}
cache := crema.NewCache(provider, codec)

value, err := cache.GetOrLoad(context.Background(), "answer", time.Minute, func(ctx context.Context) (int, error) {
	// Database or computation logic here
	return 42, nil
})
if err != nil {
	panic(err)
}

println(value)

Usage Notes

• CacheProvider: Responsible for persistence with TTL handling. Works with Redis/Memcached, files, or databases.
• CacheStorageCodec: Encodes/decodes cached objects. Swap in JSON, protobuf, or your own codec.
• CacheObject: A thin wrapper holding Value and absolute expiry (ExpireAtMillis).

Options

• WithRevalidationWindow(duration): Set the revalidation window
• WithDirectLoader(): Disable singleflight and call loaders directly
• WithMaxLoadTimeout(duration): Set max duration for singleflight loaders (ignored with WithDirectLoader())
• WithLogger(logger): Override warning logger for get/set failures

Implementations

CacheProvider

Name	Package	Notes	Example
RistrettoCacheProvider	github.com/abema/crema/ext/ristretto	dgraph-io/ristretto backend with TTL support.	✅
RedisCacheProvider	github.com/abema/crema/ext/rueidis	Redis backend using rueidis.	✅
ValkeyCacheProvider	github.com/abema/crema/ext/valkey-go	Valkey (Redis protocol) backend.	✅
MemcachedCacheProvider	github.com/abema/crema/ext/gomemcache	Memcached backend with TTL handling.	-
CacheProvider	github.com/abema/crema/ext/golang-lru	hashicorp/golang-lru backend with default TTL.	-

CacheStorageCodec

Name	Package	Notes	Example
NoopCacheStorageCodec	github.com/abema/crema	Pass-through codec for in-memory cache objects.	-
JSONByteStringCodec	github.com/abema/crema	Standard library JSON encoding to []byte.	✅
JSONByteStringCodec	github.com/abema/crema/ext/go-json	goccy/go-json encoding to []byte.	-
ProtobufCodec	github.com/abema/crema/ext/protobuf	Protobuf encoding to []byte.	✅
BinaryCompressionCodec	github.com/abema/crema	Wraps another codec and zlib-compresses encoded bytes above a threshold.	✅

MetricsProvider

Name	Package	Notes	Example
NoopMetricsProvider	github.com/abema/crema	Embedded base used as the default metrics provider.	-

Concurrency

Cache is goroutine-safe as long as CacheProvider and CacheStorageCodec implementations are goroutine-safe.

Development

go generate
go test ./...

Tools

• cmd/plot-revalidation: SVG plot generator for revalidation curves

Why "crema"?

Crema is the golden foam that forms on top of a freshly pulled espresso coffee shot. Like crema that gradually dissipates over time, this cache library probabilistically refreshes entries, ensuring your data stays fresh without the overhead of deterministic expiration checks.

Documentation

Overview

Package crema provides a probabilistic cache with revalidation and loaders.

The cache can deduplicate concurrent loads via singleflight. Use WithMaxLoadTimeout to cap the execution time of singleflight loaders. When WithDirectLoader is used, the max load timeout is ignored and loaders run with the caller context.

Index

• Constants
• Variables
• type BaseMetricsProvider
  • func (BaseMetricsProvider) RecordCacheDelete(context.Context)
  • func (BaseMetricsProvider) RecordCacheGet(context.Context)
  • func (BaseMetricsProvider) RecordCacheHit(context.Context)
  • func (BaseMetricsProvider) RecordCacheSet(context.Context)
  • func (BaseMetricsProvider) RecordLoad(context.Context)
  • func (BaseMetricsProvider) RecordLoadConcurrency(context.Context, int)
• type BufferReleasePolicy
• type Cache
  • func NewCache[V any, S any](provider CacheProvider[S], codec CacheStorageCodec[V, S], ...) Cache[V, S]
• type CacheLoadFunc
• type CacheObject
• type CacheOption
  • func WithDirectLoader[V any, S any]() CacheOption[V, S]
  • func WithLogger[V any, S any](logger *slog.Logger) CacheOption[V, S]
  • func WithMaxLoadTimeout[V any, S any](duration time.Duration) CacheOption[V, S]
  • func WithMetricsProvider[V any, S any](metrics MetricsProvider) CacheOption[V, S]
  • func WithRevalidationWindow[V any, S any](duration time.Duration) CacheOption[V, S]
• type CacheProvider
• type CacheStorageCodec
  • func NewBinaryCompressionCodec[V any](inner CacheStorageCodec[V, []byte], compressThresholdBytes int) CacheStorageCodec[V, []byte]
• type JSONByteStringCodec
  • func (j JSONByteStringCodec[V]) CanReleaseBufferOnDecode() bool
  • func (j JSONByteStringCodec[V]) Decode(data []byte) (CacheObject[V], error)
  • func (j JSONByteStringCodec[V]) Encode(value CacheObject[V]) ([]byte, error)
• type MetricsProvider
• type NoopCacheStorageCodec
  • func (n NoopCacheStorageCodec[V]) Decode(data CacheObject[V]) (CacheObject[V], error)
  • func (n NoopCacheStorageCodec[V]) Encode(value CacheObject[V]) (CacheObject[V], error)
• type NoopMetricsProvider

Examples

• Cache

Constants

const (
	// DefaultCompressThresholdBytes is the default threshold size
	// above which values are compressed in BinaryCompressionCodec.
	DefaultCompressThresholdBytes = 1024 * 2 // 2 KiB

	CompressionTypeIDNone byte = 0x00
	CompressionTypeIDZlib byte = 0x01
)

Variables

var (
	ErrDecompressZeroLengthData     = errors.New("invalid data for decompression")
	ErrUnsupportedCompressionTypeID = errors.New("unsupported compression type ID")
)

Types

type BaseMetricsProvider

type BaseMetricsProvider struct{}

func (BaseMetricsProvider) RecordCacheDelete

func (BaseMetricsProvider) RecordCacheDelete(context.Context)

func (BaseMetricsProvider) RecordCacheGet

func (BaseMetricsProvider) RecordCacheGet(context.Context)

func (BaseMetricsProvider) RecordCacheHit

func (BaseMetricsProvider) RecordCacheHit(context.Context)

func (BaseMetricsProvider) RecordCacheSet

func (BaseMetricsProvider) RecordCacheSet(context.Context)

func (BaseMetricsProvider) RecordLoad

func (BaseMetricsProvider) RecordLoad(context.Context)

func (BaseMetricsProvider) RecordLoadConcurrency

func (BaseMetricsProvider) RecordLoadConcurrency(context.Context, int)

type BufferReleasePolicy

type BufferReleasePolicy interface {
	CanReleaseBufferOnDecode() bool
}

BufferReleasePolicy declares whether Decode can safely release buffer-backed input.

type Cache

type Cache[V any, S any] interface {
	// Get returns the cached entry for key.
	Get(ctx context.Context, key string) (CacheObject[V], bool, error)
	// Set stores a cached entry for key.
	Set(ctx context.Context, key string, value CacheObject[V]) error
	// Delete removes a cached entry for key.
	Delete(ctx context.Context, key string) error
	// GetOrLoad returns a cached value or uses loader when missing or revalidating.
	GetOrLoad(ctx context.Context, key string, ttl time.Duration, loader CacheLoadFunc[V]) (V, error)
}

Cache coordinates probabilistic revalidation with optional singleflight loading. Implementations are safe for concurrent use as long as CacheProvider and CacheStorageCodec implementations are goroutine-safe. Use NewCache to construct an implementation.

Example

provider := &testMemoryProvider[int]{items: make(map[string]CacheObject[int])}
codec := NoopCacheStorageCodec[int]{}
cache := NewCache(provider, codec)

value, err := cache.GetOrLoad(context.Background(), "answer", time.Minute, func(ctx context.Context) (int, error) {
	// Database or computation logic here
	return 42, nil
})
if err != nil {
	fmt.Println(err)

	return
}

fmt.Println(value)

Output:

42

func NewCache

func NewCache[V any, S any](provider CacheProvider[S], codec CacheStorageCodec[V, S], opts ...CacheOption[V, S]) Cache[V, S]

NewCache constructs a Cache with defaults and optional overrides.

type CacheLoadFunc

type CacheLoadFunc[V any] func(ctx context.Context) (V, error)

CacheLoadFunc loads a value when it is missing or needs revalidation.

type CacheObject

type CacheObject[V any] struct {
	// Value is the cached value.
	Value V
	// ExpireAtMillis is the absolute expiration time in milliseconds since epoch.
	ExpireAtMillis int64
}

CacheObject wraps a cached value with its absolute expiration time.

type CacheOption

type CacheOption[V any, S any] func(*cacheImpl[V, S])

CacheOption configures a Cache instance.

func WithDirectLoader

func WithDirectLoader[V any, S any]() CacheOption[V, S]

WithDirectLoader disables singleflight and calls loaders directly.

func WithLogger

func WithLogger[V any, S any](logger *slog.Logger) CacheOption[V, S]

WithLogger overrides the default logger used for cache warnings.

func WithMaxLoadTimeout

func WithMaxLoadTimeout[V any, S any](duration time.Duration) CacheOption[V, S]

WithMaxLoadTimeout sets the maximum duration allowed for loader execution. A non-positive duration disables the timeout.

func WithMetricsProvider

func WithMetricsProvider[V any, S any](metrics MetricsProvider) CacheOption[V, S]

WithMetricsProvider overrides the default metrics provider.

func WithRevalidationWindow

func WithRevalidationWindow[V any, S any](duration time.Duration) CacheOption[V, S]

WithRevalidationWindow sets the target revalidation window duration.

type CacheProvider

type CacheProvider[S any] interface {
	// Get retrieves a value from the cache by key.
	Get(ctx context.Context, key string) (S, bool, error)
	// Set stores a value in the cache with the specified key.
	Set(ctx context.Context, key string, value S, ttl time.Duration) error
	// Delete removes a value from the cache by key.
	Delete(ctx context.Context, key string) error
}

CacheProvider abstracts storage for encoded cache entries. Implementations must be safe for concurrent use by multiple goroutines.

type CacheStorageCodec

type CacheStorageCodec[V any, S any] interface {
	// Encode returns the cache object encoded into storage value.
	Encode(value CacheObject[V]) (S, error)
	// Decode reads the storage value into a cache object.
	Decode(data S) (CacheObject[V], error)
}

CacheStorageCodec encodes and decodes cache objects to storage values. Implementations must be safe for concurrent use by multiple goroutines.

func NewBinaryCompressionCodec

func NewBinaryCompressionCodec[V any](
	inner CacheStorageCodec[V, []byte],
	compressThresholdBytes int,
) CacheStorageCodec[V, []byte]

NewBinaryCompressionCodec returns a codec that conditionally compresses encoded values with zlib when they reach the threshold. A threshold of 0 always compresses, and a negative threshold disables compression.

type JSONByteStringCodec

type JSONByteStringCodec[V any] struct{}

JSONByteStringCodec marshals cache objects as JSON bytes.

func (JSONByteStringCodec[V]) CanReleaseBufferOnDecode

func (j JSONByteStringCodec[V]) CanReleaseBufferOnDecode() bool

func (JSONByteStringCodec[V]) Decode

func (j JSONByteStringCodec[V]) Decode(data []byte) (CacheObject[V], error)

Decode unmarshals JSON bytes into a cache object.

func (JSONByteStringCodec[V]) Encode

func (j JSONByteStringCodec[V]) Encode(value CacheObject[V]) ([]byte, error)

Encode marshals the cache object into JSON bytes without a trailing newline.

type MetricsProvider

type MetricsProvider interface {
	// RecordCacheHit is called when a cached value is successfully returned.
	RecordCacheHit(ctx context.Context)
	// RecordCacheGet is called when a cache lookup is attempted.
	RecordCacheGet(ctx context.Context)
	// RecordCacheSet is called when a cache write is attempted.
	RecordCacheSet(ctx context.Context)
	// RecordCacheDelete is called when a cache delete is attempted.
	RecordCacheDelete(ctx context.Context)
	// RecordLoad is called when a load is started by the leader.
	RecordLoad(ctx context.Context)
	// RecordLoadConcurrency is called when a load finishes with the inflight count.
	RecordLoadConcurrency(ctx context.Context, concurrency int)
}

MetricsProvider receives cache and loader events for instrumentation. Implementations must be safe for concurrent use and should avoid blocking.

type NoopCacheStorageCodec

type NoopCacheStorageCodec[V any] struct{}

NoopCacheStorageCodec passes CacheObject values through without encoding.

func (NoopCacheStorageCodec[V]) Decode

func (n NoopCacheStorageCodec[V]) Decode(data CacheObject[V]) (CacheObject[V], error)

Decode copies the cache object.

func (NoopCacheStorageCodec[V]) Encode

func (n NoopCacheStorageCodec[V]) Encode(value CacheObject[V]) (CacheObject[V], error)

Encode copies the cache object.

type NoopMetricsProvider

type NoopMetricsProvider struct {
	BaseMetricsProvider
}