Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions apps/sim/app/api/billing/update-cost/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ async function updateCostInner(req: NextRequest, span: Span): Promise<NextRespon
const startTime = Date.now()

try {
logger.info(`[${requestId}] Update cost request started`)
logger.debug(`[${requestId}] Update cost request started`)

// Check authentication (internal API key)
const authResult = checkInternalApiKey(req)
Expand Down Expand Up @@ -324,7 +324,7 @@ async function updateCostInner(req: NextRequest, span: Span): Promise<NextRespon
...(idempotencyKey ? { [TraceAttr.BillingIdempotencyKey]: idempotencyKey } : {}),
})

logger.info(`[${requestId}] Processing cost update`, {
logger.debug(`[${requestId}] Processing cost update`, {
userId,
cost,
model,
Expand Down Expand Up @@ -408,7 +408,7 @@ async function updateCostInner(req: NextRequest, span: Span): Promise<NextRespon
}
: undefined

logger.info(`[${requestId}] Billing payer resolved`, {
logger.debug(`[${requestId}] Billing payer resolved`, {
userId,
billingProtocol: protocol,
billingResolution: getBillingResolution(isMarkerlessLegacy, billingAttribution),
Expand Down Expand Up @@ -443,7 +443,10 @@ async function updateCostInner(req: NextRequest, span: Span): Promise<NextRespon
const billed = result.billed
logger.info(`[${requestId}] Cumulative cost top-up`, {
userId,
billingPayer: billingContext?.billingEntity ?? { type: 'user', id: userId },
workspaceId: resolvedWorkspaceId,
source,
model,
cumulativeCost: cost,
billedDelta: result.delta,
newTotal: result.total,
Expand Down
4 changes: 2 additions & 2 deletions apps/sim/app/ingest/[[...path]]/route.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ describe('PostHog ingest proxy', () => {
const response = await POST(createIngestRequest(controller.signal))

expect(response.status).toBe(499)
expect(logger.error).not.toHaveBeenCalled()
expect(logger.warn).not.toHaveBeenCalled()
})

it('reports an upstream failure while the client is still connected', async () => {
Expand All @@ -40,7 +40,7 @@ describe('PostHog ingest proxy', () => {
const response = await POST(createIngestRequest())

expect(response.status).toBe(502)
expect(logger.error).toHaveBeenCalledWith(
expect(logger.warn).toHaveBeenCalledWith(
'PostHog proxy error',
expect.objectContaining({ error: 'The operation timed out.' })
)
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/app/ingest/[[...path]]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ async function handler(request: NextRequest) {
logger.info('Client closed PostHog proxy request', { method: request.method })
return new NextResponse(null, { status: 499 })
}
logger.error('PostHog proxy error', {
logger.warn('PostHog proxy error', {
url,
method: request.method,
error: toError(error).message,
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/background/webhook-execution.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1100,7 +1100,7 @@ async function executeWebhookJobInternal(
}
}

logger.info(`[${requestId}] Executing workflow for ${payload.provider} webhook`)
logger.debug(`[${requestId}] Executing workflow for ${payload.provider} webhook`)

const metadata: ExecutionMetadata = {
requestId,
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/executor/handlers/trigger/handler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ export class TriggerBlockHandler implements BlockHandler {
block: SerializedBlock,
inputs: Record<string, any>
): Promise<any> {
logger.info(`Executing trigger block: ${block.id} (Type: ${block.metadata?.id})`)
logger.debug(`Executing trigger block: ${block.id} (Type: ${block.metadata?.id})`)

if (block.metadata?.id === BlockType.STARTER) {
return this.executeStarterBlock(ctx, block, inputs)
Expand Down
5 changes: 4 additions & 1 deletion apps/sim/lib/core/idempotency/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -714,7 +714,10 @@ export class IdempotencyService {
)
}

logger.warn(`Operation failed: ${claimResult.normalizedKey} - ${errorMessage}`)
logger.warn('Idempotent operation failed', {
key: claimResult.normalizedKey,
error: errorMessage,
})
throw error
}
}
Expand Down
17 changes: 17 additions & 0 deletions apps/sim/lib/execution/preprocessing.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import {
billingUsageReservationMockFns,
} from '@sim/testing/mocks/billing-usage-reservation.mock'
import { executionLimitsMock } from '@sim/testing/mocks/execution-limits.mock'
import { getMockLogger } from '@sim/testing/mocks/logger.mock'
import { loggingSessionMockFns } from '@sim/testing/mocks/logging-session.mock'
import { createMockRedis } from '@sim/testing/mocks/redis.mock'
import { redisConfigMockFns, resetRedisConfigMock } from '@sim/testing/mocks/redis-config.mock'
Expand Down Expand Up @@ -1123,6 +1124,22 @@ describe('preprocessExecution admission rejection codes and blocked-run log thro
expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledTimes(1)
})

it('logs one warning for repeated refusals of a workflow by the same gate', async () => {
const workflowId = 'workflow-1'
const logger = getMockLogger('ExecutionPreprocessing')
logger.warn.mockClear()

for (let attempt = 0; attempt < 3; attempt++) {
await refuse(workflowId, { throttleErrorLogs: true })
}

expect(logger.warn).toHaveBeenCalledOnce()
expect(logger.warn).toHaveBeenCalledWith(
'Attributed usage gate blocked actor',
expect.objectContaining({ workflowId })
)
})

it('writes a row for a different gate refusing the same workflow inside the window', async () => {
const workflowId = 'workflow-1'
await refuse(workflowId, { throttleErrorLogs: true })
Expand Down
103 changes: 57 additions & 46 deletions apps/sim/lib/execution/preprocessing.ts
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,12 @@ function startAdmissionReads(params: {
}
}

/** A gate refusal's warning, emitted alongside its error row so both share the throttle window. */
interface GateWarning {
message: string
fields: Record<string, unknown>
}

export async function preprocessExecution(
options: PreprocessExecutionOptions
): Promise<PreprocessExecutionResult> {
Expand Down Expand Up @@ -401,11 +407,15 @@ export async function preprocessExecution(
const isFailureLogSuppressed = (failure: PreprocessExecutionError): boolean =>
suppressRetryableFailureLogs && failure.statusCode >= 500 && failure.retryable === true

/** Records an admission gate's error row, at most once per gate and outcome per window when throttled. */
/**
* Records an admission gate's warning and error row, at most once per gate and outcome per window
* when throttled, so a sender redelivering into a refusal does not log every attempt.
*/
const recordGateFailure = async (
gate: 'ban' | 'usage' | 'rate-limit' | 'reservation',
failure: PreprocessExecutionError,
record: Parameters<typeof logPreprocessingError>[0]
record: Parameters<typeof logPreprocessingError>[0],
warning?: GateWarning
): Promise<void> => {
if (isFailureLogSuppressed(failure)) return
if (
Expand All @@ -416,10 +426,12 @@ export async function preprocessExecution(
) {
return
}
if (warning)
logger.warn(warning.message, { requestId, workflowId, triggerType, ...warning.fields })
await recordPreprocessingError(record)
}

logger.info(`[${requestId}] Starting execution preprocessing`, {
logger.debug(`[${requestId}] Starting execution preprocessing`, {
workflowId,
userId,
triggerType,
Expand Down Expand Up @@ -577,21 +589,21 @@ export async function preprocessExecution(
}
actorUserId = validatedAttribution.actorUserId
billingAttribution = validatedAttribution
logger.info(`[${requestId}] Reusing serialized billing attribution`, {
logger.debug(`[${requestId}] Reusing serialized billing attribution`, {
actorUserId,
billingEntity: billingAttribution.billingEntity,
})
}

if (!actorUserId && useAuthenticatedUserAsActor && userId) {
actorUserId = userId
logger.info(`[${requestId}] Using authenticated user as actor: ${actorUserId}`)
logger.debug(`[${requestId}] Using authenticated user as actor: ${actorUserId}`)
}

if (!actorUserId) {
billingAttribution = await admission.attributionFor(workspaceId)
actorUserId = billingAttribution.actorUserId
logger.info(`[${requestId}] Using atomically resolved system actor and payer`, {
logger.debug(`[${requestId}] Using atomically resolved system actor and payer`, {
actorUserId,
billingEntity: billingAttribution.billingEntity,
})
Expand Down Expand Up @@ -672,6 +684,7 @@ export async function preprocessExecution(
interface GateFailure {
response: PreprocessExecutionFailure
recordError?: Parameters<typeof recordPreprocessingError>[0]
warning?: GateWarning
}

/** Usage figures captured by the read gate and reused by the atomic reservation. */
Expand All @@ -690,13 +703,8 @@ export async function preprocessExecution(
try {
const bannedUserIds = await gateReads.bannedUserIds
if (bannedUserIds.length > 0) {
logger.warn(`[${requestId}] Execution blocked: banned account`, {
workflowId,
bannedUserIds,
triggerType,
})

return {
warning: { message: 'Execution blocked: banned account', fields: { bannedUserIds } },
response: {
success: false,
error: {
Expand Down Expand Up @@ -774,16 +782,17 @@ export async function preprocessExecution(
}
: null
if (usageCheck.isExceeded) {
logger.warn(`[${requestId}] Attributed usage gate blocked actor ${actorUserId}.`, {
currentUsage: snapshot?.currentUsage,
limit: snapshot?.limit,
scope: usageCheck.scope,
workflowId,
triggerType,
})

return {
failure: {
warning: {
message: 'Attributed usage gate blocked actor',
fields: {
actorUserId,
currentUsage: snapshot?.currentUsage,
limit: snapshot?.limit,
scope: usageCheck.scope,
},
},
response: {
success: false,
error: {
Expand Down Expand Up @@ -876,13 +885,11 @@ export async function preprocessExecution(
)

if (!info.allowed) {
logger.warn(`[${requestId}] Rate limit exceeded for user ${actorUserId}`, {
triggerType,
remaining: info.remaining,
resetAt: info.resetAt,
})

return {
warning: {
message: 'Rate limit exceeded',
fields: { actorUserId, remaining: info.remaining, resetAt: info.resetAt },
},
response: {
success: false,
error: {
Expand Down Expand Up @@ -947,7 +954,8 @@ export async function preprocessExecution(
await recordGateFailure(
banFailure ? 'ban' : 'usage',
readGateFailure.response.error,
readGateFailure.recordError
readGateFailure.recordError,
readGateFailure.warning
)
}
return readGateFailure.response
Expand All @@ -959,7 +967,8 @@ export async function preprocessExecution(
await recordGateFailure(
'rate-limit',
rateLimitFailure.response.error,
rateLimitFailure.recordError
rateLimitFailure.recordError,
rateLimitFailure.warning
)
}
return rateLimitFailure.response
Expand Down Expand Up @@ -1009,12 +1018,6 @@ export async function preprocessExecution(
reservation.reason
)
const message = RESERVATION_DENIAL_MESSAGE[reservation.reason]
logger.warn(`[${requestId}] Admission reservation full for user ${actorUserId}`, {
workflowId,
triggerType,
constraint: reservation.reason,
})

const failure: PreprocessExecutionError = {
message,
statusCode: descriptor.statusCode,
Expand All @@ -1026,17 +1029,25 @@ export async function preprocessExecution(
constraint: reservation.reason,
},
}
await recordGateFailure('reservation', failure, {
workflowId,
executionId,
triggerType,
requestId,
userId: actorUserId,
workspaceId,
errorMessage: message,
loggingSession: providedLoggingSession,
triggerData,
})
await recordGateFailure(
'reservation',
failure,
{
workflowId,
executionId,
triggerType,
requestId,
userId: actorUserId,
workspaceId,
errorMessage: message,
loggingSession: providedLoggingSession,
triggerData,
},
{
message: 'Admission reservation full',
fields: { actorUserId, constraint: reservation.reason },
}
)

return { success: false, error: failure }
}
Expand Down Expand Up @@ -1068,7 +1079,7 @@ export async function preprocessExecution(
}
}

logger.info(`[${requestId}] All preprocessing checks passed`, {
logger.debug(`[${requestId}] All preprocessing checks passed`, {
workflowId,
actorUserId,
triggerType,
Expand Down
4 changes: 2 additions & 2 deletions apps/sim/lib/oauth/credential-service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1764,7 +1764,7 @@ export async function resolveCredentialTokenBundle(
logger.error(`[${requestId}] Missing access token for credential`)
return null
}
if (!result.refreshed) logger.info(`[${requestId}] Access token is valid for credential`)
if (!result.refreshed) logger.debug(`[${requestId}] Access token is valid for credential`)
return { accessToken: result.accessToken }
}

Expand Down Expand Up @@ -1817,7 +1817,7 @@ export async function refreshTokenIfNeeded(
throw new Error('OAuth access token expired and cannot be refreshed; reconnect the account')
}
if (result.kind === 'token') {
if (!result.refreshed) logger.info(`[${requestId}] Access token is valid`)
if (!result.refreshed) logger.debug(`[${requestId}] Access token is valid`)
/** A row storing neither token returns its empty token as before; callers decide how to fail. */
return {
accessToken: result.accessToken ?? credential.accessToken,
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/lib/table/rows/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -959,7 +959,7 @@ export async function upsertRow(
}
})

logger.info(
logger.debug(
`[${requestId}] Upserted (${result.operation}) row ${result.row.id} in table ${data.tableId}`
)

Expand Down
2 changes: 1 addition & 1 deletion apps/sim/lib/uploads/core/storage-service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,7 @@ export async function uploadFile(options: UploadFileOptions): Promise<FileInfo>
throw new Error('Upload cleanup requires a newly allocated execution or Copilot key')
}

logger.info(`Uploading file to ${context} storage: ${fileName}`)
logger.debug(`Uploading file to ${context} storage: ${fileName}`)

const config = getStorageConfig(context)

Expand Down
2 changes: 1 addition & 1 deletion apps/sim/lib/webhooks/processor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -648,7 +648,7 @@ export async function checkWebhookPreprocessing(
const error = preprocessResult.error
const transientAdmissionFailure = classifyTransientAdmissionFailure(error)
const admissionRejectionCode = getDeterministicAdmissionRejectionCode(error)
logger.warn(`[${requestId}] Webhook preprocessing failed`, {
logger.debug(`[${requestId}] Webhook preprocessing failed`, {
provider: foundWebhook.provider,
error: error.message,
statusCode: error.statusCode,
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/lib/webhooks/slack-dispatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ export async function dispatchSlackWebhooks(
if (result.outcome === 'ignored' && result.reason === 'filtered') {
const rawEvent = payload.event as Record<string, unknown> | undefined
const providerConfig = toRecord(foundWebhook.providerConfig)
logger.info(
logger.debug(
`[${requestId}] Event skipped by trigger filter for webhook ${foundWebhook.id}`,
{
eventKey: resolveSlackEventKey(payload),
Expand Down
Loading