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).
No client-side mutex is needed: Redis serializes commands single-threaded and HSET/HGET/RPUSH 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), exercised offline against an in-process miniredis so `task test` needs no live broker.
DURABILITY CAVEAT: Append/Save call RPUSH/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
- Variables
- type Config
- type Store
- func (st *Store) AcquireSessionMigrationJob(ctx context.Context, id string) (context.Context, func() error, error)
- func (st *Store) Append(ctx context.Context, id session.SessionID, ev session.Event) error
- func (*Store) CheckSessionMigrationJobOwnership(ctx context.Context) error
- func (st *Store) Close() error
- func (st *Store) Create(ctx context.Context, s *session.Session) error
- func (st *Store) Delete(ctx context.Context, id session.SessionID) error
- func (st *Store) DeleteSessionIfUnchanged(ctx context.Context, expected port.SessionDiscoveryMeta) (bool, error)
- func (st *Store) FinalizeSessionMigrationCoverage(ctx context.Context, generation string, expectedFamilies int64) (bool, error)
- func (st *Store) InspectSessionMigration(ctx context.Context) (port.SessionMigrationInspection, error)
- func (st *Store) List(ctx context.Context) ([]port.StoredSession, error)
- func (st *Store) Load(ctx context.Context, id session.SessionID) (*session.Session, error)
- func (st *Store) LoadSessionMigrationJob(ctx context.Context, id string) (port.SessionMigrationJob, error)
- func (st *Store) MigrateSessionFamily(ctx context.Context, expected port.SessionMigrationFamily) (string, error)
- func (st *Store) PageSessionMetadata(ctx context.Context, request port.SessionMetadataPageRequest) (port.SessionMetadataPage, error)
- func (st *Store) Ping(ctx context.Context) error
- func (st *Store) Read(ctx context.Context, id session.SessionID) iter.Seq2[session.Event, error]
- func (st *Store) Save(ctx context.Context, s *session.Session) error
- func (*Store) SaveSessionMigrationJob(ctx context.Context, job port.SessionMigrationJob) error
- func (st *Store) ScheduleStore() port.ScheduleStore
- func (st *Store) ToolCall(id session.SessionID, call session.ToolCall, result session.ToolResult, ...)
Constants ¶
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).
Variables ¶
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.
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 ¶
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 ¶
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 ¶
Append durably records ev under id as a format-tagged JSON record on the per-session event list (RPUSH). 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. RPUSH preserves append order, so Read returns events in the exact order Append received them.
func (*Store) CheckSessionMigrationJobOwnership ¶
CheckSessionMigrationJobOwnership fails closed once this acquisition is lost.
func (*Store) Close ¶
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 ¶
Create atomically publishes a snapshot and its derivative metadata only when no authoritative Redis session key exists for s.ID.
func (*Store) Delete ¶
Delete removes the session snapshot, derivative metadata, event log, and tool-call sidecar. It is idempotent and completes in one atomic script.
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 ¶
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 ¶
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) 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 ¶
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 ¶
Read yields the session's recorded events in APPEND order (RPUSH order = LRANGE 0 -1 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).
func (*Store) Save ¶
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 ¶
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.
Source Files