Analytics busy sentinel
apps/backend/internal/analytics/errors.go ↗Defines the stable error identity that handlers and plugins use to detect a bounded-budget exhaustion.
Sentinel
var ErrAnalyticsBusy = errors.New("analytics busy")
type analyticsBusyError struct {
cause error
}
func (e *analyticsBusyError) Error() string {
if e.cause == nil {
return ErrAnalyticsBusy.Error()
}
return fmt.Sprintf("%s: %v", ErrAnalyticsBusy, e.cause)
}
func (e *analyticsBusyError) Is(target error) bool { return target == ErrAnalyticsBusy }
func NewAnalyticsBusyError(cause error) error {
return &analyticsBusyError{cause: cause}
}
func IsAnalyticsBusy(err error) bool { return errors.Is(err, ErrAnalyticsBusy) }
Admission gate with bounded budget
apps/backend/internal/analytics/repository/sqlite/admission.go ↗Limits concurrent analytics work to two slots and bounds queue plus query time to 10 seconds.
Gate
const (
analyticsAdmissionLimit = 2
analyticsOperationLimit = 10 * time.Second
)
type analyticsAdmission struct {
once sync.Once
gate chan struct{}
}
func (r *Repository) beginAnalyticsOperation(parent context.Context) (context.Context, func(), error) {
r.ensureAnalyticsAdmission()
operationCtx, cancel := context.WithTimeout(parent, analyticsOperationLimit)
select {
case r.admission.gate <- struct{}{}:
var releaseOnce sync.Once
release := func() {
releaseOnce.Do(func() {
cancel()
<-r.admission.gate
})
}
return operationCtx, release, nil
case <-operationCtx.Done():
cancel()
if parent.Err() != nil {
return nil, nil, parent.Err()
}
return nil, nil, analytics.NewAnalyticsBusyError(operationCtx.Err())
}
}
Normalize
func normalizeAnalyticsOperationError(parent context.Context, operation context.Context, err error) error {
if err == nil || parent.Err() != nil {
return err
}
if operation.Err() == context.DeadlineExceeded && errors.Is(err, context.DeadlineExceeded) {
return analytics.NewAnalyticsBusyError(err)
}
return err
}
Scoped CTE aggregates
apps/backend/internal/analytics/repository/sqlite/stats.go ↗Scopes to eligible tasks first, then aggregates sessions, turns, and messages in separate CTEs to avoid fan-out.
GetTaskStats CTEs
func (r *Repository) GetTaskStats(ctx context.Context, workspaceID string, start *time.Time, limit int) (results []*models.TaskStats, err error) {
parentCtx := ctx
operationCtx, release, err := r.beginAnalyticsOperation(parentCtx)
if err != nil {
return nil, err
}
defer func() {
err = normalizeAnalyticsOperationError(parentCtx, operationCtx, err)
release()
}()
ctx = operationCtx
query := fmt.Sprintf(`
WITH eligible_tasks AS (
SELECT t.id, t.title, t.workspace_id, t.workflow_id, t.state,
t.created_at, t.updated_at
FROM tasks t
WHERE t.workspace_id = ? AND t.is_ephemeral = 0 AND `+andNotAutomationOriginT+` AND `+rangeStartPredicate(drv, "t.created_at")+`
ORDER BY t.updated_at DESC
LIMIT ?
), scoped_sessions AS (
SELECT s.id, s.task_id, s.completed_at
FROM task_sessions s
JOIN eligible_tasks t ON t.id = s.task_id
WHERE `+rangeStartPredicate(drv, "s.started_at")+`
)` , dur, elapsedDur)
}
GetGlobalStats CTEs
query := fmt.Sprintf(`
WITH
task_agg AS (
SELECT COUNT(*) AS total_tasks, ...
FROM tasks t LEFT JOIN workflow_steps ws ON ws.id = t.workflow_step_id
WHERE t.workspace_id = ? AND t.is_ephemeral = 0 AND `+rangeStartPredicate(drv, "t.created_at")+`
),
session_agg AS ( SELECT COUNT(*) FROM task_sessions s JOIN tasks t ON t.id = s.task_id WHERE ... ),
turn_agg AS ( SELECT COUNT(*), COALESCE(SUM(CASE WHEN turn.completed_at IS NOT NULL THEN %s ELSE 0 END),0) FROM task_session_turns turn ... ),
clean_turn_agg AS ( SELECT AVG(dur_ms), AVG(msg_count) FROM ( SELECT %s AS dur_ms, (SELECT COUNT(*) FROM task_session_messages m WHERE m.turn_id = turn.id) AS msg_count ... ) clean WHERE dur_ms >= %d AND dur_ms < %d ),
message_agg AS ( SELECT COUNT(*), SUM(CASE WHEN msg.author_type = 'user' THEN 1 ELSE 0 END) ... )
SELECT ... FROM task_agg, session_agg, turn_agg, clean_turn_agg, message_agg
`, dur, dur, cleanTurnMinDurationMs, cleanTurnMaxDurationMs, cleanTurnMinMessages)
Handler maps busy to 503
apps/backend/internal/analytics/handlers/stats_handlers.go ↗Returns a retryable 503 with Retry-After 2 for busy analytics instead of a generic 500.
fail
func (h *StatsHandlers) fail(c *gin.Context, workspaceID, section string, err error) {
if repository.IsAnalyticsBusy(err) {
h.logger.Warn("analytics stats request exceeded its bounded read budget",
zap.String("workspace_id", workspaceID),
zap.String("section", section),
)
c.Header("Retry-After", "2")
c.JSON(http.StatusServiceUnavailable, gin.H{
"error_code": "analytics_busy",
"error": "statistics are temporarily busy",
})
return
}
h.logger.Error("failed to get "+section, zap.String("workspace_id", workspaceID), zap.Error(err))
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get " + section})
}
Workspace context recovery state
apps/web/lib/state/slices/kanban/kanban-slice.ts ↗Tracks per-collection pending, errors, retryAfter, and request IDs with generation guards so stale reads do not overwrite the current workspace.
State
export type WorkspaceContextReadState = {
workspaceId: string | null;
generation: number;
pending: Record<WorkspaceContextCollection, boolean>;
errors: Record<WorkspaceContextCollection, WorkspaceContextReadError | null>;
retryAfterMs: Record<WorkspaceContextCollection, number | null>;
requestIds: Record<WorkspaceContextCollection, string | null>;
snapshotPending: boolean;
snapshotError: WorkspaceContextReadError | null;
snapshotRetryAfterMs: number | null;
snapshotRequestId: string | null;
retryVersion: number;
retryCycle: number;
};
Guard
function applyWorkspaceContextRead(draft: Draft<KanbanSlice>, update: WorkspaceContextCollectionUpdate) {
const read = prepareWorkspaceContextRead(draft, update);
if (!read || !ownsWorkspaceContextRequest(read.requestIds[update.collection], update.result, update.requestId)) return;
read.requestIds[update.collection] = nextWorkspaceContextRequestId(update.result, update.requestId);
read.pending[update.collection] = update.result === "pending";
read.errors[update.collection] = workspaceContextError(update.result);
read.retryAfterMs[update.collection] = update.result === "transient" ? (update.retryAfterMs ?? null) : null;
}
Stats section retry with backoff
apps/web/app/stats/stats-data.tsx ↗Retries each stats section independently with Retry-After and foreground recovery, and preserves ready data on transient errors.
Classify
function classifySectionError(error: unknown, t: (key: string) => string) {
const status = errorStatus(error);
const code = errorCode(error);
const retryable = !isAbortError(error) && (code === "analytics_busy" || code === "persistence_unavailable" || status === 429 || status === 502 || status === 503 || status === 504 || (status === undefined && !(error instanceof SyntaxError)));
let message = t("stats:failedToLoadSection");
if (status === 401 || status === 403) message = t("stats:statsAccessDenied");
else if (retryable) message = t("stats:statsTemporarilyUnavailable");
return { message, retryable, ...(retryAfterMs(error) !== undefined ? { retryAfterMs: retryAfterMs(error) } : {}) };
}
Retry
const RETRY_DELAYS_MS = [2_000, 5_000] as const;
// on error:
if (!classified.retryable || document.visibilityState === "hidden") return;
const attempt = scope.attempts[key] ?? 0;
if (attempt >= RETRY_DELAYS_MS.length) return;
scope.attempts[key] = attempt + 1;
const delay = Math.max(RETRY_DELAYS_MS[attempt], classified.retryAfterMs ?? 0);
scope.timers[key] = setTimeout(() => startRequest(key), delay);
