Bounded analytics admission gate
apps/backend/internal/analytics/repository/sqlite/admission.go ↗The gate limits concurrent analytics work to two slots and bounds total wait plus query time to ten seconds.
Admission 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())
}
}
Scoped aggregate queries
apps/backend/internal/analytics/repository/sqlite/stats.go ↗The query selects eligible tasks first, then aggregates sessions, turns, and messages for those IDs only.
Task stats CTE
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")+`
), session_stats AS (
SELECT task_id,
COUNT(*) AS session_count,
MAX(completed_at) AS last_completed_at
FROM scoped_sessions
GROUP BY task_id
), turn_stats AS (
SELECT s.task_id,
COUNT(turn.id) AS turn_count,
COALESCE(SUM(CASE WHEN turn.completed_at IS NOT NULL THEN %s ELSE 0 END), 0) AS active_duration_ms,
%s AS elapsed_span_ms
FROM scoped_sessions s
LEFT JOIN task_session_turns turn ON turn.task_session_id = s.id
GROUP BY s.task_id
)`, dur, elapsedDur)
Operation wrapper
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
startArg := rangeStartArg(start)
// ... scoped query follows
}
Busy error to HTTP 503
apps/backend/internal/analytics/handlers/stats_handlers.go ↗The handler turns a busy budget error into a retryable 503 with Retry-After, and keeps other errors as 500.
Fail handler
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})
}
Per-section stats recovery
apps/web/app/stats/stats-data.tsx ↗Each stats section loads independently, keeps data on retry, and retries transient failures with backoff.
Error classification
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 loop
const RETRY_DELAYS_MS = [2_000, 5_000] as const;
// inside startRequest catch
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(() => {
delete scope.timers[key];
startRequest(key);
}, delay);
Workspace context with generation guard
apps/web/src/spa-routes.tsx ↗Bootstrap records per-collection outcomes and checks generation before writing, so a late response cannot clear live data.
Bootstrap outcomes
const generation = store.getState().workspaceContextGeneration;
for (const collection of ["workflows", "repositories", "steps"] as const) {
const requestId = generateUUID();
requestIds.set(collection, { workspaceId, generation, requestId });
store.getState().setWorkspaceContextRead(collection, workspaceId, generation, "pending", undefined, requestId);
}
const [workflowsResult, repositoriesResult, stepsResult] = await Promise.all([
settleRouteRead(listWorkflows(workspaceId, { cache: "no-store" })),
settleRouteRead(listRepositories(workspaceId, undefined, { cache: "no-store" })),
settleRouteRead(listWorkspaceWorkflowSteps(workspaceId)),
]);
if (cancelled || !isCurrentWorkspaceContext(store.getState(), workspaceId, generation)) return;
for (const [collection, result] of readResults) {
if (result.ok) currentState.setWorkspaceContextRead(collection, workspaceId, generation, "success", undefined, requestId);
else currentState.setWorkspaceContextRead(collection, workspaceId, generation, classifyWorkspaceContextReadError(result.error), retryAfterMilliseconds(result.error), requestId);
}
if (workflowsResult.ok) store.getState().hydrate({ workflows: { items: workflowItems, activeId: activeWorkflowId } });
if (repositoriesResult.ok) store.getState().setRepositories(workspaceId, repositoriesResult.value.repositories);
if (stepsResult.ok) setSteps(stepsResult.value.steps);
Portable timestamp helpers
apps/backend/internal/db/dialect/time.go ↗The helper normalizes naive UTC timestamps correctly for Postgres, so time comparisons do not shift with session timezone.
Naive UTC helper
func NaiveUTCTimestampOf(driver, expr string) string {
if IsPostgres(driver) {
return fmt.Sprintf("(%s AT TIME ZONE 'UTC')", expr)
}
return fmt.Sprintf("datetime(%s)", expr)
}
func DateTimeOf(driver, expr string) string {
if IsPostgres(driver) {
return fmt.Sprintf("(%s)::timestamptz", expr)
}
return fmt.Sprintf("datetime(%s)", expr)
}
