mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1895 from pikasTech/fix/1870-p6-cache-consistency

Browse Source 
fix: 补齐 Workbench Redis cache freshness 诊断
This commit is contained in:
Lyon
2026-06-22 14:19:00 +08:00
committed by GitHub
parent 0347aa6117 0d8e3688e8
commit 19b6db4254
2 changed files with 140 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workbenchruntime/cache_test.go
+32
View File
@@ -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 {
internal/workbenchruntime/service.go
+108 -7
View File
@@ -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",
}, ", ")