From cf72e7038e279e3f03244b1fd8e1623e26b68b1a Mon Sep 17 00:00:00 2001 From: lyon Date: Mon, 22 Jun 2026 13:27:27 +0800 Subject: [PATCH] feat: cache workbench sessions summaries in redis --- internal/workbenchruntime/cache_test.go | 65 ++++++ internal/workbenchruntime/service.go | 260 +++++++++++++++++++++++- 2 files changed, 319 insertions(+), 6 deletions(-) diff --git a/internal/workbenchruntime/cache_test.go b/internal/workbenchruntime/cache_test.go index 0770409f..62c6597a 100644 --- a/internal/workbenchruntime/cache_test.go +++ b/internal/workbenchruntime/cache_test.go @@ -5,6 +5,7 @@ import ( "errors" "strings" "testing" + "time" "github.com/redis/go-redis/v9" ) @@ -93,6 +94,70 @@ func TestDisabledCacheDoesNotFailReadPath(t *testing.T) { } } +func TestSessionsSummaryCacheTTLUsesShortTTLForUnstablePages(t *testing.T) { + config := cacheConfig{SessionsSummaryTTL: 30 * time.Second, RunningPageTTL: 5 * time.Second} + stable := map[string]any{"sessions": []any{map[string]any{"terminal": true, "running": false, "projectionStatus": "caught-up"}}} + ttl, unstable, reason := sessionsSummaryCacheTTL(stable, config) + if ttl != 30*time.Second || unstable || reason != "stable_page" { + t.Fatalf("stable ttl=%s unstable=%v reason=%s", ttl, unstable, reason) + } + + running := map[string]any{"sessions": []any{map[string]any{"terminal": false, "running": true, "projectionStatus": "projecting"}}} + ttl, unstable, reason = sessionsSummaryCacheTTL(running, config) + if ttl != 5*time.Second || !unstable || reason != "running_or_projecting_page" { + t.Fatalf("running ttl=%s unstable=%v reason=%s", ttl, unstable, reason) + } + + config.RunningPageTTL = 0 + ttl, unstable, reason = sessionsSummaryCacheTTL(running, config) + if ttl != 0 || !unstable || reason != "running_page_ttl_disabled" { + t.Fatalf("disabled running ttl=%s unstable=%v reason=%s", ttl, unstable, reason) + } +} + +func TestSessionsCacheStorablePayloadRemovesRuntimeOnlySecretKeys(t *testing.T) { + payload := map[string]any{ + "ok": true, + "contractVersion": "workbench-sessions-v1", + "secretMaterialStored": false, + "persistence": map[string]any{ + "cache": map[string]any{"secretMaterialStored": false, "valuesRedacted": true}, + }, + "sessions": []any{map[string]any{"sessionId": "ses_1", "terminal": true, "projectionStatus": "caught-up", "valuesRedacted": true}}, + } + stored := sessionsCacheStorablePayload(payload) + if _, ok := stored["secretMaterialStored"]; ok { + t.Fatalf("storable payload must not keep secretMaterialStored: %#v", stored) + } + if _, ok := stored["persistence"]; ok { + t.Fatalf("storable payload must not keep runtime persistence summary: %#v", stored) + } + entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: stored, PayloadBytes: 128, TTLMillis: 30000, ValuesRedacted: true} + if _, err := cachePayloadBytes(entry, 4096); err != nil { + t.Fatalf("storable sessions payload rejected: %v", err) + } +} + +func TestSessionsSummaryCacheKeyActorIsolation(t *testing.T) { + config := cacheConfig{Enabled: true, RedisURL: "redis://localhost:6379/0", MaxKeyBytes: 512} + server := &Server{config: Config{Cache: config}, cache: newDerivedCache(config)} + query := sessionListQuery{Limit: 20, Actor: sessionActor{ID: "actor-a", Role: "user"}} + keyA, meta := server.sessionsSummaryCacheKey(query, 20, 0) + if keyA == "" || meta["status"] != "miss" || meta["enabled"] != true { + t.Fatalf("unexpected key/meta: key=%q meta=%#v", keyA, meta) + } + query.Actor.ID = "actor-b" + keyB, _ := server.sessionsSummaryCacheKey(query, 20, 0) + if keyA == keyB { + t.Fatalf("expected actor-isolated session cache keys, got %q", keyA) + } + query.Actor.ID = "" + keyMissingActor, meta := server.sessionsSummaryCacheKey(query, 20, 0) + if keyMissingActor != "" || meta["status"] != "skipped" || meta["skipReason"] != "missing_actor" { + t.Fatalf("expected missing actor skip, key=%q meta=%#v", keyMissingActor, meta) + } +} + func cacheErrKind(err error) string { var cacheErr cacheError if errors.As(err, &cacheErr) { diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index 30cb49e9..24ce1ce5 100644 --- a/internal/workbenchruntime/service.go +++ b/internal/workbenchruntime/service.go @@ -24,8 +24,17 @@ const ( sessionSummaryQueryMaxAttempts = 2 sessionSummaryQueryAttemptTimeout = 2 * time.Second sessionSummaryQueryRetryBackoff = 150 * time.Millisecond + sessionsSummaryCacheClass = "sessions.summary" ) +type sessionsCacheEntry struct { + CachedAt string `json:"cachedAt"` + Payload map[string]any `json:"payload"` + PayloadBytes int `json:"payloadBytes"` + TTLMillis int64 `json:"ttlMs"` + ValuesRedacted bool `json:"valuesRedacted"` +} + type Config struct { Addr string DatabaseURL string @@ -315,14 +324,80 @@ func (s *Server) handleSessions(w http.ResponseWriter, r *http.Request) { } func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[string]any, error) { - limit := boundedLimit(query.Limit, 20, 100) - offset := query.Offset - if offset <= 0 { - offset = cursorOffset(query.Cursor) + limit, offset := sessionListPage(query) + cacheKey, cacheMeta := s.sessionsSummaryCacheKey(query, limit, offset) + if cacheKey != "" { + var entry sessionsCacheEntry + hit, err := s.cache.Get(ctx, cacheKey, &entry) + if err != nil { + setSessionsCacheStatus(cacheMeta, "unavailable") + cacheMeta["errorKind"] = classifyCacheError(err) + } else if hit && entry.Payload != nil { + ageMs, stale := sessionsCacheEntryAgeMs(entry, time.Now().UTC()) + if !stale { + payload := sessionsCacheResponsePayload(entry.Payload, s.persistenceSummary()) + setSessionsCacheStatus(cacheMeta, "hit") + cacheMeta["cacheAgeMs"] = ageMs + cacheMeta["dbQueryAvoided"] = true + cacheMeta["dbQueryDurationMs"] = int64(0) + cacheMeta["payloadBytes"] = entry.PayloadBytes + cacheMeta["ttlMs"] = entry.TTLMillis + attachSessionsCacheMetadata(payload, cacheMeta) + return payload, nil + } + setSessionsCacheStatus(cacheMeta, "stale") + cacheMeta["cacheAgeMs"] = ageMs + } } - if offset < 0 { - offset = 0 + + dbStart := time.Now() + payload, err := s.listSessionsFromDB(ctx, query, limit, offset) + dbDuration := time.Since(dbStart) + if err != nil { + return nil, err } + cacheMeta["dbQueryAvoided"] = false + cacheMeta["dbQueryDurationMs"] = dbDuration.Milliseconds() + if cacheKey == "" { + attachSessionsCacheMetadata(payload, cacheMeta) + return payload, nil + } + + ttl, unstable, ttlReason := sessionsSummaryCacheTTL(payload, s.config.Cache) + cacheMeta["unstable"] = unstable + cacheMeta["ttlMs"] = ttl.Milliseconds() + cacheMeta["ttlReason"] = ttlReason + storedPayload := sessionsCacheStorablePayload(payload) + payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes) + if payloadErr != nil { + setSessionsCacheStatus(cacheMeta, "skipped") + cacheMeta["skipReason"] = classifyCacheError(payloadErr) + attachSessionsCacheMetadata(payload, cacheMeta) + return payload, nil + } + cacheMeta["payloadBytes"] = len(payloadBytes) + if ttl <= 0 { + setSessionsCacheStatus(cacheMeta, "skipped") + cacheMeta["skipReason"] = ttlReason + attachSessionsCacheMetadata(payload, cacheMeta) + return payload, nil + } + entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ValuesRedacted: true} + if err := s.cache.Set(ctx, cacheKey, entry, ttl); err != nil { + setSessionsCacheStatus(cacheMeta, "set_failed") + cacheMeta["errorKind"] = classifyCacheError(err) + attachSessionsCacheMetadata(payload, cacheMeta) + return payload, nil + } + if cacheMeta["status"] != "stale" { + setSessionsCacheStatus(cacheMeta, "miss") + } + cacheMeta["stored"] = true + attachSessionsCacheMetadata(payload, cacheMeta) + return payload, nil +} + +func (s *Server) listSessionsFromDB(ctx context.Context, query sessionListQuery, limit int, offset int) (map[string]any, error) { ownerUserID := strings.TrimSpace(query.Actor.ID) pageFacts, err := s.queryFacts(ctx, factQuery{OwnerUserID: ownerUserID, Limit: limit + offset + 1, Families: []string{"sessions"}, SessionsOrder: "updated_desc", SessionProjection: "summary"}) if err != nil { @@ -370,6 +445,179 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[ return map[string]any{"ok": true, "status": "succeeded", "contractVersion": "workbench-sessions-v1", "sessions": summaries, "count": len(summaries), "cursor": cursorOrNil(offset), "hasMore": hasMore, "nextCursor": nextCursor(offset, limit, hasMore), "persistence": s.persistenceSummary(), "servedBy": serviceID, "valuesRedacted": true, "secretMaterialStored": false}, nil } +func sessionListPage(query sessionListQuery) (int, int) { + limit := boundedLimit(query.Limit, 20, 100) + offset := query.Offset + if offset <= 0 { + offset = cursorOffset(query.Cursor) + } + if offset < 0 { + offset = 0 + } + return limit, offset +} + +func (s *Server) sessionsSummaryCacheKey(query sessionListQuery, limit int, offset int) (string, map[string]any) { + meta := sessionsCacheMetadata("disabled") + if s == nil || s.cache == nil || !s.config.Cache.Enabled { + return "", meta + } + meta["enabled"] = true + actorID := strings.TrimSpace(query.Actor.ID) + if actorID == "" { + setSessionsCacheStatus(meta, "skipped") + meta["skipReason"] = "missing_actor" + return "", meta + } + includeID := strings.TrimSpace(query.IncludeSessionID) + actorRole := strings.TrimSpace(query.Actor.Role) + cursor := fmt.Sprintf("idx:%d|limit:%d|include:%s|role:%s", offset, limit, includeID, actorRole) + authority := fmt.Sprintf("contract=workbench-sessions-v1|limit=%d|offset=%d|include=%t|role=%s", limit, offset, includeID != "", actorRole) + key, err := s.cache.BuildKey(cacheKeyParts{Class: sessionsSummaryCacheClass, ActorID: actorID, SessionID: includeID, Cursor: cursor, ProjectionRevision: "workbench-sessions-v1", Authority: authority}) + if err != nil { + setSessionsCacheStatus(meta, "skipped") + meta["skipReason"] = classifyCacheError(err) + return "", meta + } + setSessionsCacheStatus(meta, "miss") + return key, meta +} + +func sessionsCacheMetadata(status string) map[string]any { + meta := map[string]any{ + "class": sessionsSummaryCacheClass, + "schemaVersion": cacheKeySchemaVersion, + "enabled": false, + "status": status, + "hit": false, + "miss": false, + "stale": false, + "unavailable": false, + "dbQueryAvoided": false, + "cacheAgeMs": nil, + "dbQueryDurationMs": nil, + "payloadBytes": nil, + "ttlMs": nil, + "valuesRedacted": true, + "secretMaterialStored": false, + } + setSessionsCacheStatus(meta, status) + return meta +} + +func setSessionsCacheStatus(meta map[string]any, status string) { + if meta == nil { + return + } + meta["status"] = status + meta["hit"] = status == "hit" + meta["miss"] = status == "miss" + meta["stale"] = status == "stale" + meta["unavailable"] = status == "unavailable" || status == "set_failed" +} + +func attachSessionsCacheMetadata(payload map[string]any, meta map[string]any) { + if payload == nil || meta == nil { + return + } + payload["cache"] = meta +} + +func sessionsSummaryCacheTTL(payload map[string]any, config cacheConfig) (time.Duration, bool, string) { + if sessionsSummaryPayloadUnstable(payload) { + if config.RunningPageTTL <= 0 { + return 0, true, "running_page_ttl_disabled" + } + return config.RunningPageTTL, true, "running_or_projecting_page" + } + if config.SessionsSummaryTTL <= 0 { + return 0, false, "sessions_summary_ttl_disabled" + } + return config.SessionsSummaryTTL, false, "stable_page" +} + +func sessionsSummaryPayloadUnstable(payload map[string]any) bool { + for _, row := range sessionsPayloadRows(payload["sessions"]) { + item := objectMap(row) + if len(item) == 0 { + continue + } + if boolValue(item["running"]) || !boolValue(item["terminal"]) { + return true + } + if status := normalizeProjectionStatus(text(item["projectionStatus"])); status != "" && status != "caught-up" { + return true + } + turnSummary := objectMap(item["turnSummary"]) + if len(turnSummary) > 0 && (boolValue(turnSummary["running"]) || !boolValue(turnSummary["terminal"])) { + return true + } + } + return false +} + +func sessionsPayloadRows(value any) []any { + switch rows := value.(type) { + case []any: + return rows + case []map[string]any: + out := make([]any, 0, len(rows)) + for _, row := range rows { + out = append(out, row) + } + return out + default: + return []any{} + } +} + +func sessionsCacheStorablePayload(payload map[string]any) map[string]any { + clone := cloneJSONMap(payload) + delete(clone, "cache") + delete(clone, "persistence") + delete(clone, "secretMaterialStored") + return clone +} + +func sessionsCacheResponsePayload(payload map[string]any, persistence map[string]any) map[string]any { + clone := cloneJSONMap(payload) + clone["persistence"] = persistence + clone["secretMaterialStored"] = false + clone["valuesRedacted"] = true + clone["servedBy"] = serviceID + return clone +} + +func sessionsCacheEntryAgeMs(entry sessionsCacheEntry, now time.Time) (int64, bool) { + cachedAt, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(entry.CachedAt)) + if err != nil { + return 0, true + } + age := now.Sub(cachedAt) + if age < 0 { + age = 0 + } + if entry.TTLMillis > 0 && age.Milliseconds() > entry.TTLMillis { + return age.Milliseconds(), true + } + return age.Milliseconds(), false +} + +func cloneJSONMap(value map[string]any) map[string]any { + if value == nil { + return map[string]any{} + } + payload, err := json.Marshal(value) + if err != nil { + return map[string]any{} + } + var clone map[string]any + if err := json.Unmarshal(payload, &clone); err != nil || clone == nil { + return map[string]any{} + } + return clone +} + func (s *Server) queryIncludedSession(ctx context.Context, routeID string) (map[string][]any, error) { if strings.HasPrefix(routeID, "ses_") { return s.queryFacts(ctx, factQuery{SessionID: routeID, Limit: 1, Families: []string{"sessions"}, SessionProjection: "summary"})