cachex

package module
v0.5.1 Latest Latest
Warning

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

Go to latest
Published: Sep 28, 2026 License: MIT Imports: 18 Imported by: 0

README

cachex

Two-tier cache engine for Go: a sharded in-process LRU (L1) in front of a shared memcached tier (L2), with stampede protection, namespace invalidation, stats and a circuit breaker that degrades to L1-only when L2 is sick.

Docs: https://bakhod1r.github.io/cachex/ · API: https://pkg.go.dev/github.com/bakhod1r/cachex

Install

go get github.com/bakhod1r/cachex

Requires Go 1.25+.

Quick start

L1-only:

c, err := cachex.New()
if err != nil { return err }
defer c.Close()

_ = c.Set(ctx, "k", []byte("v"), time.Minute) // ttl <= 0 uses WithDefaultTTL
v, err := c.Get(ctx, "k")                     // errors.Is(err, cachex.ErrMiss) on miss
_ = c.Delete(ctx, "k")

With memcached:

import "github.com/bakhod1r/cachex/memcached"

store, err := memcached.New(memcached.Config{
    Servers:   []string{"localhost:11211"},
    Timeout:   100 * time.Millisecond, // default
    KeyPrefix: "app:",
})
if err != nil { return err }

c, err := cachex.New(cachex.WithL2(store), cachex.WithL1TTL(5*time.Second))
if err != nil { return err }
defer c.Close() // also closes the store

With Redis (Redis 6+, standalone or Cluster):

import cachexredis "github.com/bakhod1r/cachex/redis"

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
store, err := cachexredis.NewStore(cachexredis.StoreConfig{Client: rdb, KeyPrefix: "app:"})
if err != nil { return err }

inv, err := cachexredis.New(cachexredis.Config{Client: rdb}) // optional: same Redis for invalidation
c, err := cachex.New(cachex.WithL2(store), cachex.WithInvalidator(inv))

Runnable two-instance demo: cd redis && REDIS_ADDR=localhost:6379 go run ./example.

The Redis store supports everything the memcached one does: GetMulti (one MGET), compare-and-swap for cross-process Delete safety, Incr for namespace versions, and the distributed load lock. Each value carries an 8-byte version header used as the CAS token, so don't write to the store's keys with other clients. Unlike memcached, keys are binary-safe and used as-is, timeouts honor the request context, and a Cluster client works because every operation touches one key.

Timeouts: the memcached client has no context support. A context is checked only before an operation starts; once it is on the wire it is bounded by Config.Timeout, not by your deadline. Keep Timeout well below your request budget, and set MaxConcurrency so a slow memcached fails fast with ErrL2Unavailable instead of piling up goroutines.

Reads go L1 -> L2 (backfilling L1) -> miss. Set writes L2 then L1. Delete removes from both and returns L2 errors. With a CASStore (memcached, memstore) Delete writes a short-lived marker to L2 instead of removing the key; reads treat it as absent.

GetOrLoad and stampede protection

v, err := c.GetOrLoad(ctx, "user:42", time.Minute, func(ctx context.Context) ([]byte, error) {
    return db.LoadUser(ctx, 42) // bounded by WithLoadTimeout
})
  • Single-flight: concurrent misses for one key run the loader once per process (Stats.LoadsShared counts deduplicated callers).
  • XFetch early refresh: near expiry, a hit may trigger a background refresh with probability driven by loader duration and WithBeta (0 disables). Stats.EarlyRefreshes.
  • Stale window: with WithStaleWindow(d), an entry expired less than d ago is served while it refreshes in the background. Stats.StaleServed.
  • Stale-if-error: with WithStaleIfError(d), a loader failure within d of expiry is answered from the last good value instead of an error, so a source outage degrades to stale data. ErrNotFound is an authoritative answer and is never masked. Stats.StaleOnError.
  • Distributed lock: with WithDistributedLock(ttl, poll), a loader runs once per key across all processes (memcached add lock). Waiters poll until the value appears or ttl passes, then load themselves; any L2 problem fails open. Stats.LockWaits.
  • Negative caching: return cachex.ErrNotFound from the loader and set WithNegativeTTL(d); repeated lookups of absent records return ErrNotFound without loading. Stats.NegativeHits.
  • Background refreshes are capped at 16 concurrent; other loader errors are returned, never cached.

Zero-copy reads

err := c.GetView(ctx, "user:42", func(v []byte) error {
    return json.Unmarshal(v, &u) // v is valid only inside fn: don't modify or keep it
})

Get returns a private copy (one allocation); GetView hands the cached bytes to fn and doesn't allocate on an L1 hit. A concurrent Set of the same key never changes a slice that fn is already reading.

GetMulti

vals, err := c.GetMulti(ctx, []string{"a", "b", "c"}) // map holds only found keys

L1 answers first; remaining keys go to L2 in one round trip when the store implements cachex.MultiGetter (memcached does). Keys starting with cachex: are reserved.

_ = c.SetMulti(ctx, map[string][]byte{"a": a, "b": b}, time.Minute) // invalid key writes nothing

vals, err := c.GetOrLoadMulti(ctx, ids, time.Minute, func(ctx context.Context, missing []string) (map[string][]byte, error) {
    return db.LoadUsers(ctx, missing) // one query for all misses; absent keys = not found
})

GetOrLoadMulti has no single-flight across concurrent batches, stale window or early refresh. With WithNegativeTTL, keys the loader omits are cached as absent.

Loader panics and TTL jitter

  • A panicking loader returns cachex.ErrLoaderPanic (counted in Stats.LoadErrors) instead of crashing the process; nothing is cached.
  • WithTTLJitter(f) shortens each stored TTL by a random fraction in [0, f) so keys written together don't expire together.

Typed values

import "github.com/bakhod1r/cachex/typed"

users := typed.New(c, typed.JSON[User]())
u, err := users.GetOrLoad(ctx, "42", time.Minute, func(ctx context.Context) (User, error) {
    return db.LoadUser(ctx, 42)
})

Codecs: typed.JSON[V](), typed.String(), typed.Bytes(), or your own typed.Codec[V]. Backends: *cachex.Cache or *cachex.Namespace. An undecodable cached value is deleted and reloaded once by GetOrLoad; Get returns an error wrapping typed.ErrDecode.

Namespaces

users, err := c.Namespace("users") // non-empty, <=64 bytes, no ':'/space, not "cachex*"
_ = users.Set(ctx, "42", data, 0)
v, err := users.GetOrLoad(ctx, "42", time.Minute, load)

err = users.Invalidate(ctx) // drops every key in "users" on all nodes

Keys are stored as <name>:<version>:<key>; Invalidate atomically increments a version counter in L2 (cachex:ns:<name>:v) and frees this process's L1 copies immediately. If the version is unknown (L2 down, never fetched), Get misses, Set returns the error and GetOrLoad calls the loader without caching.

GetView, GetMulti, SetMulti and GetOrLoadMulti work the same way inside a namespace; loaders and returned maps always use your keys, not the internal <name>:<version>:<key> form.

Stats

s := c.Stats()
fmt.Println(s.L1Hits, s.L2Hits, s.Loads, s.Entries, s.Breaker, s.HitRatio())

Fields: L1Hits, L1Misses, L2Hits, L2Misses, L2Errors, L2Skipped, Loads, LoadErrors, LoadsShared, StaleServed, EarlyRefreshes, DecodeErrors, NegativeHits, LockWaits, PublishErrors, Evictions, Expirations, Entries, Bytes, Breaker ("closed", "open", "half-open"). Counters are read independently; a snapshot is not atomic across fields. HitRatio() is (L1Hits+L2Hits) / (L1Hits+L1Misses): every lookup touches L1 once, so L1 hits plus misses is the total lookup count.

Prometheus

A separate module keeps the core free of Prometheus dependencies:

import cachexprom "github.com/bakhod1r/cachex/prometheus"

prometheus.MustRegister(cachexprom.NewCollector(c, cachexprom.WithNamespace("myapp_cache")))

Metrics are read from Stats() once per scrape (*_hits_total{tier}, *_misses_total{tier}, *_l2_errors_total, *_loads_total, *_entries, *_breaker_state{state}, ...).

Events & logging

WithEvents installs hooks for logging, metrics and tracing. Nil fields are skipped, so an unset hook costs nothing.

c, _ := cachex.New(
	cachex.WithL2(store),
	cachex.WithEvents(cachex.MergeEvents(
		cachex.SlogEvents(slog.Default()),
		cachex.Events{
			OpStart: func(ctx context.Context, op cachex.Op, key string) context.Context {
				ctx, _ = tracer.Start(ctx, "cache."+op.String()) // ctx reaches L2 calls and the loader
				return ctx
			},
			OpEnd: func(ctx context.Context, info cachex.OpInfo, d time.Duration) {
				trace.SpanFromContext(ctx).End() // info.Tier: l1, l2, stale, miss, loaded, negative
			},
		},
	)),
)
  • OpStart/OpEnd: every public operation; OpInfo.Tier says who answered.
  • LoadEnd: each single-key loader call, including background refreshes.
  • L2Error, BreakerChange, LoaderPanic, PublishError: rare, actionable failures.
  • SlogEvents(l) logs only the failure hooks (L2/publish errors and breaker opening at Warn, other breaker transitions at Info, loader panics at Error), never per operation.

Hooks run synchronously and must not call back into the Cache.

OpenTelemetry

github.com/bakhod1r/cachex/otel (separate module) turns Events into spans and metrics.

ev, err := cachexotel.Events() // global TracerProvider/MeterProvider by default
c, _ := cachex.New(cachex.WithL2(store), cachex.WithEvents(cachex.MergeEvents(ev, cachex.SlogEvents(l))))
  • Span cachex.<op> per operation with cachex.tier and cachex.keys (batches). ErrMiss and ErrNotFound are not errors. Loader calls add a cachex.load span event.
  • Histograms cachex.op.duration (op, tier, error) and cachex.load.duration (error), in seconds.
  • Counters cachex.l2.errors (op), cachex.breaker.transitions (from, to), cachex.loader.panics, cachex.publish.errors.
  • Keys are never recorded unless WithKeyAttribute(true) (span attribute only; PII and cardinality).

Compression

L2 values can be compressed; L1 always keeps the uncompressed value, so L1 reads and GetView cost nothing extra. The codec lives in a separate module:

z, err := cachexzstd.New() // github.com/bakhod1r/cachex/zstd
c, _ := cachex.New(cachex.WithL2(store), cachex.WithCompression(z, 512))
  • Values of at least minSize bytes written by Set, SetMulti, GetOrLoad and GetOrLoadMulti are compressed; a value is stored raw if compression fails or doesn't make it smaller. Delete markers, cached ErrNotFound, load locks and namespace version counters are never compressed.
  • Wire format: header byte 3 (formerly reserved, always 0) holds the codec id; 0 means raw. Entries written by older versions decode unchanged. payloadLen is the stored (compressed) length.
  • WithDecompressors(codecs...) reads compressed entries without compressing writes.
  • An entry a node can't decode (unknown codec id, corrupt data) is a miss, counted in Stats.DecodeErrors; a loader then overwrites it. Older versions of cachex also see compressed entries as corrupt, i.e. a miss.
  • Custom codecs implement cachex.Compressor with a stable, non-zero ID() (cachexzstd.ID is 1) and must bound decompressed size themselves (cachexzstd.WithMaxDecodedSize, default 256 MiB).

Rolling deploy order:

  1. Deploy a version where every node can decode the codec (WithDecompressors(z)), with compression still off.
  2. Once no old node remains, enable WithCompression(z, minSize).

Rolling back reverses the order: disable WithCompression first, keep the decoder until compressed entries have expired.

Degradation (circuit breaker)

All L2 calls go through a breaker (10s window, min 20 requests, opens at 50% failures, 5s cooldown doubling on failed probe). Misses and semantic errors are not failures.

  • Breaker open: L2 calls are skipped (Stats.L2Skipped), Get falls back to L1/miss.
  • Set never fails because of L2; on L2 write failure L1 keeps the value only for WithDegradedL1TTL and Stats.L2Errors grows.
  • Delete and Namespace.Invalidate return L2 errors (cachex.ErrL2Unavailable when open), because a value left in L2 would be served to every node.

Consistency

L2 is the shared source of truth; L1 is per process.

  • After a Delete/Set on one node, other nodes may serve their L1 copy for at most WithL1TTL (default 10s).
  • After Namespace.Invalidate, other nodes see the new version within WithVersionTTL (default 2s). Old L1 entries need not expire first: the version is part of the key, so they simply become unreachable.
  • With WithInvalidator(inv) both bounds become "as fast as your message bus": Delete and Invalidate are broadcast and every subscribed process drops its L1 copies immediately. Implement cachex.Invalidator over Redis pub/sub, NATS, etc. (memstore.NewBus() is an in-process implementation). Delivery is best effort; the TTL bounds still hold.
  • Loads never undo a Delete/Set. A GetOrLoad whose loader read the source before a Delete or Set does not write its value back. Within one process this always holds. Across processes it needs a store implementing cachex.CASStore (memcached and memstore do): the load publishes with add/cas against the L2 state it saw before calling the loader, and Delete leaves a marker for WithDeleteMarkerTTL (default 2 x load timeout). A loader that ignores its context and outlives the marker TTL is not covered. GetOrLoadMulti has the per-process guard only. Without CASStore, a load on node A may re-cache a value that node B deleted, until its TTL.

Redis pub/sub adapter (separate module):

import cachexredis "github.com/bakhod1r/cachex/redis"

inv, err := cachexredis.New(cachexredis.Config{Client: redis.NewClient(&redis.Options{Addr: "localhost:6379"})})
c, err := cachex.New(cachex.WithL2(store), cachex.WithInvalidator(inv))

Other brokers, each a separate module with the same wire format and origin filtering:

Module Transport Delivery Notes
github.com/bakhod1r/cachex/redis Redis pub/sub at-most-once go-redis reconnects; messages during a disconnect are lost
github.com/bakhod1r/cachex/nats core NATS at-most-once not JetStream
github.com/bakhod1r/cachex/kafka Kafka topic, no consumer group at-least-once while subscribed topic must exist before Subscribe; each subscriber starts at the current end offsets
github.com/bakhod1r/cachex/rabbitmq fanout exchange, exclusive queue per subscriber at-most-once amqp091 doesn't reconnect; drops are reported via OnError
import cachexkafka "github.com/bakhod1r/cachex/kafka"

inv, err := cachexkafka.New(cachexkafka.Config{Brokers: []string{"localhost:9092"}})
defer inv.Close()

import cachexrabbitmq "github.com/bakhod1r/cachex/rabbitmq"

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
inv, err := cachexrabbitmq.New(cachexrabbitmq.Config{Conn: conn})

In every case a lost message only delays invalidation: the WithL1TTL and WithVersionTTL bounds still apply.

Options

Option Default Meaning
WithL2(Store) none (L1-only) Shared tier, e.g. memcached.New
WithL1MaxEntries(int) 100000 L1 entry cap (0 = unlimited)
WithL1MaxBytes(int64) 0 (unlimited) L1 approximate byte cap
WithShards(int) 0 (auto) L1 shard count
WithL1Frequency(bool) true Frequency-aware L1 eviction (count-min sketch)
WithDefaultTTL(d) 5m TTL when ttl <= 0
WithL1TTL(d) 10s Max L1 lifetime; cross-node staleness bound
WithDegradedL1TTL(d) 1s L1 TTL after a failed L2 write
WithStaleWindow(d) 0 (off) Serve expired values while refreshing
WithStaleIfError(d) 0 (off) Serve the last good value when the loader fails
WithBeta(float64) 1.0 XFetch strength (0 disables)
WithLoadTimeout(d) 5s Per-loader timeout
WithVersionTTL(d) 2s Namespace version cache; invalidation visibility bound
WithSweepInterval(d) 1s Expired-entry janitor (0 = lazy expiry only)
WithNegativeTTL(d) 0 (off) Cache a loader's ErrNotFound
WithDeleteMarkerTTL(d) 2 x load timeout L2 Delete marker lifetime (CASStore only); must exceed load timeout
WithDistributedLock(ttl, poll) off One loader per key across processes
WithInvalidator(Invalidator) none Broadcast invalidations between processes
WithPublishTimeout(d) load timeout Max wait for a broadcast; a stuck broker no longer blocks Delete
WithCompression(Compressor, minSize) off Compress L2 values >= minSize bytes
WithDecompressors(Compressor...) none Decode compressed L2 entries without compressing
WithClock(Clock) wall clock Injectable clock for tests

Example

go run ./example/

Testing

go test -race ./...

Integration against a real memcached:

docker run -d --rm -p 11211:11211 memcached:1.6-alpine
MEMCACHED_ADDR=localhost:11211 go test -race -tags=integration ./memcached/

Other modules: cd prometheus && go test -race ./...; Redis adapter: REDIS_ADDR=localhost:6379 go test -race -tags=integration ./... inside redis/; benchmarks against other Go caches live in bench/ (see bench/RESULTS.md).

Custom Store implementations can run the conformance suite: storetest.Run(t, newStore, advance). Implement cachex.CASStore too (the suite then checks it) to get the cross-process Delete guarantee.

Documentation

Overview

Package cachex is a two-tier cache engine: a sharded in-process LRU (L1) in front of a shared store such as memcached (L2), with stampede protection, namespaces and stats.

Consistency: L1 is per process. After a Delete on one node, other nodes may serve their L1 copy for up to the L1 TTL (WithL1TTL). L2 is the shared source of truth.

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrMiss          = errors.New("cachex: miss")
	ErrNotStored     = errors.New("cachex: not stored")
	ErrL2Unavailable = errors.New("cachex: l2 unavailable")
	ErrClosed        = errors.New("cachex: closed")
	ErrInvalidKey    = errors.New("cachex: invalid key")
	ErrValueTooLarge = errors.New("cachex: value too large")
	// ErrNotFound is returned by a Loader when the source has no such record. With
	// WithNegativeTTL the absence is cached and GetOrLoad returns ErrNotFound without loading.
	ErrNotFound = errors.New("cachex: not found")
	// ErrLoaderPanic wraps a panic recovered from a Loader. It is returned, never cached.
	ErrLoaderPanic = errors.New("cachex: loader panicked")
)

Sentinel errors. A miss is ErrMiss, never (nil, nil).

Functions

This section is empty.

Types

type CASStore

type CASStore interface {
	// Gets returns the value and an opaque token for CompareAndSwap; ErrMiss when absent.
	Gets(ctx context.Context, key string) (val []byte, token any, err error)
	// CompareAndSwap stores val only if key is unchanged since Gets returned token.
	// It returns ErrNotStored when the key changed and ErrMiss when it is gone.
	CompareAndSwap(ctx context.Context, key string, val []byte, token any, ttl time.Duration) error
}

CASStore is an optional Store capability. With it, loads publish with compare-and-swap and Delete leaves a short-lived marker, so a load that read its source before a Delete on any node cannot write the old value back afterwards. Without it that guard is per process only.

type Cache

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

Cache is safe for concurrent use. Create with New, release with Close.

func New

func New(opts ...Option) (*Cache, error)

New builds a Cache.

func (*Cache) Close

func (c *Cache) Close() error

Close stops background work and closes the L2 store. Idempotent.

func (*Cache) Delete

func (c *Cache) Delete(ctx context.Context, key string) (err error)

Delete removes key from both tiers. Unlike Set, an L2 failure is returned, because a value left in L2 would be served to every node.

func (*Cache) Get

func (c *Cache) Get(ctx context.Context, key string) ([]byte, error)

Get returns a copy of the cached value, or ErrMiss. L2 failures degrade to a miss.

func (*Cache) GetMulti

func (c *Cache) GetMulti(ctx context.Context, keys []string) (map[string][]byte, error)

GetMulti returns the cached values that are present; missing keys are simply absent. L1 answers first; the rest go to L2 in one round trip when the store implements MultiGetter.

func (*Cache) GetOrLoad

func (c *Cache) GetOrLoad(ctx context.Context, key string, ttl time.Duration, load Loader) ([]byte, error)

GetOrLoad returns the cached value or computes it with load, once per key per process. Near expiry it refreshes early (XFetch); within the stale window it serves the old value and refreshes in the background. Loader errors are returned and not cached.

func (*Cache) GetOrLoadMulti

func (c *Cache) GetOrLoadMulti(ctx context.Context, keys []string, ttl time.Duration, load MultiLoader) (map[string][]byte, error)

GetOrLoadMulti returns cached values for keys and loads the rest with one loader call. Unlike GetOrLoad there is no single-flight across concurrent batches, no stale window and no early refresh. Keys the source doesn't have are absent from the result.

func (*Cache) GetView added in v0.2.0

func (c *Cache) GetView(ctx context.Context, key string, fn func(val []byte) error) error

GetView calls fn with the cached value without copying it, or returns ErrMiss. The slice is valid only during fn and must not be modified or retained; copy it to keep it. A later Set of the same key doesn't change a slice already passed to fn. fn's error is returned. On an L1 hit GetView does not allocate.

func (*Cache) Namespace

func (c *Cache) Namespace(name string) (*Namespace, error)

Namespace returns a handle. name must be non-empty, at most 64 bytes, printable ASCII without ':' or spaces, and must not start with "cachex".

func (*Cache) Set

func (c *Cache) Set(ctx context.Context, key string, val []byte, ttl time.Duration) error

Set writes val to L2 then L1. ttl <= 0 uses the default TTL. A failed L2 write is not returned as an error (the cache is best effort); L1 then keeps the value only for the degraded TTL and Stats.L2Errors grows.

func (*Cache) SetMulti

func (c *Cache) SetMulti(ctx context.Context, items map[string][]byte, ttl time.Duration) (err error)

SetMulti writes every entry like Set. Keys are validated first, so an invalid key writes nothing.

func (*Cache) Stats

func (c *Cache) Stats() Stats

Stats returns a snapshot.

type Clock

type Clock interface{ Now() time.Time }

Clock is injectable so TTL tests never sleep.

type Compressor added in v0.3.0

type Compressor interface {
	ID() byte
	// Compress appends the compressed form of src to dst and returns the extended slice.
	Compress(dst, src []byte) ([]byte, error)
	// Decompress appends the decompressed form of src to dst and returns the extended slice.
	// It must return an error, not panic, on malformed input.
	Decompress(dst, src []byte) ([]byte, error)
}

Compressor compresses L2 values. Implementations must be safe for concurrent use.

The ID is written into every compressed L2 entry and selects the decoder on read, so it must be stable across versions and unique among the codecs a deployment uses. 0 is reserved for "raw".

type Events added in v0.3.0

type Events struct {
	// OpStart runs before a public operation; the returned context (e.g. carrying a span)
	// is used for L2 calls and the loader. Returning nil keeps ctx.
	OpStart func(ctx context.Context, op Op, key string) context.Context
	// OpEnd runs after the operation with the context OpStart returned.
	OpEnd func(ctx context.Context, info OpInfo, dur time.Duration)
	// LoadEnd runs after each single-key loader call, including background refreshes.
	LoadEnd func(ctx context.Context, key string, dur time.Duration, err error)
	// L2Error reports a store call that counted as a breaker failure. op is the Store
	// method, e.g. "get", "set", "add", "delete", "get_multi", "gets", "cas".
	L2Error func(op string, err error)
	// BreakerChange reports a breaker transition ("closed", "open", "half-open").
	BreakerChange func(from, to string)
	// LoaderPanic reports a recovered loader panic. key is empty for GetOrLoadMulti.
	LoaderPanic func(key string, recovered any)
	// PublishError reports a failed Invalidator.Publish.
	PublishError func(msg Invalidation, err error)
}

Events are optional hooks for logging, metrics and tracing. Nil fields are skipped. Hooks run synchronously on the calling goroutine (BreakerChange and L2Error may run on a background refresh), so they must be fast and must not call back into the Cache.

func MergeEvents added in v0.3.0

func MergeEvents(es ...Events) Events

MergeEvents combines hooks; each is called in argument order. OpStart contexts chain, so later hooks see earlier values. A hook no argument sets stays nil.

func SlogEvents added in v0.3.0

func SlogEvents(l *slog.Logger) Events

SlogEvents logs rare, actionable events to l: L2 errors and publish errors at Warn, breaker opening at Warn (other transitions at Info), loader panics at Error. Per-operation hooks stay nil, so the hot path is unaffected. Combine with other hooks via MergeEvents.

type Invalidation

type Invalidation struct {
	Keys      []string // single keys removed with Delete
	Namespace string   // namespace bumped with Invalidate
}

Invalidation tells other processes to drop L1 copies.

type Invalidator

type Invalidator interface {
	Publish(ctx context.Context, msg Invalidation) error
	// Subscribe registers fn and returns; the implementation delivers until ctx is done.
	Subscribe(ctx context.Context, fn func(Invalidation)) error
}

Invalidator broadcasts invalidations between processes (Redis pub/sub, NATS, ...), making cross-node L1 staleness sub-second instead of bounded by the L1 TTL. Delivery is best effort; the TTL bounds still hold for lost messages.

type Loader

type Loader func(ctx context.Context) ([]byte, error)

Loader computes a value on a cache miss.

type MultiCASGetter added in v0.3.0

type MultiCASGetter interface {
	GetsMulti(ctx context.Context, keys []string) (vals map[string][]byte, tokens map[string]any, err error)
}

MultiCASGetter is an optional CASStore capability: Gets for many keys in one round trip, used by GetOrLoadMulti. Absent keys are missing from both maps.

type MultiGetter

type MultiGetter interface {
	GetMulti(ctx context.Context, keys []string) (map[string][]byte, error)
}

MultiGetter is an optional Store capability used by Cache.GetMulti for one round trip. The result holds only keys that were found.

type MultiLoader

type MultiLoader func(ctx context.Context, keys []string) (map[string][]byte, error)

MultiLoader loads the given missing keys at once. Keys absent from the result don't exist; with WithNegativeTTL that absence is cached.

type Namespace

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

Namespace groups keys so they can be invalidated together with one atomic L2 counter bump. Data keys are "<name>:<version>:<key>"; the version lives in L2 at "cachex:ns:<name>:v". Other processes see an invalidation within the version TTL (WithVersionTTL).

func (*Namespace) Delete

func (n *Namespace) Delete(ctx context.Context, key string) error

Delete removes key from the namespace.

func (*Namespace) Get

func (n *Namespace) Get(ctx context.Context, key string) ([]byte, error)

Get reads key in the namespace. An unknown version (L2 down, never seen) is a miss.

func (*Namespace) GetMulti added in v0.4.0

func (n *Namespace) GetMulti(ctx context.Context, keys []string) (map[string][]byte, error)

GetMulti reads keys in the namespace; absent keys are omitted. An unknown version (L2 down, never seen) reports every key as absent.

func (*Namespace) GetOrLoad

func (n *Namespace) GetOrLoad(ctx context.Context, key string, ttl time.Duration, load Loader) ([]byte, error)

GetOrLoad is Cache.GetOrLoad inside the namespace. If the version is unknown the loader runs without caching, so callers still get a value.

func (*Namespace) GetOrLoadMulti added in v0.4.0

func (n *Namespace) GetOrLoadMulti(ctx context.Context, keys []string, ttl time.Duration, load MultiLoader) (map[string][]byte, error)

GetOrLoadMulti is Cache.GetOrLoadMulti inside the namespace; load always sees the caller's keys. If the version is unknown the loader runs without caching, so callers still get values.

func (*Namespace) GetView added in v0.4.0

func (n *Namespace) GetView(ctx context.Context, key string, fn func(val []byte) error) error

GetView is Cache.GetView inside the namespace: fn sees the value without a copy and must not retain it. An unknown version (L2 down, never seen) is a miss.

func (*Namespace) Invalidate

func (n *Namespace) Invalidate(ctx context.Context) error

Invalidate makes every key in the namespace unreachable on all nodes by bumping the version. Old entries age out of both tiers; this process also frees its L1 copies at once.

func (*Namespace) Set

func (n *Namespace) Set(ctx context.Context, key string, val []byte, ttl time.Duration) error

Set writes key in the namespace. When the version can't be read (L2 down and never fetched) nothing is written and the error is returned.

func (*Namespace) SetMulti added in v0.4.0

func (n *Namespace) SetMulti(ctx context.Context, items map[string][]byte, ttl time.Duration) error

SetMulti writes items in the namespace. When the version can't be read (L2 down and never fetched) nothing is written and the error is returned.

type Op added in v0.3.0

type Op uint8

Op names a public Cache operation in Events.

const (
	OpGet Op = iota
	OpGetView
	OpSet
	OpDelete
	OpGetMulti
	OpSetMulti
	OpGetOrLoad
	OpGetOrLoadMulti
)

Operations reported by OpStart and OpEnd.

func (Op) String added in v0.3.0

func (o Op) String() string

type OpInfo added in v0.3.0

type OpInfo struct {
	Op   Op
	Key  string
	Keys int
	Tier Tier
	Err  error
}

OpInfo describes a finished operation. Key is empty and Keys is the batch size for multi-key operations. For multi-key reads Tier is the farthest tier any key reached (L1 < L2 < Loaded); for writes it is TierMiss.

type Option

type Option func(*config) error

Option configures a Cache.

func WithBeta

func WithBeta(b float64) Option

WithBeta sets XFetch probabilistic early refresh strength (0 disables). Default 1.0.

func WithClock

func WithClock(cl Clock) Option

WithClock injects a clock (tests).

func WithCompression added in v0.3.0

func WithCompression(c Compressor, minSize int) Option

WithCompression compresses values of at least minSize bytes before writing them to L2 (Set, SetMulti, GetOrLoad, GetOrLoadMulti). A value is stored raw when compressing fails or doesn't make it smaller. L1 always holds the uncompressed value, so L1 reads and GetView are unaffected. c is also registered as a decoder (see WithDecompressors).

Roll out in two deploys: first ship a version where every node can decode the codec (WithDecompressors, or WithCompression with a huge minSize), then enable compression. Nodes that can't decode an entry treat it as a miss and count Stats.DecodeErrors.

func WithDecompressors added in v0.3.0

func WithDecompressors(cs ...Compressor) Option

WithDecompressors lets the cache read L2 entries compressed with these codecs without compressing its own writes. Uncompressed entries are always readable.

func WithDefaultTTL

func WithDefaultTTL(d time.Duration) Option

WithDefaultTTL is used when Set/GetOrLoad receive ttl <= 0. Default 5m.

func WithDegradedL1TTL

func WithDegradedL1TTL(d time.Duration) Option

WithDegradedL1TTL is the L1 TTL used when a write to L2 fails. Default 1s.

func WithDeleteMarkerTTL

func WithDeleteMarkerTTL(d time.Duration) Option

WithDeleteMarkerTTL sets how long Delete leaves a marker in L2 when the store implements CASStore. A load that read its source before the Delete fails to publish while the marker lives, so the TTL must exceed the load timeout. Default 2 * load timeout.

func WithDistributedLock

func WithDistributedLock(ttl, poll time.Duration) Option

WithDistributedLock makes GetOrLoad take a lock in L2 before loading, so across all processes one loader runs per key. Others poll every poll until the value appears or ttl passes, then load themselves. ttl should exceed the slowest load. Needs WithL2.

func WithEvents added in v0.3.0

func WithEvents(e Events) Option

WithEvents installs hooks for logging, metrics and tracing. Combine several with MergeEvents.

func WithInvalidator

func WithInvalidator(inv Invalidator) Option

WithInvalidator broadcasts Delete and Namespace.Invalidate to other processes.

func WithL1Frequency added in v0.2.0

func WithL1Frequency(on bool) Option

WithL1Frequency makes L1 eviction frequency-aware (default true): a count-min sketch tracks how often keys are read and written, and eviction prefers rarely used entries over popular ones. On a Zipf workload it raised hit ratio ~2 points for ~15% slower parallel Get. A value just written is never evicted by its own Set either way.

func WithL1MaxBytes

func WithL1MaxBytes(n int64) Option

WithL1MaxBytes bounds the in-process tier by approximate bytes (0 = unlimited).

func WithL1MaxEntries

func WithL1MaxEntries(n int) Option

WithL1MaxEntries bounds the in-process tier by entry count (0 = unlimited). Default 100000.

func WithL1TTL

func WithL1TTL(d time.Duration) Option

WithL1TTL caps how long a per-process copy lives. It is the cross-node staleness bound: L1 is eventually consistent with L2 within this window. Default 10s.

func WithL2

func WithL2(s Store) Option

WithL2 sets the shared tier (e.g. memcached.New). Without it the cache is L1-only.

func WithLoadTimeout

func WithLoadTimeout(d time.Duration) Option

WithLoadTimeout bounds each loader call. Default 5s.

func WithNegativeTTL

func WithNegativeTTL(d time.Duration) Option

WithNegativeTTL caches a Loader's ErrNotFound for d, so repeated lookups of absent records don't hit the source. 0 (default) disables.

func WithPublishTimeout added in v0.5.1

func WithPublishTimeout(d time.Duration) Option

WithPublishTimeout bounds how long Delete and Namespace.Invalidate wait for the invalidator to accept a broadcast (default: the load timeout). Broker clients such as franz-go retry until their context ends, so without a bound an unreachable broker would block every Delete. A publish that times out is reported through Events.PublishError; the local delete has already happened.

func WithShards

func WithShards(n int) Option

WithShards sets the L1 shard count (0 = auto).

func WithStaleIfError added in v0.4.0

func WithStaleIfError(d time.Duration) Option

WithStaleIfError lets GetOrLoad answer with the last good value for d past its expiry when the loader fails, so a source outage degrades to stale data instead of errors. It covers transport and server failures, not ErrNotFound, which is an authoritative answer from the source. The stale answer reports TierStale and counts StaleOnError. Default 0 (off): loader errors reach the caller.

func WithStaleWindow

func WithStaleWindow(d time.Duration) Option

WithStaleWindow lets GetOrLoad serve an expired value for d while it refreshes in the background.

func WithSweepInterval

func WithSweepInterval(d time.Duration) Option

WithSweepInterval sets the expired-entry janitor period (0 disables; expiry stays lazy). Default 1s.

func WithTTLJitter

func WithTTLJitter(f float64) Option

WithTTLJitter shortens every stored TTL by a random fraction in [0, f), so keys written together don't expire together. f must be in [0, 1). 0 (default) disables.

func WithVersionTTL

func WithVersionTTL(d time.Duration) Option

WithVersionTTL sets how long a namespace version is cached locally, which is the staleness window after Namespace.Invalidate on other processes. Default 2s.

type Stats

type Stats struct {
	L1Hits, L1Misses       uint64
	L2Hits, L2Misses       uint64
	L2Errors               uint64 // transport failures
	L2Skipped              uint64 // calls short-circuited by the open breaker
	Loads, LoadErrors      uint64
	LoadsShared            uint64 // callers deduplicated by single-flight
	StaleServed            uint64
	StaleOnError           uint64 // stale answers served because the loader failed
	EarlyRefreshes         uint64
	DecodeErrors           uint64
	NegativeHits           uint64 // cached ErrNotFound answers
	LockWaits              uint64 // loads that waited on another process's lock
	PublishErrors          uint64 // failed invalidation broadcasts
	LoadsDiscarded         uint64 // loads not cached because Set/Delete raced them
	Evictions, Expirations uint64
	Entries                int
	Bytes                  int64
	Breaker                string
}

Stats is a point-in-time snapshot. Counters are read independently, so a snapshot taken under load is not atomic across fields.

func (Stats) HitRatio

func (s Stats) HitRatio() float64

HitRatio is hits over lookups across both tiers.

type Store

type Store interface {
	// Get returns ErrMiss when the key is absent.
	Get(ctx context.Context, key string) ([]byte, error)
	// Set stores val. ttl <= 0 means no expiry.
	Set(ctx context.Context, key string, val []byte, ttl time.Duration) error
	// Add stores only if absent, else ErrNotStored.
	Add(ctx context.Context, key string, val []byte, ttl time.Duration) error
	// Delete of a missing key is success.
	Delete(ctx context.Context, key string) error
	// Incr adds delta to a decimal counter; ErrMiss when absent.
	Incr(ctx context.Context, key string, delta uint64) (uint64, error)
	Close() error
}

Store is the shared L2 tier (memcached or Redis in production, an in-memory fake in tests). Values are opaque encoded envelopes. Implementations must be safe for concurrent use.

type Tier added in v0.3.0

type Tier uint8

Tier reports who answered an operation.

const (
	TierL1       Tier = iota // in-process copy
	TierL2                   // shared store, backfilled into L1
	TierStale                // expired value served within the stale window
	TierMiss                 // nothing cached; also used by writes and failed calls
	TierLoaded               // loader ran (here or in a shared flight)
	TierNegative             // cached ErrNotFound
)

Tiers reported in OpInfo.

func (Tier) String added in v0.3.0

func (t Tier) String() string

Directories

Path Synopsis
Command example shows L1-only cachex usage: GetOrLoad, a Namespace, Invalidate and Stats.
Command example shows L1-only cachex usage: GetOrLoad, a Namespace, Invalidate and Stats.
internal
breaker
Package breaker is a small circuit breaker guarding the L2 cache tier so a dead L2 costs zero latency: while Open, Allow returns false immediately.
Package breaker is a small circuit breaker guarding the L2 cache tier so a dead L2 costs zero latency: while Open, Allow returns false immediately.
clock
Package clock provides a cheap, coarse wall-clock reading for hot paths.
Package clock provides a cheap, coarse wall-clock reading for hot paths.
counter
Package counter provides a striped uint64 counter for hot paths where many goroutines increment the same statistic.
Package counter provides a striped uint64 counter for hot paths where many goroutines increment the same statistic.
envelope
Package envelope is the binary wrapper stored in L2.
Package envelope is the binary wrapper stored in L2.
fastrand
Package fastrand exposes the runtime's cheap per-thread random source for hot paths that only need to spread work over stripes, not statistical quality.
Package fastrand exposes the runtime's cheap per-thread random source for hot paths that only need to spread work over stripes, not statistical quality.
flight
Package flight provides context-aware singleflight for cache stampede protection.
Package flight provides context-aware singleflight for cache stampede protection.
lru
Package lru implements a sharded, byte-valued LRU cache with per-entry TTL.
Package lru implements a sharded, byte-valued LRU cache with per-entry TTL.
sketch
Package sketch implements a lock-free count-min sketch with 4-bit saturating counters and periodic aging, used to estimate recent access frequency (TinyLFU).
Package sketch implements a lock-free count-min sketch with 4-bit saturating counters and periodic aging, used to estimate recent access frequency (TinyLFU).
kafka module
Package memcached implements cachex.Store over github.com/bradfitz/gomemcache.
Package memcached implements cachex.Store over github.com/bradfitz/gomemcache.
Package memstore is an in-memory cachex.Store fake with fault injection for tests.
Package memstore is an in-memory cachex.Store fake with fault injection for tests.
nats module
prometheus module
rabbitmq module
redis module
Package storetest is a conformance suite for cachex.Store implementations.
Package storetest is a conformance suite for cachex.Store implementations.
Package typed wraps a cachex backend with a Codec so callers work with values, not bytes.
Package typed wraps a cachex backend with a Codec so callers work with values, not bytes.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL