redisstore

package
v0.0.25 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: Apache-2.0 Imports: 32 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. A LIST written before that change stays readable and is migrated in place, atomically, by the next append — see cursoreventlog.go.

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

This section is empty.

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
	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) AcquireSessionMigrationJob 

func (st *Store) AcquireSessionMigrationJob(ctx context.Context, id string) (context.Context, func() error, error)

AcquireSessionMigrationJob binds one fenced acquisition to the returned context. Renewal and every checkpoint/release use that exact acquisition.

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 added in v0.0.22 

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 added in v0.0.22 

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) CheckSessionMigrationJobOwnership 

func (*Store) CheckSessionMigrationJobOwnership(ctx context.Context) error

CheckSessionMigrationJobOwnership fails closed once this acquisition is lost.

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. It never force-closes a client still in use; that client closes exactly once when its final lease is released.

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 added in v0.0.25 

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 added in v0.0.25 

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) FinalizeSessionMigrationCoverage 

func (st *Store) FinalizeSessionMigrationCoverage(ctx context.Context, generation string, expectedFamilies int64) (bool, error)

FinalizeSessionMigrationCoverage proves the exact global and per-owner index contents with bounded client-side scans before publishing readiness with a constant-work Redis CAS. The CAS rechecks the generation, so a concurrent Save or Delete cannot invalidate the proof before publication.

func (*Store) InspectSessionMigration 

func (st *Store) InspectSessionMigration(ctx context.Context) (port.SessionMigrationInspection, error)

InspectSessionMigration explicitly inventories legacy Redis snapshots for the authenticated maintenance workflow. Inventory pages never call this path.

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 session keys (MATCH mecatl:session:*), 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) LoadSessionMigrationJob 

func (st *Store) LoadSessionMigrationJob(ctx context.Context, id string) (port.SessionMigrationJob, error)

LoadSessionMigrationJob reloads one validated durable adoption checkpoint.

func (*Store) MigrateSessionFamily 

func (st *Store) MigrateSessionFamily(ctx context.Context, expected port.SessionMigrationFamily) (string, error)

MigrateSessionFamily derives and conditionally installs one metadata row. The Lua compare-and-publish prevents a concurrent Save or Delete from being lost.

func (*Store) OpenWorkspace added in v0.0.25 

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 a STREAM (XRANGE) or, for a log written before the Stream migration and not appended to since, a LIST (LRANGE) — chosen by the key's actual type rather than by a stored flag, so no migration bookkeeping can disagree with the keyspace. Reading does NOT migrate: a read must not mutate, and the session's next append migrates it anyway.

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 added in v0.0.22 

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 added in v0.0.25 

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 added in v0.0.22 

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) SaveSessionMigrationJob 

func (*Store) SaveSessionMigrationJob(ctx context.Context, job port.SessionMigrationJob) error

SaveSessionMigrationJob durably checkpoints a caller-bound adoption job.

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) 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 added in v0.0.25 

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

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

func (*Workspace) AuthorityResourcePath added in v0.0.25 

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

AuthorityResourcePath projects a confined virtual path for policy evaluation.

func (*Workspace) CreateFile added in v0.0.25 

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 added in v0.0.25 

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

Glob returns sorted paths matching a doublestar pattern.

func (*Workspace) Grep added in v0.0.25 

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 added in v0.0.25 

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

func (*Workspace) ReadVersion added in v0.0.25 

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) ReplaceFile added in v0.0.25 

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 added in v0.0.25 

func (*Workspace) Root() string

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

func (*Workspace) Stat added in v0.0.25 

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.