coherent

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 22, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

README

coherent

A local, in-process cache for Go that stays coherent across a fleet of processes.

Go Reference CI Release License

Go has great local caches (Otter, Ristretto) and great distributed caches (Redis). coherent fills the gap between them: a local cache whose entries are invalidated in near‑real‑time when the source of truth changes — so every process serves microsecond‑fast reads that are also fresh.

Why

A service that reads slowly-changing data (config, entitlements, reference data) from a central owner has two bad options: call the owner on every read (milliseconds of latency, a hard dependency, and read load that scales with your traffic), or cache locally with a TTL (stale for up to the TTL after every change).

coherent gives you the third option — a local cache plus a pluggable real-time invalidation channel:

  • Microsecond reads. Hits are served from local memory. (See benchmarks below.)
  • Fresh. When the owner changes a value, every process evicts its copy within milliseconds.
  • Correct under churn. Two simple rules (below) keep caches correct across reconnects and faults.
  • Degrades gracefully. If the invalidation channel drops, it falls back to TTL — never hard-fails.

Install

go get github.com/sagarsinghdev/coherent

The core module is zero-dependency (standard library only). Optional transports and backends live in separate modules so they never add weight to the core: the gRPC streaming source in examples/grpc and the Otter backend in contrib/otter.

Quickstart

cache := coherent.NewMemCache[string, string](coherent.Options[string, string]{
    MaxEntries: 10_000,
    TTL:        5 * time.Minute, // fallback consistency; invalidation is primary
})

// A source pushes invalidation events. Use MemorySource in-process, or a
// gRPC-streaming source across the network (see examples/grpc).
src := coherent.NewMemorySource[string](64)

handler := coherent.NewHandler[string, string](cache, src, nil)
go handler.Run(ctx) // applies invalidations to the cache

// Read path — a hit never leaves the process:
if v, ok := cache.Get("user:42"); ok {
    use(v)
}

For read-through loading with cache-stampede protection:

lc := coherent.NewLoadingCache(cache, func(ctx context.Context, id string) (string, error) {
    return ownerClient.FetchUser(ctx, id) // called once per key even under concurrent misses
})
v, err := lc.GetOrLoad(ctx, "user:42")

Architecture

flowchart LR
  subgraph owner["Owner service"]
    WR[write/mutation] --> PUB[("Redis Pub/Sub fan-out")]
    PUB --> CM[server.ConnectionManager]
    CM --> GS[gRPC stream]
  end
  subgraph consumer["Consumer process (coherent)"]
    GS --> SRC[InvalidationSource]
    SRC --> H["Handler: EvictKey / Clear"]
    H --> LC[(Cache)]
    APP[app.Get] -->|hit| LC
    LC -.->|"miss → Loader → Set"| OWN[fetch from owner]
  end

Three small interfaces:

Interface Role Provided
Cache[K,V] storage MemCache (bundled, zero-dep); adapt Otter etc.
InvalidationSource[K] push channel of events MemorySource (in-proc); gRPC/bus (see examples)
Handler[K,V] applies events to the cache bundled

The server subpackage provides the owner-side primitives — ConnectionManager (non-blocking broadcast) and ReplayService (watermark replay) — for building the service that emits events.

Correctness — the two rules

  1. Clear on reconnect. A reconnecting consumer can't know what changed while it was disconnected, so on every (re)connection the source emits a cache-clear event and the Handler flushes the cache; reads lazily re-fill. Brief re-warm, never a stale read across a gap.
  2. Idempotent, key-precise eviction. Each event evicts exactly one key; deleting an absent key is a no-op, so duplicate/overlapping events (e.g. from replay) are harmless.

On the owner side, a subscribe handler must register the consumer before starting replay, then drain events buffered during replay, then stream live — so no event is lost in the reconnect gap.

Backends

MemCache is the bundled default: a thread-safe LRU with optional TTL and Caffeine-style RemovalListener (RemovalCause: explicit / replaced / expired / size). It is correct and convenient. For high-concurrency production workloads, adapt a specialized cache (e.g. Otter's adaptive W-TinyLFU) behind the Cache interface — see contrib/otter.

Transports

  • In-process: MemorySource (bundled) — tests, local dev, single-binary usage.
  • gRPC streaming: the recommended cross-process transport — see examples/grpc for the .proto, the source adapter, and the owner-side wiring using the server primitives.
  • Message bus: consume a topic directly with per-pod consumer groups — pattern documented in examples/grpc.

Benchmarks

MemCache backend, Apple M2 Pro, Go 1.26. Numbers are indicative; reproduce with make bench (or go test -run='^$' -bench=. -benchmem ./bench/). Full methodology: bench/.

Benchmark Time/op Allocs/op
Get hit (single goroutine) ~17.5 ns 0
Set ~19 ns 0
Get hit (high parallel contention) ~154 ns 0
Invalidation apply (in-process, publish → evicted) ~4.9 µs 2

Local hits are tens of nanoseconds and allocation-free; invalidation apply is single-digit microseconds in-process (network transport adds its own RTT on top — still milliseconds versus a TTL measured in minutes). The bundled MemCache uses a single lock, so heavy multi-core read contention costs more than a lock-free cache would; that's the tradeoff for a zero-dependency default. Adapt Otter (see contrib/otter) when you need lock-free scaling — coherent's invalidation layer is unchanged.

Status

v0.x — the API may change before v1.0.0. Semantic versioning; changes tracked in CHANGELOG.md.

Roadmap: sharded MemCache backend, an official Otter adapter module, a ready-to-import gRPC source package, and OpenTelemetry metrics hooks.

Contributing

See CONTRIBUTING.md. Issues and PRs welcome.

License

Apache-2.0.

Documentation

Overview

Package coherent provides a local, in-process cache for Go that stays coherent across a fleet of processes via real-time invalidation.

Go has excellent local caches (for example Otter and Ristretto) and excellent distributed caches (for example Redis). coherent fills the gap between them: a local cache whose entries are invalidated in near-real-time when the source of truth changes, so every process serves microsecond-fast reads that are also fresh.

Design

coherent separates three concerns behind small interfaces:

  • Cache[K, V] the storage layer (a zero-dependency default is provided; adapt Otter or any other cache behind it).
  • InvalidationSource[K] a pluggable push channel of invalidation events (gRPC streaming, a message bus, or in-memory).
  • Handler[K, V] binds a source to a cache and applies events.

Correctness

Two rules make the design correct under process churn and network faults:

  1. Clear on reconnect. A reconnecting consumer cannot know what it missed while disconnected, so on any (re)connection the source emits a cache-clear event and the Handler flushes the cache. Subsequent reads lazily re-fill. This trades a brief re-warm for never serving a value that changed during the gap.

  2. Idempotent eviction. Deleting a key that is absent is a no-op, so duplicate events (for example from replay overlap) are harmless.

The server subpackage provides the owner-side primitives (a broadcast ConnectionManager and a watermark-based ReplayService) for building the service that pushes invalidation events.

Example

Example shows a consumer wiring: a local cache kept coherent by a Handler that applies invalidation events from a source. Here the source is in-process (MemorySource); in production it would be a gRPC-streaming source (see examples/grpc). 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/sagarsinghdev/coherent"
)

// Example shows a consumer wiring: a local cache kept coherent by a Handler that
// applies invalidation events from a source. Here the source is in-process
// (MemorySource); in production it would be a gRPC-streaming source (see
// examples/grpc).
func main() {
	cache := coherent.NewMemCache[string, string](coherent.Options[string, string]{
		MaxEntries: 10_000,
		TTL:        5 * time.Minute, // fallback consistency; invalidation is primary
	})

	src := coherent.NewMemorySource[string](16)
	handler := coherent.NewHandler[string, string](cache, src, nil)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = handler.Run(ctx) }()

	cache.Set("user:42", "Ada")
	fmt.Println(get(cache, "user:42"))

	// The owner service changed user:42; it publishes an invalidation.
	src.Publish(coherent.Event[string]{Key: "user:42", EventType: "updated", TimestampMs: 1})

	// Wait for the asynchronous eviction to be applied.
	for {
		if _, ok := cache.Get("user:42"); !ok {
			break
		}
		time.Sleep(time.Millisecond)
	}
	fmt.Println(get(cache, "user:42"))

}

func get(c *coherent.MemCache[string, string], k string) string {
	if v, ok := c.Get(k); ok {
		return v
	}
	return "<miss>"
}

Output:
Ada
<miss>

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Cache 

type Cache[K comparable, V any] interface {
	// Get returns the value for key and whether it was present.
	Get(key K) (V, bool)
	// Set stores value under key, overwriting any existing entry.
	Set(key K, value V)
	// Delete removes key. Deleting an absent key is a no-op.
	Delete(key K)
	// Clear removes all entries.
	Clear()
	// Len returns the current number of entries.
	Len() int
}

Cache is the storage layer coherent operates on. It is intentionally small so that any backend — the bundled MemCache, an Otter-backed adapter, or your own — can satisfy it.

Implementations must be safe for concurrent use by multiple goroutines.

type Event 

type Event[K comparable] struct {
	// Key identifies the entry to evict. Ignored when IsCacheClear is true.
	Key K
	// EventType is an optional, informational label such as "created",
	// "updated", or "deleted". coherent does not interpret it.
	EventType string
	// TimestampMs is the source-assigned event time in Unix milliseconds. It is
	// used as a watermark for replay on reconnect.
	TimestampMs int64
	// IsCacheClear requests that the entire cache be flushed. Sources emit this
	// on (re)connection and when a retention gap means missed events cannot be
	// replayed.
	IsCacheClear bool
}

Event is a single cache-invalidation signal delivered by an InvalidationSource.

An Event either targets one key (the common case) or requests a full flush via IsCacheClear. When IsCacheClear is true, Key is ignored.

type Handler 

type Handler[K comparable, V any] struct {
	// contains filtered or unexported fields
}

Handler binds an InvalidationSource to a Cache and applies incoming events. It is the correctness core of the library (see the package doc for the two rules it enforces).

func NewHandler 

func NewHandler[K comparable, V any](cache Cache[K, V], source InvalidationSource[K], logger *slog.Logger) *Handler[K, V]

NewHandler returns a Handler that applies events from source to cache. If logger is nil, slog.Default is used.

func (*Handler[K, V]) Run 

func (h *Handler[K, V]) Run(ctx context.Context) error

Run consumes the source's event stream until ctx is cancelled or the stream closes. It applies each event to the cache:

  • IsCacheClear -> cache.Clear() (flush on (re)connect / retention gap)
  • otherwise -> cache.Delete(Key) (idempotent, key-precise eviction)

Run blocks; call it from its own goroutine. It returns ctx.Err() on cancellation, or nil when the source stream closes cleanly.

type InvalidationSource 

type InvalidationSource[K comparable] interface {
	// Events returns the stream of invalidation events. It should be called
	// once; calling it again may return the same channel or panic, depending on
	// the implementation.
	Events(ctx context.Context) <-chan Event[K]
	// Watermark returns the highest TimestampMs observed so far, or 0 before any
	// event is received. Consumers pass this as the resume point on reconnect so
	// the source can replay only what was missed.
	Watermark() int64
}

InvalidationSource is a pluggable transport that pushes invalidation events to a consumer. Implementations own their own connection lifecycle and MUST emit an Event with IsCacheClear set to true on every fresh (re)connection, before resuming key-level events.

The returned channel is closed when the source is permanently done (for example, when the provided context is cancelled).

type Loader 

type Loader[K comparable, V any] func(ctx context.Context, key K) (V, error)

Loader fetches the authoritative value for a key on a cache miss (typically an RPC or HTTP call to the owner service).

type LoadingCache 

type LoadingCache[K comparable, V any] struct {
	Cache[K, V]
	// contains filtered or unexported fields
}

LoadingCache wraps a Cache with read-through loading and stampede protection: concurrent misses for the same key collapse into a single Loader call.

It embeds the underlying Cache, so Get/Set/Delete/Clear/Len remain available; use GetOrLoad for the read-through path.

Example

ExampleLoadingCache demonstrates read-through loading with cache-stampede protection: concurrent misses for the same key collapse into a single Loader call. 

package main

import (
	"context"
	"fmt"
	"sync/atomic"

	"github.com/sagarsinghdev/coherent"
)

func main() {
	var loads atomic.Int64
	lc := coherent.NewLoadingCache(
		coherent.NewMemCache[string, int](coherent.Options[string, int]{}),
		func(_ context.Context, key string) (int, error) {
			loads.Add(1)
			return len(key), nil // stand-in for an RPC to the owner service
		},
	)

	ctx := context.Background()
	v, _ := lc.GetOrLoad(ctx, "user:42") // miss -> loads
	fmt.Println("value:", v)
	_, _ = lc.GetOrLoad(ctx, "user:42") // hit -> no load
	fmt.Println("loads:", loads.Load())

}

Output:
value: 7
loads: 1

func NewLoadingCache 

func NewLoadingCache[K comparable, V any](cache Cache[K, V], loader Loader[K, V]) *LoadingCache[K, V]

NewLoadingCache returns a LoadingCache backed by cache, filling misses via loader.

func (*LoadingCache[K, V]) GetOrLoad 

func (lc *LoadingCache[K, V]) GetOrLoad(ctx context.Context, key K) (V, error)

GetOrLoad returns the cached value for key, or loads it via the Loader on a miss and caches the result. Concurrent misses for the same key share one load. A load error is returned and the value is not cached.

type MemCache 

type MemCache[K comparable, V any] struct {
	// contains filtered or unexported fields
}

MemCache is the bundled, zero-dependency Cache implementation: a thread-safe LRU cache with optional TTL and removal notifications. It is correct and convenient for moderate throughput. For high-throughput production workloads, adapt a specialized cache (for example Otter's adaptive W-TinyLFU) behind the Cache interface — see contrib/otter.

MemCache satisfies Cache[K, V].

Example (RemovalListener)

ExampleMemCache_removalListener shows Caffeine-style removal notifications: a RemovalListener is told why each entry left the cache. 

package main

import (
	"fmt"
	"time"

	"github.com/sagarsinghdev/coherent"
)

func main() {
	cache := coherent.NewMemCache[string, int](coherent.Options[string, int]{
		MaxEntries: 2,
		TTL:        time.Minute,
		OnRemoval: func(key string, _ int, cause coherent.RemovalCause) {
			fmt.Printf("removed %s: %s\n", key, cause)
		},
	})

	cache.Set("a", 1)
	cache.Set("a", 2) // overwrite -> CauseReplaced
	cache.Set("b", 1)
	cache.Set("c", 1) // exceeds MaxEntries -> LRU victim evicted (CauseSize)
	cache.Delete("c") // explicit removal -> CauseExplicit

}

Output:
removed a: replaced
removed a: size
removed c: explicit

func NewMemCache 

func NewMemCache[K comparable, V any](opts Options[K, V]) *MemCache[K, V]

NewMemCache returns a MemCache configured by opts.

func (*MemCache[K, V]) Clear 

func (c *MemCache[K, V]) Clear()

Clear removes all entries, reporting each as CauseExplicit.

func (*MemCache[K, V]) Delete 

func (c *MemCache[K, V]) Delete(key K)

Delete removes key if present, reporting it as CauseExplicit.

func (*MemCache[K, V]) Get 

func (c *MemCache[K, V]) Get(key K) (V, bool)

Get returns the value for key. Expired entries are removed and reported as a miss.

func (*MemCache[K, V]) Len 

func (c *MemCache[K, V]) Len() int

Len returns the number of entries currently held (including any that are expired but not yet lazily evicted).

func (*MemCache[K, V]) Set 

func (c *MemCache[K, V]) Set(key K, value V)

Set stores value under key. An overwrite reports the previous value as CauseReplaced; a size eviction reports the victim as CauseSize.

type MemorySource 

type MemorySource[K comparable] struct {
	// contains filtered or unexported fields
}

MemorySource is an in-process InvalidationSource. It is intended for tests, local development, and single-process usage where invalidations are produced in the same binary that consumes them. For cross-process invalidation, use a transport-backed source (see examples/grpc).

MemorySource satisfies InvalidationSource[K].

func NewMemorySource 

func NewMemorySource[K comparable](buffer int) *MemorySource[K]

NewMemorySource returns a MemorySource with the given channel buffer size.

func (*MemorySource[K]) Close 

func (s *MemorySource[K]) Close()

Close closes the event stream. It is safe to call once; further calls are no-ops.

func (*MemorySource[K]) Events 

func (s *MemorySource[K]) Events(_ context.Context) <-chan Event[K]

Events returns the event stream. The stream closes when Close is called.

func (*MemorySource[K]) Publish 

func (s *MemorySource[K]) Publish(ev Event[K])

Publish sends an event to consumers and advances the watermark. It blocks if the buffer is full. Publishing after Close panics, matching channel semantics.

func (*MemorySource[K]) Watermark 

func (s *MemorySource[K]) Watermark() int64

Watermark returns the highest TimestampMs published so far.

type Options 

type Options[K comparable, V any] struct {
	// MaxEntries is the maximum number of entries retained; 0 means unlimited.
	// When exceeded, the least-recently-used entry is evicted (CauseSize).
	MaxEntries int
	// TTL is the time-to-live for each entry, measured from its last write.
	// 0 disables expiry. Expiry is applied lazily on access.
	TTL time.Duration
	// OnRemoval, if set, is notified of every removal with its cause.
	OnRemoval RemovalListener[K, V]
	// contains filtered or unexported fields
}

Options configures a MemCache.

type RemovalCause 

type RemovalCause int

RemovalCause explains why an entry left the cache. It mirrors the causes used by Caffeine and is reported to a RemovalListener. 

const (
	// CauseExplicit means the entry was removed by Delete or Clear (including
	// invalidation events).
	CauseExplicit RemovalCause = iota
	// CauseReplaced means the entry's value was overwritten by Set.
	CauseReplaced
	// CauseExpired means the entry's TTL elapsed.
	CauseExpired
	// CauseSize means the entry was evicted to keep the cache within MaxEntries.
	CauseSize
)

func (RemovalCause) String 

func (c RemovalCause) String() string

String returns a human-readable name for the cause.

type RemovalListener 

type RemovalListener[K comparable, V any] func(key K, value V, cause RemovalCause)

RemovalListener is invoked after an entry is removed from the cache. It is called without the cache lock held, but it MUST NOT block for long and MUST NOT call back into the same cache instance.

Source Files

View all Source files

Directories

Path Synopsis
Package bench holds reproducible benchmarks for coherent: the local cache-hit cost and the end-to-end invalidation apply latency.
 Package bench holds reproducible benchmarks for coherent: the local cache-hit cost and the end-to-end invalidation apply latency.
contrib
otter module
examples
grpc module
Package server provides the owner-side primitives for building a service that pushes cache-invalidation events to coherent consumers.
 Package server provides the owner-side primitives for building a service that pushes cache-invalidation events to coherent consumers.

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.