diff --git a/.runtime/portal/RUNBOOK.txt b/.runtime/portal/RUNBOOK.txt index 75f8430..fbf3e02 100644 --- a/.runtime/portal/RUNBOOK.txt +++ b/.runtime/portal/RUNBOOK.txt @@ -36,6 +36,8 @@ Deployment and operations transport: Streaming runtime operations: node scripts/check-stream-runtime.mjs node scripts/runtime-worker-drain.mjs status + node scripts/runtime-worker-drain.mjs reconcile + node scripts/runtime-worker-drain.mjs reconcile --apply node scripts/runtime-worker-drain.mjs drain goosed-3 node scripts/runtime-worker-drain.mjs undrain goosed-3 node scripts/check-tool-runtime.mjs diff --git a/.runtime/portal/scripts/runtime-worker-drain.mjs b/.runtime/portal/scripts/runtime-worker-drain.mjs index e9dbbff..1317723 100755 --- a/.runtime/portal/scripts/runtime-worker-drain.mjs +++ b/.runtime/portal/scripts/runtime-worker-drain.mjs @@ -16,7 +16,7 @@ function loadEnvFile(filePath) { } } -loadEnvFile(path.join(process.cwd(), '.env')); +loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env')); const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0'; const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime'; @@ -26,17 +26,24 @@ const configuredWorkers = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_ .filter(Boolean) .map((_, index) => `goosed-${index + 1}`); const action = process.argv[2] || 'status'; -const workerId = process.argv[3] || null; +const positionalArgs = process.argv.slice(3).filter((arg) => !arg.startsWith('--')); +const workerId = positionalArgs[0] || null; +const apply = process.argv.includes('--apply'); +const staleMsArg = process.argv.find((arg) => arg.startsWith('--stale-ms=')); +const staleMs = Math.max( + 60_000, + Number(staleMsArg?.split('=').at(1) || process.env.MEMIND_RUNTIME_STALE_STREAM_MS || 15 * 60 * 1000), +); function usage() { - console.error('Usage: node scripts/runtime-worker-drain.mjs [goosed-N]'); + console.error('Usage: node scripts/runtime-worker-drain.mjs [goosed-N] [--apply] [--stale-ms=900000]'); } function workerKey(id, field) { return [namespace, 'worker', id, field].join(':'); } -if (!['status', 'drain', 'undrain'].includes(action)) { +if (!['status', 'drain', 'undrain', 'reconcile'].includes(action)) { usage(); process.exit(2); } @@ -68,6 +75,7 @@ const workers = [ if (workerId && !workers.includes(workerId)) workers.push(workerId); const rows = []; +const now = Date.now(); for (const id of [...new Set(workers)].sort()) { const values = await client.mGet([ workerKey(id, 'active_streams'), @@ -77,16 +85,41 @@ for (const id of [...new Set(workers)].sort()) { workerKey(id, 'stream_error_count'), workerKey(id, 'last_stream_started_at'), workerKey(id, 'last_stream_ended_at'), + workerKey(id, 'last_stream_reconciled_at'), + workerKey(id, 'stream_reconcile_count'), ]); + const activeStreams = Number(values[0] || 0); + const lastStreamStartedAt = values[5] ? Number(values[5]) : null; + const lastStreamEndedAt = values[6] ? Number(values[6]) : null; + const staleAgeMs = lastStreamStartedAt ? now - lastStreamStartedAt : null; + const stale = Boolean( + activeStreams > 0 && + lastStreamStartedAt && + staleAgeMs > staleMs && + (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt), + ); + if (action === 'reconcile' && stale && apply) { + await client + .multi() + .set(workerKey(id, 'active_streams'), '0') + .set(workerKey(id, 'last_stream_reconciled_at'), String(now)) + .incr(workerKey(id, 'stream_reconcile_count')) + .exec(); + } rows.push({ id, - activeStreams: Number(values[0] || 0), + activeStreams, drain: /^(1|true|yes)$/i.test(String(values[1] || '')), streamOpenCount: Number(values[2] || 0), streamAbortCount: Number(values[3] || 0), streamErrorCount: Number(values[4] || 0), - lastStreamStartedAt: values[5] ? Number(values[5]) : null, - lastStreamEndedAt: values[6] ? Number(values[6]) : null, + lastStreamStartedAt, + lastStreamEndedAt, + lastStreamReconciledAt: values[7] ? Number(values[7]) : null, + streamReconcileCount: Number(values[8] || 0), + stale, + staleAgeMs, + reconciled: action === 'reconcile' && stale && apply, }); } @@ -97,5 +130,7 @@ console.log(JSON.stringify({ action, workerId, namespace, + dryRun: action === 'reconcile' ? !apply : undefined, + staleMs: action === 'reconcile' ? staleMs : undefined, workers: rows, }, null, 2)); diff --git a/scripts/build-portal-runtime.mjs b/scripts/build-portal-runtime.mjs index 26d7cc9..0633174 100755 --- a/scripts/build-portal-runtime.mjs +++ b/scripts/build-portal-runtime.mjs @@ -324,6 +324,8 @@ async function writeMetadata() { 'Streaming runtime operations:', ' node scripts/check-stream-runtime.mjs', ' node scripts/runtime-worker-drain.mjs status', + ' node scripts/runtime-worker-drain.mjs reconcile', + ' node scripts/runtime-worker-drain.mjs reconcile --apply', ' node scripts/runtime-worker-drain.mjs drain goosed-3', ' node scripts/runtime-worker-drain.mjs undrain goosed-3', ' node scripts/check-tool-runtime.mjs', diff --git a/scripts/runtime-worker-drain.mjs b/scripts/runtime-worker-drain.mjs index e9dbbff..1317723 100755 --- a/scripts/runtime-worker-drain.mjs +++ b/scripts/runtime-worker-drain.mjs @@ -16,7 +16,7 @@ function loadEnvFile(filePath) { } } -loadEnvFile(path.join(process.cwd(), '.env')); +loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env')); const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0'; const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime'; @@ -26,17 +26,24 @@ const configuredWorkers = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_ .filter(Boolean) .map((_, index) => `goosed-${index + 1}`); const action = process.argv[2] || 'status'; -const workerId = process.argv[3] || null; +const positionalArgs = process.argv.slice(3).filter((arg) => !arg.startsWith('--')); +const workerId = positionalArgs[0] || null; +const apply = process.argv.includes('--apply'); +const staleMsArg = process.argv.find((arg) => arg.startsWith('--stale-ms=')); +const staleMs = Math.max( + 60_000, + Number(staleMsArg?.split('=').at(1) || process.env.MEMIND_RUNTIME_STALE_STREAM_MS || 15 * 60 * 1000), +); function usage() { - console.error('Usage: node scripts/runtime-worker-drain.mjs [goosed-N]'); + console.error('Usage: node scripts/runtime-worker-drain.mjs [goosed-N] [--apply] [--stale-ms=900000]'); } function workerKey(id, field) { return [namespace, 'worker', id, field].join(':'); } -if (!['status', 'drain', 'undrain'].includes(action)) { +if (!['status', 'drain', 'undrain', 'reconcile'].includes(action)) { usage(); process.exit(2); } @@ -68,6 +75,7 @@ const workers = [ if (workerId && !workers.includes(workerId)) workers.push(workerId); const rows = []; +const now = Date.now(); for (const id of [...new Set(workers)].sort()) { const values = await client.mGet([ workerKey(id, 'active_streams'), @@ -77,16 +85,41 @@ for (const id of [...new Set(workers)].sort()) { workerKey(id, 'stream_error_count'), workerKey(id, 'last_stream_started_at'), workerKey(id, 'last_stream_ended_at'), + workerKey(id, 'last_stream_reconciled_at'), + workerKey(id, 'stream_reconcile_count'), ]); + const activeStreams = Number(values[0] || 0); + const lastStreamStartedAt = values[5] ? Number(values[5]) : null; + const lastStreamEndedAt = values[6] ? Number(values[6]) : null; + const staleAgeMs = lastStreamStartedAt ? now - lastStreamStartedAt : null; + const stale = Boolean( + activeStreams > 0 && + lastStreamStartedAt && + staleAgeMs > staleMs && + (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt), + ); + if (action === 'reconcile' && stale && apply) { + await client + .multi() + .set(workerKey(id, 'active_streams'), '0') + .set(workerKey(id, 'last_stream_reconciled_at'), String(now)) + .incr(workerKey(id, 'stream_reconcile_count')) + .exec(); + } rows.push({ id, - activeStreams: Number(values[0] || 0), + activeStreams, drain: /^(1|true|yes)$/i.test(String(values[1] || '')), streamOpenCount: Number(values[2] || 0), streamAbortCount: Number(values[3] || 0), streamErrorCount: Number(values[4] || 0), - lastStreamStartedAt: values[5] ? Number(values[5]) : null, - lastStreamEndedAt: values[6] ? Number(values[6]) : null, + lastStreamStartedAt, + lastStreamEndedAt, + lastStreamReconciledAt: values[7] ? Number(values[7]) : null, + streamReconcileCount: Number(values[8] || 0), + stale, + staleAgeMs, + reconciled: action === 'reconcile' && stale && apply, }); } @@ -97,5 +130,7 @@ console.log(JSON.stringify({ action, workerId, namespace, + dryRun: action === 'reconcile' ? !apply : undefined, + staleMs: action === 'reconcile' ? staleMs : undefined, workers: rows, }, null, 2));