From 464b1cb9e56a6907837b65741fd4c96d17a98424 Mon Sep 17 00:00:00 2001 From: Tobias Date: Wed, 22 Apr 2026 11:09:35 +0200 Subject: [PATCH 1/2] fix(watchdog): raise INACTIVE and INACTIVITY timeouts to 30 min Both INACTIVE_TIMEOUT_MS (10 min) and INACTIVITY_TIMEOUT_MS (15 min) were killing Fargate tasks mid-conversation when the CloudWatch MessageLatency history for the current KST hour had fewer than ACTIVE_HOUR_THRESHOLD=2 matching datapoints. Raising both to match ACTIVE_TIMEOUT_MS (30 min) makes the sparse-history misclassification harmless. Investigation details in docs/fargate-watchdog-investigation.md. Tests updated to reflect the new uniform 30-min cutoff. --- .../__tests__/handlers/watchdog.test.ts | 38 +++++++++---------- packages/shared/src/constants.ts | 4 +- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/packages/gateway/__tests__/handlers/watchdog.test.ts b/packages/gateway/__tests__/handlers/watchdog.test.ts index 7fd95af..1e13246 100644 --- a/packages/gateway/__tests__/handlers/watchdog.test.ts +++ b/packages/gateway/__tests__/handlers/watchdog.test.ts @@ -19,7 +19,10 @@ vi.mock("@aws-sdk/client-dynamodb", () => ({ vi.mock("@aws-sdk/client-ecs", () => ({ ECSClient: vi.fn(() => ({ send: mockEcsSend })), StopTaskCommand: vi.fn((params: unknown) => ({ input: params, _tag: "StopTaskCommand" })), - DescribeTasksCommand: vi.fn((params: unknown) => ({ input: params, _tag: "DescribeTasksCommand" })), + DescribeTasksCommand: vi.fn((params: unknown) => ({ + input: params, + _tag: "DescribeTasksCommand", + })), })); vi.mock("@aws-sdk/client-cloudwatch", () => ({ @@ -31,21 +34,21 @@ describe("watchdog handler", () => { beforeEach(() => { vi.clearAllMocks(); vi.stubEnv("ECS_CLUSTER_ARN", "arn:cluster"); - // Default: CW returns no data → fallback 15-min timeout + // Default: CW returns no data → fallback 30-min timeout mockCloudWatchSend.mockResolvedValue({ Datapoints: [] }); }); - it("should stop tasks inactive for more than 15 minutes", async () => { + it("should stop tasks inactive for more than 30 minutes", async () => { const { handler } = await import("../../src/handlers/watchdog.js"); - const oldTime = new Date(Date.now() - 20 * 60 * 1000).toISOString(); + const oldTime = new Date(Date.now() - 35 * 60 * 1000).toISOString(); mockDynamoSend.mockResolvedValueOnce({ Items: [ { PK: "USER#user-1", taskArn: "arn:task-1", status: "Running", - startedAt: new Date(Date.now() - 30 * 60 * 1000).toISOString(), + startedAt: new Date(Date.now() - 40 * 60 * 1000).toISOString(), lastActivity: oldTime, }, ], @@ -303,7 +306,7 @@ describe("watchdog handler", () => { it("should stop tasks with expired prewarmUntil when inactive", async () => { const { handler } = await import("../../src/handlers/watchdog.js"); - const oldTime = new Date(Date.now() - 20 * 60 * 1000).toISOString(); + const oldTime = new Date(Date.now() - 35 * 60 * 1000).toISOString(); mockDynamoSend.mockResolvedValueOnce({ Items: [ { @@ -311,7 +314,7 @@ describe("watchdog handler", () => { taskArn: "arn:prewarm-task", status: "Running", publicIp: "1.2.3.4", - startedAt: new Date(Date.now() - 30 * 60 * 1000).toISOString(), + startedAt: new Date(Date.now() - 40 * 60 * 1000).toISOString(), lastActivity: oldTime, prewarmUntil: Date.now() - 5 * 60 * 1000, // expired 5 min ago }, @@ -380,7 +383,7 @@ describe("watchdog handler", () => { ); }); - it("should use 10-min timeout during inactive hours (< 2 total datapoints at current hour)", async () => { + it("should use 30-min timeout during inactive hours (< 2 total datapoints at current hour)", async () => { const { handler } = await import("../../src/handlers/watchdog.js"); const now = new Date(); @@ -388,18 +391,15 @@ describe("watchdog handler", () => { // First channel (telegram): 1 datapoint at current hour mockCloudWatchSend.mockResolvedValueOnce({ - Datapoints: [ - { Timestamp: createTimestampForKSTHour(currentHourKST, 1), SampleCount: 1 }, - ], + Datapoints: [{ Timestamp: createTimestampForKSTHour(currentHourKST, 1), SampleCount: 1 }], }); // Second channel (web): 0 datapoints → total = 1, below threshold of 2 mockCloudWatchSend.mockResolvedValueOnce({ Datapoints: [], }); - // Task inactive for 12 min — would NOT be stopped with 15-min default, - // but SHOULD be stopped with 10-min inactive timeout - const lastActivity = new Date(Date.now() - 12 * 60 * 1000).toISOString(); + // Task inactive for 35 min — exceeds the 30-min inactive timeout + const lastActivity = new Date(Date.now() - 35 * 60 * 1000).toISOString(); mockDynamoSend.mockResolvedValueOnce({ Items: [ { @@ -420,7 +420,7 @@ describe("watchdog handler", () => { await handler(); - // Should stop — 12 min > 10 min inactive timeout + // Should stop — 35 min > 30 min inactive timeout expect(mockEcsSend).toHaveBeenCalledWith( expect.objectContaining({ input: expect.objectContaining({ @@ -463,13 +463,13 @@ describe("watchdog handler", () => { ); }); - it("should fall back to 15-min timeout when CW returns empty data", async () => { + it("should fall back to 30-min timeout when CW returns empty data", async () => { const { handler } = await import("../../src/handlers/watchdog.js"); mockCloudWatchSend.mockResolvedValue({ Datapoints: [] }); - // Task inactive for 20 min — should be stopped with 15-min fallback - const lastActivity = new Date(Date.now() - 20 * 60 * 1000).toISOString(); + // Task inactive for 35 min — should be stopped with 30-min fallback + const lastActivity = new Date(Date.now() - 35 * 60 * 1000).toISOString(); mockDynamoSend.mockResolvedValueOnce({ Items: [ { @@ -490,7 +490,7 @@ describe("watchdog handler", () => { await handler(); - // Should stop — 20 min > 15 min fallback + // Should stop — 35 min > 30 min fallback expect(mockEcsSend).toHaveBeenCalledWith( expect.objectContaining({ input: expect.objectContaining({ diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index 6cfb046..7c9fc9b 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -22,7 +22,7 @@ export const BRIDGE_HTTP_TIMEOUT_MS = 3000; export const GATEWAY_PORT = 18789; // Timeouts (ms) -export const INACTIVITY_TIMEOUT_MS = 15 * 60 * 1000; +export const INACTIVITY_TIMEOUT_MS = 30 * 60 * 1000; export const PENDING_MESSAGE_TTL_SEC = 5 * 60; export const CONNECTION_TTL_SEC = 24 * 60 * 60; export const PERIODIC_BACKUP_INTERVAL_MS = 5 * 60 * 1000; @@ -46,7 +46,7 @@ export const DEFAULT_PREWARM_DURATION_MIN = 60; // Dynamic Timeout export const ACTIVE_TIMEOUT_MS = 30 * 60 * 1000; // 30 min — active hours -export const INACTIVE_TIMEOUT_MS = 10 * 60 * 1000; // 10 min — inactive hours +export const INACTIVE_TIMEOUT_MS = 30 * 60 * 1000; // 30 min — inactive hours (raised to match ACTIVE after investigation 2026-04-22) export const ACTIVITY_LOOKBACK_DAYS = 7; export const ACTIVE_HOUR_THRESHOLD = 2; // >= 2 days with activity at this hour export const METRICS_NAMESPACE = "ServerlessOpenClaw"; From 506ff9c998d671d0d379e05ea353c18dbefea190 Mon Sep 17 00:00:00 2001 From: Tobias Date: Wed, 22 Apr 2026 18:01:13 +0200 Subject: [PATCH 2/2] fix(gateway): refresh lastActivity when routing to running container MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The routeFargate happy path (task already Running with publicIp) called sendToBridge and returned 'sent' without writing lastActivity back to DynamoDB. lastActivity was only advanced on task start or prewarm claim, so an active conversation with many messages to an already-running container looked idle to the watchdog and got killed mid-session. Reproduced in production on 2026-04-22: task started at 14:57:37 UTC, user pinged at 15:00:30 UTC (gateway logs show 'taskStatus: Running, needsColdStart: false, message routed successfully'), but DynamoDB lastActivity stayed pinned to 14:57:37. Watchdog killed the task at 15:27:49 UTC for 'inactivity timeout' (exit 137) despite a ping 27 min earlier. Fix: add updateLastActivity() helper using UpdateCommand (single-attribute, atomic) and call it after every successful bridge delivery on the happy path. UpdateCommand avoids the whole-item-replace race a prior PutCommand attempt hit against the container's concurrent lifecycle.updateTaskState writes — it only touches the lastActivity attribute. Live verified post-deploy: three successive pings to the same running task all advanced lastActivity in DynamoDB (14:57:37 baseline -> 15:37:10 -> 15:46:26 -> 15:50:58), and ResponseLength stayed non-zero on every reply. --- .../__tests__/services/message.test.ts | 2 ++ .../__tests__/services/task-state.test.ts | 26 ++++++++++++++- .../gateway/src/handlers/telegram-webhook.ts | 8 ++++- packages/gateway/src/handlers/ws-message.ts | 8 ++++- packages/gateway/src/services/message.ts | 9 +++++ packages/gateway/src/services/task-state.ts | 33 +++++++++++-------- 6 files changed, 70 insertions(+), 16 deletions(-) diff --git a/packages/gateway/__tests__/services/message.test.ts b/packages/gateway/__tests__/services/message.test.ts index 3e22aa5..991ade6 100644 --- a/packages/gateway/__tests__/services/message.test.ts +++ b/packages/gateway/__tests__/services/message.test.ts @@ -127,6 +127,7 @@ describe("message service", () => { getTaskState: vi.fn().mockResolvedValue(null), startTask: vi.fn().mockResolvedValue("arn:new-task"), putTaskState: vi.fn(), + updateLastActivity: vi.fn(), savePendingMessage: vi.fn(), deleteTaskState: vi.fn(), ...overrides, @@ -151,6 +152,7 @@ describe("message service", () => { expect(mockFetch).toHaveBeenCalled(); expect(deps.startTask).not.toHaveBeenCalled(); expect(deps.deleteTaskState).not.toHaveBeenCalled(); + expect(deps.updateLastActivity).toHaveBeenCalledWith("user-123"); }); it("should save pending + start task when no active task", async () => { diff --git a/packages/gateway/__tests__/services/task-state.test.ts b/packages/gateway/__tests__/services/task-state.test.ts index 9b365e6..c4f6a20 100644 --- a/packages/gateway/__tests__/services/task-state.test.ts +++ b/packages/gateway/__tests__/services/task-state.test.ts @@ -1,9 +1,11 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; -import { getTaskState, putTaskState } from "../../src/services/task-state.js"; +import { getTaskState, putTaskState, updateLastActivity } from "../../src/services/task-state.js"; vi.mock("@aws-sdk/lib-dynamodb", () => ({ GetCommand: vi.fn((params: unknown) => ({ input: params, _tag: "GetCommand" })), PutCommand: vi.fn((params: unknown) => ({ input: params, _tag: "PutCommand" })), + DeleteCommand: vi.fn((params: unknown) => ({ input: params, _tag: "DeleteCommand" })), + UpdateCommand: vi.fn((params: unknown) => ({ input: params, _tag: "UpdateCommand" })), })); describe("task-state service", () => { @@ -109,4 +111,26 @@ describe("task-state service", () => { ); }); }); + + describe("updateLastActivity", () => { + it("should issue an UpdateCommand touching only lastActivity", async () => { + mockSend.mockResolvedValueOnce({}); + + await updateLastActivity(mockSend, "user-123"); + + expect(mockSend).toHaveBeenCalledWith( + expect.objectContaining({ + _tag: "UpdateCommand", + input: expect.objectContaining({ + TableName: expect.stringContaining("TaskState"), + Key: { PK: "USER#user-123" }, + UpdateExpression: "SET lastActivity = :la", + ExpressionAttributeValues: expect.objectContaining({ + ":la": expect.any(String), + }), + }), + }), + ); + }); + }); }); diff --git a/packages/gateway/src/handlers/telegram-webhook.ts b/packages/gateway/src/handlers/telegram-webhook.ts index 56d6664..36776ed 100644 --- a/packages/gateway/src/handlers/telegram-webhook.ts +++ b/packages/gateway/src/handlers/telegram-webhook.ts @@ -4,7 +4,12 @@ import { DynamoDBClient } from "@aws-sdk/client-dynamodb"; import { DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb"; import { ECSClient } from "@aws-sdk/client-ecs"; -import { getTaskState, putTaskState, deleteTaskState } from "../services/task-state.js"; +import { + getTaskState, + putTaskState, + deleteTaskState, + updateLastActivity, +} from "../services/task-state.js"; import { routeMessage, savePendingMessage } from "../services/message.js"; import { startTask } from "../services/container.js"; import { sendTelegramMessage } from "../services/telegram.js"; @@ -169,6 +174,7 @@ export async function handler(event: { getTaskState: (uid) => getTaskState(dynamoSend, uid), startTask: (params) => startTask(ecsSend, params), putTaskState: (item) => putTaskState(dynamoSend, item), + updateLastActivity: (uid) => updateLastActivity(dynamoSend, uid), savePendingMessage: (item) => savePendingMessage(dynamoSend, item), deleteTaskState: (uid) => deleteTaskState(dynamoSend, uid), startTaskParams: { diff --git a/packages/gateway/src/handlers/ws-message.ts b/packages/gateway/src/handlers/ws-message.ts index 67efe3d..59f556f 100644 --- a/packages/gateway/src/handlers/ws-message.ts +++ b/packages/gateway/src/handlers/ws-message.ts @@ -9,7 +9,12 @@ import { import type { ClientMessage, ServerMessage } from "@serverless-openclaw/shared"; import { getConnection } from "../services/connections.js"; -import { getTaskState, putTaskState, deleteTaskState } from "../services/task-state.js"; +import { + getTaskState, + putTaskState, + deleteTaskState, + updateLastActivity, +} from "../services/task-state.js"; import { routeMessage, savePendingMessage } from "../services/message.js"; import { startTask } from "../services/container.js"; import { resolveSecrets } from "../services/secrets.js"; @@ -89,6 +94,7 @@ export async function handler(event: { getTaskState: (uid) => getTaskState(dynamoSend, uid), startTask: (params) => startTask(ecsSend, params), putTaskState: (item) => putTaskState(dynamoSend, item), + updateLastActivity: (uid) => updateLastActivity(dynamoSend, uid), savePendingMessage: (item) => savePendingMessage(dynamoSend, item), deleteTaskState: (uid) => deleteTaskState(dynamoSend, uid), startTaskParams: { diff --git a/packages/gateway/src/services/message.ts b/packages/gateway/src/services/message.ts index 2148aad..0a57ce5 100644 --- a/packages/gateway/src/services/message.ts +++ b/packages/gateway/src/services/message.ts @@ -64,6 +64,7 @@ export interface RouteDeps { getTaskState: (userId: string) => Promise; startTask: (params: StartTaskParams) => Promise; putTaskState: (item: TaskStateItem) => Promise; + updateLastActivity: (userId: string) => Promise; savePendingMessage: (item: PendingMessageItem) => Promise; deleteTaskState: (userId: string) => Promise; startTaskParams: StartTaskParams; @@ -92,6 +93,14 @@ async function routeFargate( connectionId: deps.connectionId, callbackUrl: deps.callbackUrl, }); + // Refresh lastActivity so the watchdog doesn't kill an active container. + // UpdateCommand (single attribute) avoids clobbering concurrent writes + // from the container's lifecycle.updateTaskState(). + try { + await deps.updateLastActivity(deps.userId); + } catch (err) { + console.warn("Failed to refresh lastActivity, continuing", err); + } return "sent"; } catch (err) { console.warn( diff --git a/packages/gateway/src/services/task-state.ts b/packages/gateway/src/services/task-state.ts index 5432064..f6d0c18 100644 --- a/packages/gateway/src/services/task-state.ts +++ b/packages/gateway/src/services/task-state.ts @@ -1,13 +1,10 @@ -import { GetCommand, PutCommand, DeleteCommand } from "@aws-sdk/lib-dynamodb"; +import { GetCommand, PutCommand, DeleteCommand, UpdateCommand } from "@aws-sdk/lib-dynamodb"; import { TABLE_NAMES, KEY_PREFIX } from "@serverless-openclaw/shared"; import type { TaskStateItem } from "@serverless-openclaw/shared"; type Send = (command: unknown) => Promise; -export async function getTaskState( - send: Send, - userId: string, -): Promise { +export async function getTaskState(send: Send, userId: string): Promise { const result = (await send( new GetCommand({ TableName: TABLE_NAMES.TASK_STATE, @@ -20,10 +17,7 @@ export async function getTaskState( return item; } -export async function putTaskState( - send: Send, - item: TaskStateItem, -): Promise { +export async function putTaskState(send: Send, item: TaskStateItem): Promise { await send( new PutCommand({ TableName: TABLE_NAMES.TASK_STATE, @@ -32,10 +26,7 @@ export async function putTaskState( ); } -export async function deleteTaskState( - send: Send, - userId: string, -): Promise { +export async function deleteTaskState(send: Send, userId: string): Promise { await send( new DeleteCommand({ TableName: TABLE_NAMES.TASK_STATE, @@ -43,3 +34,19 @@ export async function deleteTaskState( }), ); } + +/** + * Atomically refresh `lastActivity` on an existing TaskState row. + * Uses UpdateCommand (single-attribute) to avoid clobbering concurrent writes + * from the container (e.g. `updateTaskState("Running", publicIp)` at boot). + */ +export async function updateLastActivity(send: Send, userId: string): Promise { + await send( + new UpdateCommand({ + TableName: TABLE_NAMES.TASK_STATE, + Key: { PK: `${KEY_PREFIX.USER}${userId}` }, + UpdateExpression: "SET lastActivity = :la", + ExpressionAttributeValues: { ":la": new Date().toISOString() }, + }), + ); +}