Help us shape the future of Go! Take the 2026 Go Developer Survey

redisstore

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Sep 30, 2026 License: Apache-2.0 Imports: 31 Imported by: 0

Documentation

Overview

Package redisstore implements the Redis-backed port.SessionStore, port.EventLog, port.PrunableStore, and port.ToolCallRecorder for the cloud-native posture (ADR 0048, mecak8s). The agent pods are storage-free: session snapshots and the durable event log live in Redis as a managed service, and this adapter is the single store the server relay binds when the operator selects a Redis backend (a transport alternative to the local jsonlstore).

It REUSES the existing storage formats — it is a TRANSPORT, not a format:

  • snapshots are encoded with engine/adapter/sessnap (sessnap-json/1), shared with jsonlstore and the gRPC driver; and
  • the event-log record is the same {"v":<tag>,"ev":<json>} envelope shape as jsonlstore, with its own format tag (redisstore-eventlog/1) so a forward-incompatible log fails loud on Read (an unknown tag is an error); a gap marker is a sibling tag on that same shape, never an event.

No client-side mutex is needed: Redis serializes commands single-threaded and HSET/HGET/XADD are atomic, so the in-process sync.Mutex that jsonlstore carries is absent here. The adapter is validated by the SAME conformance suites as jsonlstore (storeconformance + eventlogconformance, including the cursor table), exercised offline against an in-process miniredis so `task test` needs no live broker.

The event log is a STREAM, not a LIST (ADR 0250): XADD IDs are opaque, monotonic and durable, so they serve as cursors directly, and XREAD BLOCK is a cross-process blocking follow a LIST cannot express. Older Redis namespaces are not inspected or migrated.

DURABILITY CAVEAT: Append/Save call XADD/HSET synchronously and return only once Redis acknowledges the command, but Redis's own persistence config (RDB snapshotting vs AOF fsync policy) determines durability-on-crash. An operator selecting this backend must configure Redis persistence to match their durability requirement; the adapter makes no durability claim beyond "Redis accepted the write".

Index

Constants

View Source 
const EventLogFormat = "redisstore-eventlog/1"

EventLogFormat is the per-record format tag written on every event-log record. It versions the on-disk encoding so Read can reject an unknown tag as an infra error (a forward-incompatible log must fail loud, not silently skip).

View Source 
const (

	// EventLogGapFormat tags a gap marker: a position where an append is KNOWN
	// to have failed.
	//
	// A sibling tag rather than a new event type, mirroring the jsonlstore
	// envelope: a gap is a fact about DELIVERY, not something that happened in
	// the run, so it never becomes a session.Event (ADR 0250 decision 5). The
	// legacy Read skips it; cursor readers see it via ReadAfter.
	EventLogGapFormat = "redisstore-eventlog-gap/1"
)

Variables

View Source 
var ErrNotFound = fmt.Errorf("redisstore: session not found: %w", port.ErrSessionNotFound)

ErrNotFound is returned by Load when no snapshot exists for the id. It wraps port.ErrSessionNotFound so a consumer that may not import this adapter can distinguish not-found from an infra failure via errors.Is.

View Source 
var ErrScheduleNotFound = fmt.Errorf("redisstore: schedule not found: %w", port.ErrScheduleNotFound)

ErrScheduleNotFound is returned by Load/Delete/Claim/RecordFire/LoadFire when no schedule (or fire) exists under the requested name/id. It wraps port.ErrScheduleNotFound so a consumer that may not import this adapter can distinguish not-found from an infra failure via errors.Is, the same discipline the session store applies for ErrNotFound (and the jsonl/mem schedule stores apply for their ErrNotFound/ErrScheduleNotFound). It is a SEPARATE sentinel from the session-store ErrNotFound so a reader can tell which seam missed.

Functions

func NewClient 

func NewClient(cfg Config) (redis.UniversalClient, error)

NewClient builds a standalone redis.UniversalClient using the SAME connection policy (address, credential files, TLS) as NewWithConfig, minus the session-store scaffolding (metadata index, reload lifecycle, client generations). It is for a caller that needs its OWN Redis connection to the SAME managed instance — e.g. the bundled MCP-broker's embedded OAuth authorization server, which stores under a distinct key prefix — without reaching into this Store's internal, rotation-managed client. The caller owns the returned client's lifecycle (Close it when done).

Credentials/CA material are read ONCE, here, at construction: unlike the main session-store client (NewWithConfig with cfg.reloadEnabled()), this client does NOT watch its credential/CA files and does NOT hot-reload after an ACL or CA rotation — go-redis reconnects using the SAME static Username/Password/TLS baked into redisconn.Config, which exposes no credentials-provider or reload hook. After a rotation, this client's reconnects fail with the stale material while /readyz (driven by the main store's reload-aware client) stays green, masking the failure as broker-only OAuth breakage. A rotation therefore requires restarting the process for this specific client. Document this limitation at every call site's own operator-facing docs rather than implying parity with NewWithConfig.

Types

type Config 

type Config struct {
	Addr         string
	UsernameFile string
	PasswordFile string
	// CAFile is a PEM CA bundle path. It REPLACES the system trust store, so a
	// managed service with a private CA needs it and one with a publicly-rooted
	// certificate does not.
	CAFile string
	// TLS enables verified TLS against the host's system trust store. CAFile
	// takes precedence when both are set.
	TLS            bool
	AllowPlaintext bool
	// FollowPoolSize is the hard connection bound for the isolated client used
	// only by ReadAfter calls with Follow set. Zero selects the default of 32.
	FollowPoolSize int
	// MaxFollowers is the process-local number of concurrently admitted follow
	// iterators. Zero selects the default of 32.
	MaxFollowers int
	Diagnostics  port.Diagnostics
}

Config configures a Redis connection using Kubernetes Secret-mounted files. Address-only configuration is intentionally supported for the disposable local and Kind path, and requires an explicit AllowPlaintext opt-in. Any credential requires VERIFIED TLS — either the host's system trust store (TLS) or an explicit PEM CA bundle (CAFile). Certificate verification is never disabled.

Client-certificate (mTLS) authentication is NOT supported. The shared toolhive-core Redis layer this adapter delegates to exposes no client-certificate field; see ADR 0233 for the decision and the upstream tracking issue.

type Store 

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

Store is the Redis-backed SessionStore + EventLog + PrunableStore + ToolCallRecorder. Every operation leases one replaceable client generation; the manager lock is held only for acquisition/publication, never Redis I/O.

func New 

func New(addr string) (*Store, error)

New connects to a plaintext, unauthenticated Redis broker. It is retained for local and Kind fixtures; production callers should use NewWithConfig with Secret-mounted credential files and verified TLS.

func NewWithConfig 

func NewWithConfig(cfg Config) (*Store, error)

NewWithConfig connects to Redis and pings it to fail fast. Secret values are read only from their mounted files and are never included in returned errors.

Client construction, TLS assembly, dial/read/write timeout defaults, and the connectivity Ping are delegated to the shared toolhive-core Redis layer (ADR 0233). What stays here is the half that layer deliberately leaves to its callers: reading credentials from mounted files, and the policy that a credential implies verified TLS.

func (*Store) Append 

func (st *Store) Append(ctx context.Context, id session.SessionID, ev session.Event) error

Append durably records ev under id as a format-tagged JSON record on the per-session event STREAM (XADD). It satisfies port.EventLog. The event is marshalled to its session.Event JSON verbatim (already redacted at the relay) and wrapped in the {"v":"redisstore-eventlog/1","ev":...} envelope so Read can validate the format. XADD preserves append order, so Read returns events in the exact order Append received them.

It delegates to AppendEvent and drops the cursor. There is deliberately ONE write path: two would have to agree on the datatype, and the first append through the other one would meet a WRONGTYPE — the failure mode a "leave the old path alone" migration produces. A caller that wants the position calls AppendEvent; this signature exists for the shipped port.

func (*Store) AppendEvent 

func (st *Store) AppendEvent(ctx context.Context, id session.SessionID, ev session.Event) (port.Cursor, error)

AppendEvent durably records ev and returns the cursor positioned after it. It satisfies port.CursorEventLog.

It carries the same durability obligation as Append — a nil error means the record is on stable storage — and the same at-most-once, no-retry contract.

func (*Store) AppendGap 

func (st *Store) AppendGap(ctx context.Context, id session.SessionID, reason string) (port.Cursor, error)

AppendGap durably records a gap marker at the next position. It satisfies port.CursorEventLog.

This is the best-effort, cross-process tier of ADR 0250's three-tier append-gap guarantee: when an append fails, one gap marker is attempted, and if it lands then every watcher everywhere learns of the gap deterministically rather than silently skipping it. It covers the LIKELY failure — one rejected or unencodable record — and not a total backend outage, which by construction cannot record its own failure.

func (*Store) Close 

func (st *Store) Close() error

Close stops credential reload, rejects new work, and gives pinned operations a fixed grace interval to finish. Followers are store-owned: they are cancelled and joined, then only isolated follow clients are force-closed when a read ignores cancellation. Durability clients retain generation-lease semantics.

func (*Store) Create 

func (st *Store) Create(ctx context.Context, s *session.Session) error

Create atomically publishes a snapshot and its derivative metadata only when no authoritative Redis session key exists for s.ID.

func (*Store) CreateWorkspace 

func (st *Store) CreateWorkspace(ctx context.Context, scope string) (*Workspace, error)

CreateWorkspace creates or opens a principal namespace. Reopening an existing namespace validates its format marker; it never repairs malformed state.

func (*Store) Delete 

func (st *Store) Delete(ctx context.Context, id session.SessionID) error

Delete removes the session snapshot, derivative metadata, event log, and tool-call sidecar. It is idempotent and completes in one atomic script.

func (*Store) DeleteReadLedger 

func (st *Store) DeleteReadLedger(ctx context.Context, id session.SessionID) error

DeleteReadLedger removes every entry for the session's read-ledger scope (the whole Redis hash). Canonical session deletion already removes this key atomically with the other session sidecars; this method supports an explicit ledger-only reset. It is idempotent — deleting an already-empty or never-used scope is a no-op.

func (*Store) DeleteSessionIfUnchanged 

func (st *Store) DeleteSessionIfUnchanged(ctx context.Context, expected port.SessionDiscoveryMeta) (bool, error)

DeleteSessionIfUnchanged atomically compares the indexed durable row and removes the snapshot plus sidecars in one Redis script.

func (*Store) List 

func (st *Store) List(ctx context.Context) ([]port.StoredSession, error)

List returns every stored session's id and SAVE-time mtime. It SCANs the keyspace for current session keys, then HGETs the mtime field for each. The mtime is the value written at Save time, so two Lists with no intervening Save agree exactly (the stable-across-reads invariant). SCAN is cursor-based and non-blocking; a corrupt mtime field (absent or unparseable) is skipped best-effort rather than failing the whole inventory.

func (*Store) Load 

func (st *Store) Load(ctx context.Context, id session.SessionID) (*session.Session, error)

Load reads the snapshot blob for id and restores it. A missing key (redis.Nil on HGET) wraps port.ErrSessionNotFound with the id in the message.

func (*Store) OpenWorkspace 

func (st *Store) OpenWorkspace(ctx context.Context, scope string) (*Workspace, error)

OpenWorkspace reattaches an existing namespace and fails closed when it is missing.

func (*Store) PageSessionMetadata 

func (st *Store) PageSessionMetadata(ctx context.Context, request port.SessionMetadataPageRequest) (port.SessionMetadataPage, error)

PageSessionMetadata reads one owner-filtered keyset page from the derivative Redis metadata index. It never reads a snapshot blob or traverses rows before the cursor; legacy stores without a complete index report unsupported.

func (*Store) Ping 

func (st *Store) Ping(ctx context.Context) error

Ping checks the Redis broker is reachable. It is the readyz health probe a storage-free deployment (mecak8s) consults on /readyz: if Redis is down the endpoint controller removes the pod. It uses a short timeout so a stalled broker fails the probe quickly rather than wedging readiness.

func (*Store) Read 

func (st *Store) Read(ctx context.Context, id session.SessionID) iter.Seq2[session.Event, error]

Read yields the session's recorded events in APPEND order. It satisfies port.EventLog. A MISS (no event key) yields an EMPTY sequence: absence is data, not an error. A genuine fault — an undecodable record, an unknown format tag, or a Redis error — is yielded as the error on a zero-value event and the consumer stops (the standard iter.Seq2 error idiom).

It reads only the current STREAM representation. Unsupported old datatypes are rejected without mutation.

GAP MARKERS ARE SKIPPED. This port's shipped contract is that it returns EVENTS, and a gap is a delivery envelope (ADR 0250 decision 5) that the event-sourced fold of ADR 0038 would choke on. Cursor readers see gaps via ReadAfter.

func (*Store) ReadAfter 

func (st *Store) ReadAfter(ctx context.Context, id session.SessionID, after port.Cursor, opts port.ReadOptions) iter.Seq2[port.LogRecord, error]

ReadAfter yields the log's records strictly after the given cursor. It satisfies port.CursorEventLog.

The zero cursor starts from the beginning. A cursor from a superseded log generation yields ErrCursorExpired; one that cannot be decoded or resolved to a stream ID yields ErrCursorMalformed. Neither is ever coerced to a position: resuming from approximately the right place is indistinguishable from resuming from the right one until data is already lost.

func (*Store) ReadLedger 

func (st *Store) ReadLedger(id session.SessionID) tool.ReadLedger

ReadLedger returns a tool.ReadLedger bound to one session scope, sharing this Store's Redis client. Independently constructed handles for the SAME session id (including from separate *Store instances/process — see New) all read and write the same durable Redis hash, so a version recorded through one handle is visible after reopening another (ADR 0294 Scenario 2).

func (*Store) ReadSessionLineage 

func (st *Store) ReadSessionLineage(ctx context.Context, query port.SessionLineageQuery) (port.SessionLineageResult, error)

ReadSessionLineage returns deterministic direct edges without loading snapshots.

func (*Store) Save 

func (st *Store) Save(ctx context.Context, s *session.Session) error

Save stores a sessnap-encoded snapshot of s under the session key, stamping the current time as the mtime field so List can report the SAVE time (not a fresh time.Now() at list time — the stable-mtime conformance invariant). HSET overwrites the blob field, so a second Save replaces the first (the overwrite contract).

func (*Store) ScheduleStore 

func (st *Store) ScheduleStore() port.ScheduleStore

ScheduleStore returns a port.ScheduleStore backed by the SAME Redis client as the session store (a sibling struct sharing the connection). Composition discovers it via type-assertion on this accessor — NOT by asserting the *Store itself implements port.ScheduleStore (the schedule store is a separate concern; the accessor keeps session-store and schedule-store methods from bloating one struct, the jsonlstore.ScheduleStore precedent — and the way PrunableStore is discovered on the store itself but here the schedule store is a sibling struct, not the session store). A caller that does not need schedules never calls this; the byte-identical default is no schedules.

The schedule store carries NO client-side mutex: Redis serializes commands single-threaded, and the Claim path is a Lua CAS (EVAL) that is the cross-replica at-most-once fence — the multi-host counterpart of the single-process mutex the jsonl schedule store carries. See schedulestore.go.

func (*Store) SupportsSessionActivityProjection 

func (*Store) SupportsSessionActivityProjection() bool

SupportsSessionActivityProjection reports that the Redis metadata index atomically reflects the latest snapshot's activity.

func (*Store) ToolCall 

func (st *Store) ToolCall(id session.SessionID, call session.ToolCall, result session.ToolResult, queued, took time.Duration)

ToolCall appends a structured tool-call record to the per-session tool list (RPUSH). It satisfies port.ToolCallRecorder. Errors are intentionally swallowed (the port has no error return); the record is best-effort durable.

type Workspace 

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

Workspace is a principal-scoped, shell-less virtual filesystem backed by Redis.

func (*Workspace) AuthorityResourcePath 

func (*Workspace) AuthorityResourcePath(p string) (string, string, error)

AuthorityResourcePath projects a confined virtual path for policy evaluation. "." names the workspace root itself — the same root ReadDir/cleanWorkspaceDir already accept for a directory listing — and maps directly to workspaceRoot without relaxing the rejection of empty, absolute, NUL-containing, or ".."-traversing paths, which still flow through cleanWorkspacePath unchanged.

func (*Workspace) CopyFile 

func (w *Workspace) CopyFile(ctx context.Context, source, destination string) (tool.FileVersion, error)

CopyFile atomically copies one regular file to an absent destination.

func (*Workspace) CreateFile 

func (w *Workspace) CreateFile(ctx context.Context, p string, data []byte) (tool.FileVersion, error)

CreateFile atomically creates a file and refuses an existing path.

func (*Workspace) Glob 

func (w *Workspace) Glob(ctx context.Context, pattern string) ([]string, error)

Glob returns sorted paths matching a doublestar pattern.

func (*Workspace) Grep 

func (w *Workspace) Grep(ctx context.Context, pattern, pathGlob string) ([]tool.GrepMatch, error)

Grep searches non-binary files and returns deterministic path/line matches.

func (*Workspace) Read 

func (w *Workspace) Read(ctx context.Context, p string) ([]byte, error)

func (*Workspace) ReadDir 

func (w *Workspace) ReadDir(ctx context.Context, p string) ([]tool.FileInfo, error)

ReadDir returns immediate children of a prefix-derived virtual directory.

func (*Workspace) ReadVersion 

func (w *Workspace) ReadVersion(ctx context.Context, p string) ([]byte, tool.FileVersion, error)

ReadVersion atomically reads file content and its opaque content version.

func (*Workspace) Remove 

func (w *Workspace) Remove(ctx context.Context, p string) error

Remove atomically removes one regular file and refuses non-recursive removal of a prefix-derived directory.

func (*Workspace) Rename 

func (w *Workspace) Rename(ctx context.Context, oldPath, newPath string) error

Rename atomically moves one file or a complete virtual-directory prefix and refuses an existing destination.

func (*Workspace) ReplaceFile 

func (w *Workspace) ReplaceFile(ctx context.Context, p string, old tool.FileVersion, data []byte) (tool.FileVersion, error)

ReplaceFile atomically replaces a file only when its version still matches.

func (*Workspace) Root 

func (*Workspace) Root() string

Root returns the stable virtual root; the principal namespace stays private.

func (*Workspace) Stat 

func (w *Workspace) Stat(ctx context.Context, p string) (tool.FileInfo, error)

Stat returns metadata for a virtual regular file.

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.