From 0d8e3688e8102977bf74f676dcd6e607d49ac468 Mon Sep 17 00:00:00 2001 From: lyon Date: Mon, 22 Jun 2026 14:17:12 +0800 Subject: [PATCH] fix: add workbench cache freshness diagnostics --- internal/workbenchruntime/cache_test.go | 32 +++++++ internal/workbenchruntime/service.go | 115 ++++++++++++++++++++++-- 2 files changed, 140 insertions(+), 7 deletions(-) diff --git a/internal/workbenchruntime/cache_test.go b/internal/workbenchruntime/cache_test.go index be33588a..3b6be802 100644 --- a/internal/workbenchruntime/cache_test.go +++ b/internal/workbenchruntime/cache_test.go @@ -158,6 +158,38 @@ func TestSessionsSummaryCacheKeyActorIsolation(t *testing.T) { } } +func TestSessionsSummaryRevisionMetadataExplainsFreshness(t *testing.T) { + payload := map[string]any{"sessions": []any{ + map[string]any{"sessionId": "ses_1", "projectedSeq": int64(12), "updatedAt": "2026-06-22T06:00:00Z", "projection": map[string]any{"lastProjectedSeq": int64(30)}}, + map[string]any{"sessionId": "ses_2", "projectedSeq": int64(44), "updatedAt": "2026-06-22T06:01:00Z", "turnSummary": map[string]any{"eventCount": int64(41)}}, + }} + revision := sessionsSummaryRevisionForPayload(payload) + if revision.SessionCount != 2 || revision.ProjectionSeqMax != 44 || revision.SessionUpdatedAtMax != "2026-06-22T06:01:00Z" { + t.Fatalf("unexpected revision: %#v", revision) + } + if !strings.HasPrefix(revision.ProjectionRevision, "digest:") { + t.Fatalf("revision should use a bounded digest, got %q", revision.ProjectionRevision) + } + + meta := sessionsCacheMetadata("miss") + applySessionsCacheRevisionMetadata(meta, revision) + if meta["invalidationMode"] != "ttl_slo" || meta["revisionSource"] != "response_payload" { + t.Fatalf("unexpected freshness metadata: %#v", meta) + } + if meta["projectionRevision"] == nil || meta["projectionSeq"] != int64(44) || meta["sessionCount"] != 2 { + t.Fatalf("missing revision metadata: %#v", meta) + } +} + +func TestProjectionRevisionTokenUsesProjectedSeq(t *testing.T) { + if got := projectionRevisionToken(42); got != "seq:42" { + t.Fatalf("projection revision token=%q", got) + } + if got := projectionRevisionToken(0); got != "" { + t.Fatalf("zero projection revision token=%q", got) + } +} + func TestFactsCacheClassAndTTL(t *testing.T) { tracePage := factQuery{TraceID: "trc_1", Families: []string{"traceEvents"}, AfterProjectedSeq: 10, Limit: 51} if got := factsCacheClass(tracePage); got != terminalTracePageCacheClass { diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index bb35c509..d8880ab8 100644 --- a/internal/workbenchruntime/service.go +++ b/internal/workbenchruntime/service.go @@ -30,11 +30,15 @@ const ( ) 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"` + CachedAt string `json:"cachedAt"` + Payload map[string]any `json:"payload"` + PayloadBytes int `json:"payloadBytes"` + TTLMillis int64 `json:"ttlMs"` + ProjectionRevision string `json:"projectionRevision,omitempty"` + ProjectionSeqMax int64 `json:"projectionSeqMax,omitempty"` + SessionUpdatedAtMax string `json:"sessionUpdatedAtMax,omitempty"` + SessionCount int `json:"sessionCount,omitempty"` + ValuesRedacted bool `json:"valuesRedacted"` } type factsCacheEntry struct { @@ -56,6 +60,14 @@ type factsCachePlan struct { Status string } +type sessionsSummaryRevision struct { + ProjectionRevision string + ProjectionSeqMax int64 + SessionUpdatedAtMax string + SessionCount int + RevisionSource string +} + type Config struct { Addr string DatabaseURL string @@ -283,11 +295,13 @@ func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[s plan.Meta["dbQueryDurationMs"] = int64(0) plan.Meta["payloadBytes"] = entry.PayloadBytes plan.Meta["ttlMs"] = entry.TTLMillis + plan.Meta["freshnessSloMs"] = entry.TTLMillis attachFactsCacheMetadata(payload, plan.Meta) return payload, nil } setFactsCacheStatus(plan.Meta, "stale") plan.Meta["cacheAgeMs"] = ageMs + plan.Meta["freshnessSloMs"] = entry.TTLMillis } } @@ -308,6 +322,7 @@ func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[s ttl, unstable, ttlReason := factsCacheTTL(plan, s.config.Cache) plan.Meta["unstable"] = unstable plan.Meta["ttlMs"] = ttl.Milliseconds() + plan.Meta["freshnessSloMs"] = ttl.Milliseconds() plan.Meta["ttlReason"] = ttlReason storedPayload := factsCacheStorablePayload(payload) payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes) @@ -377,9 +392,12 @@ func (s *Server) factsCachePlan(ctx context.Context, query factQuery) factsCache plan.ProjectionHealth = state.ProjectionHealth plan.Status = state.Status meta["projectionSeq"] = state.ProjectedSeq + meta["projectionRevision"] = projectionRevisionToken(state.ProjectedSeq) meta["projectionStatus"] = state.ProjectionStatus meta["projectionHealth"] = state.ProjectionHealth meta["traceStatus"] = state.Status + meta["revisionSource"] = "workbench_projection_checkpoints" + meta["invalidationMode"] = "projection_revision_key" if state.ProjectedSeq <= 0 { setFactsCacheStatus(meta, "skipped") meta["skipReason"] = "missing_projection_revision" @@ -513,6 +531,10 @@ func factsCacheMetadata(class string, status string) map[string]any { "dbQueryDurationMs": nil, "payloadBytes": nil, "ttlMs": nil, + "freshnessSloMs": nil, + "projectionRevision": nil, + "invalidationMode": nil, + "revisionSource": nil, "valuesRedacted": true, "secretMaterialStored": false, } @@ -707,11 +729,15 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[ cacheMeta["dbQueryDurationMs"] = int64(0) cacheMeta["payloadBytes"] = entry.PayloadBytes cacheMeta["ttlMs"] = entry.TTLMillis + cacheMeta["freshnessSloMs"] = entry.TTLMillis + applySessionsCacheRevisionMetadata(cacheMeta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, SessionUpdatedAtMax: entry.SessionUpdatedAtMax, SessionCount: entry.SessionCount, RevisionSource: "cached_payload"}) attachSessionsCacheMetadata(payload, cacheMeta) return payload, nil } setSessionsCacheStatus(cacheMeta, "stale") cacheMeta["cacheAgeMs"] = ageMs + cacheMeta["freshnessSloMs"] = entry.TTLMillis + applySessionsCacheRevisionMetadata(cacheMeta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, SessionUpdatedAtMax: entry.SessionUpdatedAtMax, SessionCount: entry.SessionCount, RevisionSource: "cached_payload"}) } } @@ -731,7 +757,10 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[ ttl, unstable, ttlReason := sessionsSummaryCacheTTL(payload, s.config.Cache) cacheMeta["unstable"] = unstable cacheMeta["ttlMs"] = ttl.Milliseconds() + cacheMeta["freshnessSloMs"] = ttl.Milliseconds() cacheMeta["ttlReason"] = ttlReason + revision := sessionsSummaryRevisionForPayload(payload) + applySessionsCacheRevisionMetadata(cacheMeta, revision) storedPayload := sessionsCacheStorablePayload(payload) payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes) if payloadErr != nil { @@ -747,7 +776,7 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[ 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} + entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ProjectionRevision: revision.ProjectionRevision, ProjectionSeqMax: revision.ProjectionSeqMax, SessionUpdatedAtMax: revision.SessionUpdatedAtMax, SessionCount: revision.SessionCount, ValuesRedacted: true} if err := s.cache.Set(ctx, cacheKey, entry, ttl); err != nil { setSessionsCacheStatus(cacheMeta, "set_failed") cacheMeta["errorKind"] = classifyCacheError(err) @@ -863,6 +892,13 @@ func sessionsCacheMetadata(status string) map[string]any { "dbQueryDurationMs": nil, "payloadBytes": nil, "ttlMs": nil, + "freshnessSloMs": nil, + "projectionRevision": nil, + "projectionSeq": nil, + "sessionUpdatedAtMax": nil, + "sessionCount": nil, + "revisionSource": nil, + "invalidationMode": "ttl_slo", "valuesRedacted": true, "secretMaterialStored": false, } @@ -901,6 +937,68 @@ func sessionsSummaryCacheTTL(payload map[string]any, config cacheConfig) (time.D return config.SessionsSummaryTTL, false, "stable_page" } +func sessionsSummaryRevisionForPayload(payload map[string]any) sessionsSummaryRevision { + rows := sessionsPayloadRows(payload["sessions"]) + revision := sessionsSummaryRevision{RevisionSource: "response_payload"} + maxUpdatedAt := "" + var maxSeq int64 + for _, row := range rows { + item := objectMap(row) + if len(item) == 0 { + continue + } + revision.SessionCount++ + if updatedAt := text(item["updatedAt"]); updatedAt > maxUpdatedAt { + maxUpdatedAt = updatedAt + } + maxSeq = maxInt64(maxSeq, int64FromAny(item["projectedSeq"])) + projection := objectMap(item["projection"]) + maxSeq = maxInt64(maxSeq, int64FromAny(projection["lastProjectedSeq"])) + turnSummary := objectMap(item["turnSummary"]) + maxSeq = maxInt64(maxSeq, int64FromAny(turnSummary["eventCount"])) + } + revision.ProjectionSeqMax = maxSeq + revision.SessionUpdatedAtMax = maxUpdatedAt + raw := fmt.Sprintf("contract=workbench-sessions-v1|count=%d|maxProjectionSeq=%d|maxUpdatedAt=%s", revision.SessionCount, revision.ProjectionSeqMax, revision.SessionUpdatedAtMax) + revision.ProjectionRevision = "digest:" + cacheHash(raw) + return revision +} + +func applySessionsCacheRevisionMetadata(meta map[string]any, revision sessionsSummaryRevision) { + if meta == nil { + return + } + if revision.ProjectionRevision != "" { + meta["projectionRevision"] = revision.ProjectionRevision + } + meta["projectionSeq"] = revision.ProjectionSeqMax + meta["sessionUpdatedAtMax"] = nilIfEmpty(revision.SessionUpdatedAtMax) + meta["sessionCount"] = revision.SessionCount + meta["revisionSource"] = first(revision.RevisionSource, "response_payload") + meta["invalidationMode"] = "ttl_slo" +} + +func projectionRevisionToken(projectedSeq int64) string { + if projectedSeq <= 0 { + return "" + } + return "seq:" + strconv.FormatInt(projectedSeq, 10) +} + +func maxInt64(a int64, b int64) int64 { + if b > a { + return b + } + return a +} + +func nilIfEmpty(value string) any { + if strings.TrimSpace(value) == "" { + return nil + } + return value +} + func sessionsSummaryPayloadUnstable(payload map[string]any) bool { for _, row := range sessionsPayloadRows(payload["sessions"]) { item := objectMap(row) @@ -1221,8 +1319,9 @@ func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string, defer rows.Close() for rows.Next() { var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID sql.NullString + var projectedSeq sql.NullInt64 var createdAt, updatedAt sql.NullString - if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &createdAt, &updatedAt); err != nil { + if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &createdAt, &updatedAt); err != nil { return err } if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" { @@ -1237,6 +1336,7 @@ func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string, "threadId": nullableString(threadID), "status": first(nullableStringValue(status), "unknown"), "lastTraceId": nullableString(lastTraceID), + "projectedSeq": nullableInt(projectedSeq), "createdAt": nullableTimeText(createdAt), "updatedAt": first(nullableTimeText(updatedAt), nullableTimeText(createdAt)), "valuesPrinted": false, @@ -1601,6 +1701,7 @@ func sessionSummarySelectClause() string { "thread_id", "status", "last_trace_id", + "projected_seq", "created_at", "updated_at", }, ", ")