mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
chore(logging): remove redis-progress-markers feature flag (#5287)
* chore(logging): remove redis-progress-markers feature flag Promote the Redis progress-marker write path to permanent behavior now that the flag is fully rolled out. Block markers always write to Redis (primary), keeping the durable jsonb_set UPDATE fallback when Redis is unavailable. Removes the flag registry entry, its REDIS_PROGRESS_MARKERS env fallback, and the per-session flag resolution in the logging session. * chore(logging): drop vestigial readProgressMarkers plumbing Follow-up to the redis-progress-markers flag removal. With the flag gone the completion fold always reads Redis markers, so the readProgressMarkers param (and its dead false branch) is removed end-to-end: - drop the param from completeWorkflowExecution and CompleteWorkflowExecutionParams - read progress markers unconditionally in the completion fold - delete the orphaned isFeatureEnabled mock and the flag-off / flag-throws marker tests (states the code can no longer produce)
This commit is contained in:
@@ -77,7 +77,6 @@ export const env = createEnv({
|
||||
TABLE_SNAPSHOT_CACHE: z.boolean().optional(), // Mount tables into sandboxes by reference via a version-keyed CSV snapshot in object storage instead of draining the whole table into web-process heap
|
||||
PII_REDACTION: z.boolean().optional(), // Redact PII from workflow logs via configurable Data Retention rules (Presidio at the logger persist choke point) and expose the Data Retention config UI
|
||||
TRIGGER_EU_REGION: z.boolean().optional(), // Route Trigger.dev runs to eu-central-1 instead of the default us-east-1 (fallback for the trigger-eu-region flag when AppConfig is not the source of truth)
|
||||
REDIS_PROGRESS_MARKERS: z.boolean().optional(), // Write per-block live progress markers to Redis instead of jsonb_set UPDATEs on workflow_execution_logs (fallback for the redis-progress-markers flag when AppConfig is not the source of truth)
|
||||
|
||||
// Table feature limits (per plan). Apply when billing is disabled (free tier defaults) or for billed plans.
|
||||
FREE_TABLES_LIMIT: z.number().optional(), // Max user tables per workspace on free tier (default: 5)
|
||||
|
||||
@@ -97,14 +97,6 @@ const FEATURE_FLAGS = {
|
||||
'resolveTriggerRegion, so the whole deployment switches regions together.',
|
||||
fallback: 'TRIGGER_EU_REGION',
|
||||
},
|
||||
'redis-progress-markers': {
|
||||
description:
|
||||
'Write per-block live progress markers (lastStartedBlock/lastCompletedBlock) to Redis ' +
|
||||
'instead of jsonb_set UPDATEs on workflow_execution_logs, folding them into the single ' +
|
||||
'terminal UPDATE at completion. Eliminates the heaviest write query. Resolved once per ' +
|
||||
'logging session (no user/org context) so an execution never mixes write paths.',
|
||||
fallback: 'REDIS_PROGRESS_MARKERS',
|
||||
},
|
||||
'workspace-forking': {
|
||||
description:
|
||||
'Runtime rollout gate for workspace forking (fork/promote/rollback), layered on top of ' +
|
||||
|
||||
@@ -679,7 +679,6 @@ export class ExecutionLogger implements IExecutionLoggerService {
|
||||
isResume?: boolean
|
||||
level?: 'info' | 'error'
|
||||
status?: 'completed' | 'failed' | 'cancelled' | 'pending'
|
||||
readProgressMarkers?: boolean
|
||||
}): Promise<WorkflowExecutionLog> {
|
||||
const {
|
||||
executionId,
|
||||
@@ -695,7 +694,6 @@ export class ExecutionLogger implements IExecutionLoggerService {
|
||||
isResume,
|
||||
level: levelOverride,
|
||||
status: statusOverride,
|
||||
readProgressMarkers = true,
|
||||
} = params
|
||||
|
||||
let execLog = logger.withMetadata({ executionId })
|
||||
@@ -753,7 +751,7 @@ export class ExecutionLogger implements IExecutionLoggerService {
|
||||
models: costSummary.models,
|
||||
}
|
||||
|
||||
const progressMarkers = readProgressMarkers ? await getProgressMarkers(executionId) : null
|
||||
const progressMarkers = await getProgressMarkers(executionId)
|
||||
|
||||
const builtExecutionData = this.buildCompletedExecutionData({
|
||||
existingExecutionData,
|
||||
|
||||
@@ -67,23 +67,17 @@ vi.mock('@/lib/logs/execution/logger', () => ({
|
||||
}))
|
||||
|
||||
const {
|
||||
isFeatureEnabledMock,
|
||||
setLastStartedBlockMock,
|
||||
setLastCompletedBlockMock,
|
||||
getProgressMarkersMock,
|
||||
clearProgressMarkersMock,
|
||||
} = vi.hoisted(() => ({
|
||||
isFeatureEnabledMock: vi.fn().mockResolvedValue(false),
|
||||
setLastStartedBlockMock: vi.fn().mockResolvedValue(false),
|
||||
setLastCompletedBlockMock: vi.fn().mockResolvedValue(false),
|
||||
getProgressMarkersMock: vi.fn().mockResolvedValue({}),
|
||||
clearProgressMarkersMock: vi.fn().mockResolvedValue(undefined),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/core/config/feature-flags', () => ({
|
||||
isFeatureEnabled: isFeatureEnabledMock,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/logs/execution/progress-markers', () => ({
|
||||
setLastStartedBlock: setLastStartedBlockMock,
|
||||
setLastCompletedBlock: setLastCompletedBlockMock,
|
||||
@@ -720,8 +714,7 @@ describe('LoggingSession progress-marker write path', () => {
|
||||
dbMocks.execute.mockResolvedValue(undefined)
|
||||
})
|
||||
|
||||
it('writes markers to Redis (not the row) when the flag is on and Redis accepts the write', async () => {
|
||||
isFeatureEnabledMock.mockResolvedValue(true)
|
||||
it('writes markers to Redis (not the row) when Redis accepts the write', async () => {
|
||||
setLastStartedBlockMock.mockResolvedValue(true)
|
||||
setLastCompletedBlockMock.mockResolvedValue(true)
|
||||
const session = new LoggingSession('wf-1', 'exec-redis', 'manual', 'req-1')
|
||||
@@ -741,8 +734,7 @@ describe('LoggingSession progress-marker write path', () => {
|
||||
expect(dbMocks.execute).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('falls back to the SQL UPDATE when the flag is on but the Redis write fails', async () => {
|
||||
isFeatureEnabledMock.mockResolvedValue(true)
|
||||
it('falls back to the SQL UPDATE when the Redis write fails', async () => {
|
||||
setLastStartedBlockMock.mockResolvedValue(false)
|
||||
const session = new LoggingSession('wf-1', 'exec-redis-down', 'manual', 'req-1')
|
||||
await session.start({ workspaceId: 'ws-1' })
|
||||
@@ -752,46 +744,4 @@ describe('LoggingSession progress-marker write path', () => {
|
||||
expect(setLastStartedBlockMock).toHaveBeenCalled()
|
||||
expect(dbMocks.execute).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('writes markers via jsonb_set UPDATE when the flag is off', async () => {
|
||||
isFeatureEnabledMock.mockResolvedValue(false)
|
||||
const session = new LoggingSession('wf-1', 'exec-sql', 'manual', 'req-1')
|
||||
await session.start({ workspaceId: 'ws-1' })
|
||||
|
||||
await session.onBlockStart('b1', 'Fetch', 'api', '2026-06-27T10:00:00.000Z')
|
||||
|
||||
expect(dbMocks.execute).toHaveBeenCalledTimes(1)
|
||||
expect(setLastStartedBlockMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('falls back to the SQL path when flag resolution throws', async () => {
|
||||
isFeatureEnabledMock.mockRejectedValue(new Error('appconfig unavailable'))
|
||||
const session = new LoggingSession('wf-1', 'exec-fallback', 'manual', 'req-1')
|
||||
await session.start({ workspaceId: 'ws-1' })
|
||||
|
||||
await session.onBlockStart('b1', 'Fetch', 'api', '2026-06-27T10:00:00.000Z')
|
||||
|
||||
expect(dbMocks.execute).toHaveBeenCalledTimes(1)
|
||||
expect(setLastStartedBlockMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('tells completion to read Redis markers only when the flag is on (no wasted ops when off)', async () => {
|
||||
completeWorkflowExecutionMock.mockResolvedValue({})
|
||||
|
||||
isFeatureEnabledMock.mockResolvedValue(true)
|
||||
const onSession = new LoggingSession('wf-1', 'exec-on', 'manual', 'req-1')
|
||||
await onSession.start({ workspaceId: 'ws-1' })
|
||||
await onSession.safeComplete({ finalOutput: { ok: true } })
|
||||
expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({ executionId: 'exec-on', readProgressMarkers: true })
|
||||
)
|
||||
|
||||
isFeatureEnabledMock.mockResolvedValue(false)
|
||||
const offSession = new LoggingSession('wf-1', 'exec-off', 'manual', 'req-1')
|
||||
await offSession.start({ workspaceId: 'ws-1' })
|
||||
await offSession.safeComplete({ finalOutput: { ok: true } })
|
||||
expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({ executionId: 'exec-off', readProgressMarkers: false })
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -4,7 +4,6 @@ import { createLogger } from '@sim/logger'
|
||||
import { describeError, toError } from '@sim/utils/errors'
|
||||
import { and, eq, sql } from 'drizzle-orm'
|
||||
import { releaseExecutionSlot } from '@/lib/billing/calculations/usage-reservation'
|
||||
import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
|
||||
import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
|
||||
import { executionLogger } from '@/lib/logs/execution/logger'
|
||||
import {
|
||||
@@ -136,13 +135,6 @@ export class LoggingSession {
|
||||
private workflowState?: WorkflowState
|
||||
private correlation?: NonNullable<ExecutionTrigger['data']>['correlation']
|
||||
private isResume = false
|
||||
/**
|
||||
* Whether per-block progress markers go to Redis (vs jsonb_set UPDATEs on the
|
||||
* log row). Resolved once in {@link start} and cached so an execution never
|
||||
* mixes write paths across its block callbacks. Defaults to the legacy SQL
|
||||
* path until resolved.
|
||||
*/
|
||||
private useRedisMarkers = false
|
||||
private completed = false
|
||||
/** Synchronous flag to prevent concurrent completion attempts (race condition guard) */
|
||||
private completing = false
|
||||
@@ -182,24 +174,12 @@ export class LoggingSession {
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the per-block marker write path (Redis vs jsonb_set UPDATE) for this
|
||||
* session. Defaults to the legacy SQL path if flag resolution fails.
|
||||
*/
|
||||
private async resolveRedisMarkerMode(): Promise<boolean> {
|
||||
try {
|
||||
return await isFeatureEnabled('redis-progress-markers')
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist the last-started-block marker. Redis is the primary path when the
|
||||
* flag is on; falls back to the durable jsonb_set UPDATE when Redis is
|
||||
* unavailable or the write fails, so a marker is never dropped.
|
||||
* Persist the last-started-block marker. Redis is the primary path; falls back
|
||||
* to the durable jsonb_set UPDATE when Redis is unavailable or the write fails,
|
||||
* so a marker is never dropped.
|
||||
*/
|
||||
private async persistLastStartedBlock(marker: ExecutionLastStartedBlock): Promise<void> {
|
||||
if (this.useRedisMarkers && (await setLastStartedBlock(this.executionId, marker))) {
|
||||
if (await setLastStartedBlock(this.executionId, marker)) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
@@ -220,12 +200,12 @@ export class LoggingSession {
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist the last-completed-block marker. Redis is the primary path when the
|
||||
* flag is on; falls back to the durable jsonb_set UPDATE when Redis is
|
||||
* unavailable or the write fails, so a marker is never dropped.
|
||||
* Persist the last-completed-block marker. Redis is the primary path; falls
|
||||
* back to the durable jsonb_set UPDATE when Redis is unavailable or the write
|
||||
* fails, so a marker is never dropped.
|
||||
*/
|
||||
private async persistLastCompletedBlock(marker: ExecutionLastCompletedBlock): Promise<void> {
|
||||
if (this.useRedisMarkers && (await setLastCompletedBlock(this.executionId, marker))) {
|
||||
if (await setLastCompletedBlock(this.executionId, marker)) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
@@ -308,7 +288,6 @@ export class LoggingSession {
|
||||
isResume: this.isResume,
|
||||
level: params.level,
|
||||
status: params.status,
|
||||
readProgressMarkers: this.useRedisMarkers,
|
||||
})
|
||||
|
||||
// Release the admission reservation from preprocessing. Skipped on pause: a
|
||||
@@ -356,8 +335,6 @@ export class LoggingSession {
|
||||
} = params
|
||||
|
||||
try {
|
||||
this.useRedisMarkers = await this.resolveRedisMarkerMode()
|
||||
|
||||
this.trigger = createTriggerObject(this.triggerType, triggerData)
|
||||
this.correlation = triggerData?.correlation
|
||||
this.environment = createEnvironmentObject(
|
||||
|
||||
@@ -463,12 +463,5 @@ export interface ExecutionLoggerService {
|
||||
isResume?: boolean
|
||||
level?: 'info' | 'error'
|
||||
status?: 'completed' | 'failed' | 'cancelled' | 'pending'
|
||||
/**
|
||||
* Whether this session wrote live progress markers to Redis. When false, the
|
||||
* completion fold skips the Redis read/clear entirely (markers are already on
|
||||
* the row via the SQL path). Defaults to true so non-session callers keep the
|
||||
* safe read-and-fold behavior.
|
||||
*/
|
||||
readProgressMarkers?: boolean
|
||||
}): Promise<WorkflowExecutionLog>
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user