diff --git a/package.json b/package.json index ea52a6cc..25593cae 100644 --- a/package.json +++ b/package.json @@ -13,9 +13,9 @@ "coverage": "npm run test:coverage", "server": "node server/server.mjs", "test:api": "node tests/api/auth-secret.test.mjs && node --env-file=tests/api/smoke.env tests/api/smoke.mjs && node tests/api/ratelimit.test.mjs && node tests/api/attachment-status.test.mjs && node tests/api/session-revocation.test.mjs && node tests/api/orchestrator-attribution.test.mjs && node tests/api/billing-checkout.test.mjs && node tests/api/billing-live-checkout.test.mjs && node tests/api/billing-effective-plan-status.test.mjs && node tests/api/stripe-webhook.test.mjs && node tests/api/stripe-reconciliation-dead-letter-recovery.test.mjs", - "test:unit": "node tests/unit/opencode-config.test.mjs && node tests/unit/changelog-release-notes.test.mjs && node tests/unit/analytics.test.mjs && node tests/unit/cpm.test.mjs && node tests/unit/baseline-compare.test.mjs && node tests/unit/workload.test.mjs && node tests/unit/cost-evm.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/dep-types.test.mjs && node tests/unit/weekly-report.test.mjs && node tests/unit/clearfolio.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/sprint-stats.test.mjs && node tests/unit/burndown.test.mjs && node tests/unit/pm-analysis.test.mjs && node tests/unit/cloud-sync-security.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/coverage-script-contract.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap-package-contract.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/stripe-reconciliation-worker.test.mjs && node tests/unit/stripe-reconciliation-worker-time-budget.test.mjs && node tests/unit/stripe-reconciliation-dead-letter-recovery.test.mjs && node tests/fuzz/stripeReconciliationRecovery.fuzz.mjs && node tests/unit/toast-accessibility.test.mjs", - "test:coverage": "c8 --all --include=app.js --include=cloud-sync.js --include=scripts/ci/static_coverage_evidence.mjs --include=server/attachment_status.mjs --include=server/app.mjs --include=server/application_routes.mjs --include=server/auth.mjs --include=server/billing.mjs --include=server/billing_checkout_attempt.mjs --include=server/billing_configuration.mjs --include=server/billing_status_response.mjs --include=server/clearfolio.mjs --include=server/orchestrator.mjs --include=server/stripe_webhook.mjs --include=server/stripe_webhook_event_ledger.mjs --include=server/stripe_subscription_provider.mjs --include=server/stripe_subscription_observation_ledger.mjs --include=server/stripe_subscription_current_projection.mjs --include=server/stripe_entitlement_policy.mjs --include=server/stripe_invoice_provider.mjs --include=server/stripe_invoice_observation_ledger.mjs --include=server/stripe_invoice_current_projection.mjs --include=server/stripe_entitlement_claim_ledger.mjs --include=server/stripe_billing_reconciliation.mjs --include=server/stripe_webhook_reconciliation_queue.mjs --include=server/stripe_checkout_identity_bootstrap.mjs --include=server/stripe_reconciliation_worker.mjs --include=server/stripe_reconciliation_recovery.mjs --include=server/stripe_reconciliation_recovery_routes.mjs --reporter=json --reporter=json-summary npm run test:coverage:cases", - "test:coverage:cases": "node tests/unit/coverage-script-contract.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/stripe-reconciliation-worker.test.mjs && node tests/unit/stripe-reconciliation-worker-time-budget.test.mjs && node tests/unit/stripe-reconciliation-dead-letter-recovery.test.mjs && node tests/fuzz/stripeReconciliationRecovery.fuzz.mjs && npm run test:api", + "test:unit": "node tests/unit/opencode-config.test.mjs && node tests/unit/changelog-release-notes.test.mjs && node tests/unit/analytics.test.mjs && node tests/unit/cpm.test.mjs && node tests/unit/baseline-compare.test.mjs && node tests/unit/workload.test.mjs && node tests/unit/cost-evm.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/dep-types.test.mjs && node tests/unit/weekly-report.test.mjs && node tests/unit/clearfolio.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/sprint-stats.test.mjs && node tests/unit/burndown.test.mjs && node tests/unit/pm-analysis.test.mjs && node tests/unit/cloud-sync-security.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/coverage-script-contract.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap-package-contract.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/stripe-reconciliation-worker.test.mjs && node tests/unit/stripe-reconciliation-worker-time-budget.test.mjs && node tests/unit/stripe-reconciliation-scheduler.test.mjs && node tests/unit/server-runtime-force-close.test.mjs && node tests/unit/stripe-reconciliation-dead-letter-recovery.test.mjs && node tests/fuzz/stripeReconciliationRecovery.fuzz.mjs && node tests/unit/toast-accessibility.test.mjs", + "test:coverage": "c8 --all --include=app.js --include=cloud-sync.js --include=scripts/ci/static_coverage_evidence.mjs --include=server/attachment_status.mjs --include=server/app.mjs --include=server/application_routes.mjs --include=server/auth.mjs --include=server/billing.mjs --include=server/billing_checkout_attempt.mjs --include=server/billing_configuration.mjs --include=server/billing_status_response.mjs --include=server/clearfolio.mjs --include=server/orchestrator.mjs --include=server/stripe_webhook.mjs --include=server/stripe_webhook_event_ledger.mjs --include=server/stripe_subscription_provider.mjs --include=server/stripe_subscription_observation_ledger.mjs --include=server/stripe_subscription_current_projection.mjs --include=server/stripe_entitlement_policy.mjs --include=server/stripe_invoice_provider.mjs --include=server/stripe_invoice_observation_ledger.mjs --include=server/stripe_invoice_current_projection.mjs --include=server/stripe_entitlement_claim_ledger.mjs --include=server/stripe_billing_reconciliation.mjs --include=server/stripe_webhook_reconciliation_queue.mjs --include=server/stripe_checkout_identity_bootstrap.mjs --include=server/stripe_reconciliation_worker.mjs --include=server/stripe_reconciliation_recovery.mjs --include=server/stripe_reconciliation_recovery_routes.mjs --include=server/stripe_reconciliation_scheduler.mjs --include=server/server_runtime.mjs --reporter=json --reporter=json-summary npm run test:coverage:cases", + "test:coverage:cases": "node tests/unit/coverage-script-contract.test.mjs && node tests/unit/attachment-status.test.mjs && node tests/unit/clearfolio-status-signal.test.mjs && node tests/unit/clearfolio-adapter-mock-hmac.test.mjs && node tests/unit/orchestrator.test.mjs && node tests/unit/orchestrator-coverage.test.mjs && node tests/unit/orchestrator-attribution.test.mjs && node tests/unit/msproject.test.mjs && node tests/unit/auth-password.test.mjs && node tests/unit/editor-unsaved.test.mjs && node tests/unit/static-coverage-evidence.test.mjs && node tests/unit/billing-configuration.test.mjs && node tests/unit/billing-effective-plan.test.mjs && node tests/unit/billing-effective-plan-bootstrap.test.mjs && node tests/unit/billing-status-package-contract.test.mjs && node tests/unit/billing-status-response.test.mjs && node tests/unit/billing-checkout-attempt.test.mjs && node tests/unit/billing-checkout-attempt-authority.test.mjs && node tests/unit/billing-checkout-reconciliation.test.mjs && node tests/unit/billing-checkout-reconciliation-authority.test.mjs && node tests/unit/billing-checkout.test.mjs && node tests/unit/billing-provider-boundary.test.mjs && node tests/unit/billing-checkout-review-regressions.test.mjs && node tests/unit/stripe-webhook-boundary.test.mjs && node tests/unit/stripe-webhook-event-ledger.test.mjs && node tests/unit/stripe-webhook-recorder-integration.test.mjs && node tests/unit/stripe-subscription-provider.test.mjs && node tests/unit/stripe-subscription-metadata-propagation.test.mjs && node tests/unit/stripe-subscription-observation-ledger.test.mjs && node tests/unit/stripe-subscription-current-projection.test.mjs && node tests/unit/stripe-entitlement-policy.test.mjs && node tests/unit/stripe-entitlement-policy-edge.test.mjs && node tests/unit/stripe-entitlement-policy-duplicate-claims.test.mjs && node tests/unit/stripe-invoice-provider.test.mjs && node tests/unit/stripe-invoice-provider-edge.test.mjs && node tests/unit/stripe-invoice-observation-ledger.test.mjs && node tests/unit/stripe-invoice-current-projection-package-contract.test.mjs && node tests/unit/stripe-invoice-current-projection.test.mjs && node tests/unit/stripe-entitlement-claim-package-contract.test.mjs && node tests/unit/stripe-entitlement-claim-ledger.test.mjs && node tests/unit/stripe-billing-authoritative-reconciliation.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue.test.mjs && node tests/unit/stripe-checkout-identity-bootstrap.test.mjs && node tests/unit/stripe-webhook-reconciliation-queue-integration.test.mjs && node tests/unit/stripe-reconciliation-worker.test.mjs && node tests/unit/stripe-reconciliation-worker-time-budget.test.mjs && node tests/unit/stripe-reconciliation-scheduler.test.mjs && node tests/unit/server-runtime-force-close.test.mjs && node tests/unit/stripe-reconciliation-dead-letter-recovery.test.mjs && node tests/fuzz/stripeReconciliationRecovery.fuzz.mjs && npm run test:api", "test:e2e": "playwright test", "test:e2e:headed": "playwright test --headed", "test:e2e:cloud": "playwright install chromium && playwright test tests/e2e/cloud.spec.js tests/e2e/toast-accessibility.spec.js", diff --git a/server/server.mjs b/server/server.mjs index c84c2e25..4f05dd88 100644 --- a/server/server.mjs +++ b/server/server.mjs @@ -1,7 +1,30 @@ +import process from 'node:process'; import { serve } from '@hono/node-server'; import { app } from './app.mjs'; +import { reconcileNextStripeBillingTrigger } from './db.mjs'; +import { bindScopeWeaveRuntime } from './server_runtime.mjs'; +import { + createStripeReconciliationScheduler, + stripeReconciliationPollIntervalMs, +} from './stripe_reconciliation_scheduler.mjs'; const port = Number(process.env.PORT) || 8787; -serve({ fetch: app.fetch, port }, (info) => { +const reconciliationScheduler = createStripeReconciliationScheduler({ + runOnce: reconcileNextStripeBillingTrigger, + intervalMs: stripeReconciliationPollIntervalMs( + process.env.SCOPEWEAVE_STRIPE_RECONCILIATION_POLL_MS, + ), + onFailure: (code) => console.error(code), +}); +const server = serve({ fetch: app.fetch, port }, (info) => { console.log(`ScopeWeave API listening on http://localhost:${info.port}`); }); + +bindScopeWeaveRuntime({ + server, + scheduler: reconciliationScheduler, + onShutdownFailure: (code) => { + console.error(code); + process.exitCode = 1; + }, +}); diff --git a/server/server_runtime.mjs b/server/server_runtime.mjs new file mode 100644 index 00000000..1b7a69a8 --- /dev/null +++ b/server/server_runtime.mjs @@ -0,0 +1,136 @@ +import process from 'node:process'; + +const SHUTDOWN_GRACE_MS = 10_000; + +function reportShutdownFailure(onShutdownFailure, code) { + try { + onShutdownFailure(code); + } catch { + // Shutdown must continue even when the operational telemetry sink is unavailable. + } +} + +/** + * Bind one HTTP server and one background scheduler to process termination semantics. + * + * The scheduler is stopped before the HTTP listener closes so a terminating process + * cannot schedule new provider work. The listener then gets a bounded ten-second + * graceful drain window. Node HTTP servers expose `closeAllConnections()`, which is + * used after that window so long-lived SSE responses cannot keep deployment shutdown + * open indefinitely. Test/minimal server adapters without that optional Node method + * retain the historical close-only behavior. + * + * Signal listeners are removed before closing the server, making repeated + * SIGINT/SIGTERM delivery and direct `shutdown()` calls idempotent. Only stable + * failure codes are exposed to the operational sink. + * + * @param {object} input runtime resources + * @param {{close:(callback:(error?:Error)=>void)=>void,closeAllConnections?:()=>void}} input.server active HTTP server + * @param {{start:()=>boolean,stop:()=>boolean}} input.scheduler reconciliation scheduler + * @param {object} [input.signalTarget=process] EventEmitter-like process signal target + * @param {(code:string)=>void} [input.onShutdownFailure] bounded failure sink + * @param {(callback:Function,delayMs:number)=>unknown} [input.scheduleShutdown=setTimeout] shutdown watchdog timer arm + * @param {(timer:unknown)=>void} [input.cancelShutdown=clearTimeout] shutdown watchdog timer cancellation + * @returns {Readonly<{shutdown:()=>boolean}>} idempotent shutdown control + */ +export function bindScopeWeaveRuntime({ + server, + scheduler, + signalTarget = process, + onShutdownFailure = () => {}, + scheduleShutdown = setTimeout, + cancelShutdown = clearTimeout, +} = {}) { + if (!server || typeof server.close !== 'function') { + throw new TypeError('server must provide close()'); + } + if (!scheduler || typeof scheduler.start !== 'function' || typeof scheduler.stop !== 'function') { + throw new TypeError('scheduler must provide start()/stop()'); + } + if (!signalTarget + || typeof signalTarget.once !== 'function' + || typeof signalTarget.off !== 'function') { + throw new TypeError('signalTarget must provide once()/off()'); + } + if (typeof onShutdownFailure !== 'function') { + throw new TypeError('onShutdownFailure must be a function'); + } + if (typeof scheduleShutdown !== 'function') { + throw new TypeError('scheduleShutdown must be a function'); + } + if (typeof cancelShutdown !== 'function') { + throw new TypeError('cancelShutdown must be a function'); + } + + const forceCloseConnections = typeof server.closeAllConnections === 'function' + ? () => server.closeAllConnections() + : () => {}; + let shuttingDown = false; + let serverClosed = false; + let shutdownTimer = null; + + function detachSignals() { + signalTarget.off('SIGINT', handleSignal); + signalTarget.off('SIGTERM', handleSignal); + } + + function cancelShutdownWatchdog() { + const pendingTimer = shutdownTimer; + shutdownTimer = null; + if (pendingTimer !== null) cancelShutdown(pendingTimer); + } + + function forceCloseAfterGrace() { + shutdownTimer = null; + try { + forceCloseConnections(); + } catch { + reportShutdownFailure(onShutdownFailure, 'scopeweave_server_shutdown_failed'); + } + } + + function shutdown() { + if (shuttingDown) return false; + shuttingDown = true; + detachSignals(); + + try { + scheduler.stop(); + } catch { + reportShutdownFailure(onShutdownFailure, 'scopeweave_scheduler_shutdown_failed'); + } + + try { + server.close((error) => { + serverClosed = true; + cancelShutdownWatchdog(); + if (error) { + reportShutdownFailure(onShutdownFailure, 'scopeweave_server_shutdown_failed'); + } + }); + if (!serverClosed) { + shutdownTimer = scheduleShutdown(forceCloseAfterGrace, SHUTDOWN_GRACE_MS); + } + } catch { + cancelShutdownWatchdog(); + reportShutdownFailure(onShutdownFailure, 'scopeweave_server_shutdown_failed'); + } + return true; + } + + function handleSignal() { + shutdown(); + } + + signalTarget.once('SIGINT', handleSignal); + signalTarget.once('SIGTERM', handleSignal); + try { + scheduler.start(); + } catch (error) { + detachSignals(); + shuttingDown = true; + throw error; + } + + return Object.freeze({ shutdown }); +} diff --git a/server/stripe_reconciliation_scheduler.mjs b/server/stripe_reconciliation_scheduler.mjs new file mode 100644 index 00000000..efddb85f --- /dev/null +++ b/server/stripe_reconciliation_scheduler.mjs @@ -0,0 +1,145 @@ +const DEFAULT_POLL_INTERVAL_MS = 1_000; +const MIN_POLL_INTERVAL_MS = 250; +const MAX_POLL_INTERVAL_MS = 60_000; +const CANONICAL_INTEGER_PATTERN = /^(?:0|[1-9][0-9]*)$/u; + +/** Stable fail-closed error for Stripe reconciliation scheduler configuration. */ +export class StripeReconciliationSchedulerError extends Error { + /** + * Create one sanitized scheduler failure. + * @param {string} code stable machine-readable failure code + */ + constructor(code) { + super(code); + this.name = 'StripeReconciliationSchedulerError'; + this.code = code; + } +} + +function schedulerError(code) { + return new StripeReconciliationSchedulerError(code); +} + +function reportFailure(onFailure, code) { + try { + onFailure(code); + } catch { + // A broken telemetry sink must never multiply or terminate the reconciliation loop. + } +} + +/** + * Parse the operator-owned Stripe reconciliation poll interval. + * + * Only a canonical base-10 millisecond integer is accepted. The lower bound prevents + * a configuration typo from creating a provider hot loop; the upper bound keeps a + * healthy worker from leaving verified billing events unattended for more than one + * minute between passes. + * + * @param {string|number|undefined} value configured interval in milliseconds + * @returns {number} validated interval between 250 ms and 60 seconds + */ +export function stripeReconciliationPollIntervalMs(value) { + if (value === undefined) return DEFAULT_POLL_INTERVAL_MS; + if ( + (typeof value !== 'string' && typeof value !== 'number') + || (typeof value === 'string' && !CANONICAL_INTEGER_PATTERN.test(value)) + ) { + throw schedulerError('stripe_reconciliation_scheduler_interval_invalid'); + } + const intervalMs = Number(value); + if ( + !Number.isSafeInteger(intervalMs) + || intervalMs < MIN_POLL_INTERVAL_MS + || intervalMs > MAX_POLL_INTERVAL_MS + ) { + throw schedulerError('stripe_reconciliation_scheduler_interval_invalid'); + } + return intervalMs; +} + +/** + * Create a single-flight scheduler for durable Stripe reconciliation work. + * + * The scheduler arms its next timer only after the current reconciliation promise has + * settled, so provider latency can never create overlapping poll iterations. Runtime + * failures are collapsed to a stable non-secret code and polling continues. `stop()` + * flips the running state before cancelling a waiting timer, which also prevents an + * already in-flight iteration from resurrecting the loop during graceful shutdown. + * + * @param {object} input scheduler ports and configuration + * @param {() => Promise|unknown} input.runOnce consume at most one durable job + * @param {number} [input.intervalMs=1000] validated delay between completed passes + * @param {(callback:Function,delayMs:number)=>unknown} [input.schedule=setTimeout] timer arm + * @param {(timer:unknown)=>void} [input.cancel=clearTimeout] timer cancellation + * @param {(code:string)=>void} [input.onFailure] bounded operational failure sink + * @returns {Readonly<{start:()=>boolean,stop:()=>boolean}>} scheduler lifecycle controls + */ +export function createStripeReconciliationScheduler({ + runOnce, + intervalMs = DEFAULT_POLL_INTERVAL_MS, + schedule = setTimeout, + cancel = clearTimeout, + onFailure = () => {}, +} = {}) { + if (typeof runOnce !== 'function') throw new TypeError('runOnce must be a function'); + if (typeof schedule !== 'function') throw new TypeError('schedule must be a function'); + if (typeof cancel !== 'function') throw new TypeError('cancel must be a function'); + if (typeof onFailure !== 'function') throw new TypeError('onFailure must be a function'); + const delayMs = stripeReconciliationPollIntervalMs(intervalMs); + + let running = false; + let timer = null; + + function arm(delay) { + try { + timer = schedule(tick, delay); + } catch { + timer = null; + running = false; + throw schedulerError('stripe_reconciliation_scheduler_timer_failed'); + } + } + + async function tick() { + timer = null; + if (!running) return; + try { + await runOnce(); + } catch { + reportFailure(onFailure, 'stripe_reconciliation_scheduler_iteration_failed'); + } + if (!running) return; + try { + arm(delayMs); + } catch { + reportFailure(onFailure, 'stripe_reconciliation_scheduler_timer_failed'); + } + } + + return Object.freeze({ + /** Start one immediate asynchronous pass. Repeated starts are side-effect free. */ + start() { + if (running) return false; + running = true; + arm(0); + return true; + }, + + /** Stop future passes. An in-flight pass may finish but cannot schedule another. */ + stop() { + if (!running) return false; + running = false; + const pendingTimer = timer; + timer = null; + if (pendingTimer !== null) { + try { + cancel(pendingTimer); + } catch { + reportFailure(onFailure, 'stripe_reconciliation_scheduler_timer_cancel_failed'); + } + } + return true; + }, + }); +} diff --git a/tests/unit/server-runtime-force-close.test.mjs b/tests/unit/server-runtime-force-close.test.mjs new file mode 100644 index 00000000..d907c47d --- /dev/null +++ b/tests/unit/server-runtime-force-close.test.mjs @@ -0,0 +1,35 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { test } from 'node:test'; + +import { bindScopeWeaveRuntime } from '../../server/server_runtime.mjs'; + +test('forced-close failures are sanitized when the graceful shutdown window expires', () => { + const signalTarget = new EventEmitter(); + const failures = []; + const watchdogs = []; + const runtime = bindScopeWeaveRuntime({ + server: { + close() { + // Keep one response open so the bounded shutdown watchdog must fire. + }, + closeAllConnections() { + throw new Error('socket /secret/runtime.sock must not escape'); + }, + }, + scheduler: { start: () => true, stop: () => true }, + signalTarget, + onShutdownFailure: (code) => failures.push(code), + scheduleShutdown(callback, delayMs) { + watchdogs.push({ callback, delayMs }); + return 1; + }, + cancelShutdown: () => {}, + }); + + assert.equal(runtime.shutdown(), true); + assert.equal(watchdogs.length, 1); + assert.equal(watchdogs[0].delayMs, 10_000); + assert.doesNotThrow(() => watchdogs[0].callback()); + assert.deepEqual(failures, ['scopeweave_server_shutdown_failed']); +}); diff --git a/tests/unit/stripe-reconciliation-scheduler.test.mjs b/tests/unit/stripe-reconciliation-scheduler.test.mjs new file mode 100644 index 00000000..99833b29 --- /dev/null +++ b/tests/unit/stripe-reconciliation-scheduler.test.mjs @@ -0,0 +1,477 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { test } from 'node:test'; + +import { + createStripeReconciliationScheduler, + stripeReconciliationPollIntervalMs, +} from '../../server/stripe_reconciliation_scheduler.mjs'; +import { bindScopeWeaveRuntime } from '../../server/server_runtime.mjs'; + +function createFakeTimers() { + let nextId = 0; + const pending = new Map(); + const cancelled = []; + + function schedule(callback, delayMs) { + const id = ++nextId; + pending.set(id, { callback, delayMs }); + return id; + } + + function cancel(id) { + cancelled.push(id); + pending.delete(id); + } + + async function runNext() { + const next = pending.entries().next().value; + assert.ok(next, 'expected one pending scheduler callback'); + const [id, entry] = next; + pending.delete(id); + await entry.callback(); + return entry.delayMs; + } + + function beginNext() { + const next = pending.entries().next().value; + assert.ok(next, 'expected one pending scheduler callback'); + const [id, entry] = next; + pending.delete(id); + return { delayMs: entry.delayMs, promise: entry.callback(), callback: entry.callback }; + } + + return { schedule, cancel, runNext, beginNext, pending, cancelled }; +} + +test('poll interval parsing is bounded and rejects ambiguous configuration', () => { + assert.equal(stripeReconciliationPollIntervalMs(undefined), 1_000); + assert.equal(stripeReconciliationPollIntervalMs(250), 250); + assert.equal(stripeReconciliationPollIntervalMs('250'), 250); + assert.equal(stripeReconciliationPollIntervalMs('60000'), 60_000); + + for (const invalid of [ + '', ' 250', '250 ', '1e3', '0', '99', '60001', '-1', 'NaN', null, {}, 250.5, + ]) { + assert.throws( + () => stripeReconciliationPollIntervalMs(invalid), + (error) => error?.code === 'stripe_reconciliation_scheduler_interval_invalid', + `expected ${JSON.stringify(invalid)} to fail closed`, + ); + } +}); + +test('scheduler validates lifecycle ports before arming provider work', () => { + assert.throws(() => createStripeReconciliationScheduler(), /runOnce must be a function/); + assert.throws( + () => createStripeReconciliationScheduler({ runOnce: async () => {}, schedule: null }), + /schedule must be a function/, + ); + assert.throws( + () => createStripeReconciliationScheduler({ runOnce: async () => {}, cancel: null }), + /cancel must be a function/, + ); + assert.throws( + () => createStripeReconciliationScheduler({ runOnce: async () => {}, onFailure: null }), + /onFailure must be a function/, + ); + + const realTimerScheduler = createStripeReconciliationScheduler({ runOnce: async () => {} }); + assert.equal(realTimerScheduler.start(), true); + assert.equal(realTimerScheduler.stop(), true); +}); + +test('scheduler never overlaps reconciliation and stop during an in-flight run prevents rescheduling', async () => { + const timers = createFakeTimers(); + let resolveRun; + let calls = 0; + const runGate = new Promise((resolve) => { + resolveRun = resolve; + }); + const scheduler = createStripeReconciliationScheduler({ + runOnce: async () => { + calls += 1; + await runGate; + return { status: 'idle' }; + }, + intervalMs: 250, + schedule: timers.schedule, + cancel: timers.cancel, + }); + + assert.equal(scheduler.start(), true); + assert.equal(scheduler.start(), false, 'start is idempotent'); + assert.equal(timers.pending.size, 1); + const first = timers.beginNext(); + assert.equal(first.delayMs, 0, 'startup schedules an immediate first pass without blocking boot'); + await Promise.resolve(); + assert.equal(calls, 1); + assert.equal(timers.pending.size, 0, 'no second timer exists while reconciliation is in flight'); + + assert.equal(scheduler.stop(), true); + assert.equal(scheduler.stop(), false, 'stop is idempotent'); + resolveRun(); + await first.promise; + assert.equal(timers.pending.size, 0, 'an in-flight completion cannot resurrect a stopped scheduler'); +}); + +test('a stale timer callback observed after stop cannot restart reconciliation', async () => { + let callback; + const failures = []; + const scheduler = createStripeReconciliationScheduler({ + runOnce: async () => { + throw new Error('must not run after stop'); + }, + intervalMs: 250, + schedule: (scheduled) => { + callback = scheduled; + return 1; + }, + cancel: () => {}, + onFailure: (code) => failures.push(code), + }); + + scheduler.start(); + scheduler.stop(); + await callback(); + assert.deepEqual(failures, []); +}); + +test('scheduler sanitizes iteration failures and continues with one bounded delayed retry', async () => { + const timers = createFakeTimers(); + const failures = []; + let calls = 0; + const scheduler = createStripeReconciliationScheduler({ + runOnce: async () => { + calls += 1; + if (calls === 1) throw new Error('provider failed with sk_live_must_not_escape'); + return { status: 'idle' }; + }, + intervalMs: 500, + schedule: timers.schedule, + cancel: timers.cancel, + onFailure: (...args) => failures.push(args), + }); + + scheduler.start(); + assert.equal(await timers.runNext(), 0); + assert.deepEqual(failures, [['stripe_reconciliation_scheduler_iteration_failed']]); + assert.equal(timers.pending.size, 1); + const retryDelay = await timers.runNext(); + assert.equal(retryDelay, 500); + assert.equal(calls, 2); + assert.equal(timers.pending.size, 1, 'successful idle passes keep bounded polling alive'); + scheduler.stop(); + assert.equal(timers.pending.size, 0); +}); + +test('failure reporting cannot crash or multiply the scheduler loop', async () => { + const timers = createFakeTimers(); + const scheduler = createStripeReconciliationScheduler({ + runOnce: async () => { + throw new Error('causal runtime failure'); + }, + intervalMs: 250, + schedule: timers.schedule, + cancel: timers.cancel, + onFailure: () => { + throw new Error('broken telemetry sink'); + }, + }); + + scheduler.start(); + await timers.runNext(); + assert.equal(timers.pending.size, 1); + assert.equal(await timers.runNext(), 250); + assert.equal(timers.pending.size, 1); + scheduler.stop(); +}); + +test('scheduler fails closed if its timer cannot be armed or re-armed', async () => { + const startupScheduler = createStripeReconciliationScheduler({ + runOnce: async () => {}, + intervalMs: 250, + schedule: () => { + throw new Error('timer allocation failure'); + }, + cancel: () => {}, + }); + assert.throws( + () => startupScheduler.start(), + (error) => error?.code === 'stripe_reconciliation_scheduler_timer_failed', + ); + assert.equal(startupScheduler.stop(), false); + + let callback; + let armCount = 0; + const failures = []; + const retryScheduler = createStripeReconciliationScheduler({ + runOnce: async () => ({ status: 'idle' }), + intervalMs: 250, + schedule: (scheduled) => { + armCount += 1; + if (armCount > 1) throw new Error('timer re-arm failure'); + callback = scheduled; + return 1; + }, + cancel: () => {}, + onFailure: (code) => failures.push(code), + }); + retryScheduler.start(); + await callback(); + assert.deepEqual(failures, ['stripe_reconciliation_scheduler_timer_failed']); + assert.equal(retryScheduler.stop(), false, 'timer failure terminates the loop instead of hot-spinning'); +}); + +test('stopping a waiting scheduler cancels the exact pending timer', () => { + const timers = createFakeTimers(); + const scheduler = createStripeReconciliationScheduler({ + runOnce: async () => ({ status: 'idle' }), + intervalMs: 1_000, + schedule: timers.schedule, + cancel: timers.cancel, + }); + + scheduler.start(); + const timerId = [...timers.pending.keys()][0]; + scheduler.stop(); + assert.deepEqual(timers.cancelled, [timerId]); + assert.equal(timers.pending.size, 0); +}); + +test('timer cancellation failure is sanitized after the loop is already stopped', () => { + const failures = []; + const scheduler = createStripeReconciliationScheduler({ + runOnce: async () => {}, + intervalMs: 250, + schedule: () => 7, + cancel: () => { + throw new Error('secret timer implementation detail'); + }, + onFailure: (code) => failures.push(code), + }); + scheduler.start(); + assert.equal(scheduler.stop(), true); + assert.deepEqual(failures, ['stripe_reconciliation_scheduler_timer_cancel_failed']); + assert.equal(scheduler.stop(), false); +}); + +test('runtime binding validates server, scheduler, signal, telemetry, and shutdown-timer ports', () => { + const signalTarget = new EventEmitter(); + const server = { close: () => {} }; + const scheduler = { start: () => true, stop: () => true }; + + assert.throws(() => bindScopeWeaveRuntime(), /server must provide close/); + assert.throws(() => bindScopeWeaveRuntime({ server: {}, scheduler }), /server must provide close/); + assert.throws(() => bindScopeWeaveRuntime({ server, scheduler: {} }), /scheduler must provide/); + assert.throws( + () => bindScopeWeaveRuntime({ server, scheduler: { start: () => true } }), + /scheduler must provide/, + ); + assert.throws( + () => bindScopeWeaveRuntime({ server, scheduler, signalTarget: {} }), + /signalTarget must provide/, + ); + assert.throws( + () => bindScopeWeaveRuntime({ + server, + scheduler, + signalTarget: { once: () => {} }, + }), + /signalTarget must provide/, + ); + assert.throws( + () => bindScopeWeaveRuntime({ server, scheduler, signalTarget, onShutdownFailure: null }), + /onShutdownFailure must be a function/, + ); + assert.throws( + () => bindScopeWeaveRuntime({ server, scheduler, signalTarget, scheduleShutdown: null }), + /scheduleShutdown must be a function/, + ); + assert.throws( + () => bindScopeWeaveRuntime({ server, scheduler, signalTarget, cancelShutdown: null }), + /cancelShutdown must be a function/, + ); +}); + +test('runtime binding starts reconciliation and drains it before closing on termination signals', () => { + const order = []; + const signalTarget = new EventEmitter(); + const scheduler = { + start() { + order.push('scheduler:start'); + return true; + }, + stop() { + order.push('scheduler:stop'); + return true; + }, + }; + const server = { + close(callback) { + order.push('server:close'); + callback(); + }, + }; + const failures = []; + + const runtime = bindScopeWeaveRuntime({ + server, + scheduler, + signalTarget, + onShutdownFailure: (...args) => failures.push(args), + }); + + assert.deepEqual(order, ['scheduler:start']); + signalTarget.emit('SIGTERM', 'SIGTERM'); + assert.deepEqual(order, ['scheduler:start', 'scheduler:stop', 'server:close']); + assert.deepEqual(failures, []); + assert.equal(runtime.shutdown(), false, 'shutdown is idempotent after the first signal'); + signalTarget.emit('SIGINT', 'SIGINT'); + assert.deepEqual(order, ['scheduler:start', 'scheduler:stop', 'server:close']); +}); + +test('runtime force-closes long-lived connections after the bounded graceful drain window', async () => { + const shutdownTimers = createFakeTimers(); + const order = []; + const signalTarget = new EventEmitter(); + let closeCallback; + const server = { + close(callback) { + order.push('server:close'); + closeCallback = callback; + }, + closeAllConnections() { + order.push('server:force-close'); + }, + }; + const scheduler = { + start() { + order.push('scheduler:start'); + return true; + }, + stop() { + order.push('scheduler:stop'); + return true; + }, + }; + const failures = []; + + const runtime = bindScopeWeaveRuntime({ + server, + scheduler, + signalTarget, + onShutdownFailure: (code) => failures.push(code), + scheduleShutdown: shutdownTimers.schedule, + cancelShutdown: shutdownTimers.cancel, + }); + + assert.equal(runtime.shutdown(), true); + assert.deepEqual(order, ['scheduler:start', 'scheduler:stop', 'server:close']); + assert.equal(shutdownTimers.pending.size, 1, 'active responses need one bounded shutdown watchdog'); + assert.equal(await shutdownTimers.runNext(), 10_000); + assert.deepEqual(order, ['scheduler:start', 'scheduler:stop', 'server:close', 'server:force-close']); + closeCallback(); + assert.deepEqual(failures, []); +}); + +test('runtime cancels the forced-close watchdog when graceful drain completes first', () => { + const shutdownTimers = createFakeTimers(); + const signalTarget = new EventEmitter(); + let closeCallback; + let forceCloseCalls = 0; + const runtime = bindScopeWeaveRuntime({ + server: { + close(callback) { + closeCallback = callback; + }, + closeAllConnections() { + forceCloseCalls += 1; + }, + }, + scheduler: { start: () => true, stop: () => true }, + signalTarget, + scheduleShutdown: shutdownTimers.schedule, + cancelShutdown: shutdownTimers.cancel, + }); + + assert.equal(runtime.shutdown(), true); + const watchdogId = [...shutdownTimers.pending.keys()][0]; + assert.ok(watchdogId, 'graceful shutdown must arm one bounded watchdog'); + closeCallback(); + assert.deepEqual(shutdownTimers.cancelled, [watchdogId]); + assert.equal(shutdownTimers.pending.size, 0); + assert.equal(forceCloseCalls, 0, 'clean drain must not terminate already-finished connections'); +}); + +test('runtime binding exposes only stable scheduler and server shutdown failure codes', () => { + const signalTarget = new EventEmitter(); + const failures = []; + const runtime = bindScopeWeaveRuntime({ + server: { + close(callback) { + callback(new Error('socket path /secret/internal.sock failed')); + }, + }, + scheduler: { + start: () => true, + stop() { + throw new Error('provider state must not escape'); + }, + }, + signalTarget, + onShutdownFailure: (...args) => failures.push(args), + }); + + assert.equal(runtime.shutdown(), true); + assert.deepEqual(failures, [ + ['scopeweave_scheduler_shutdown_failed'], + ['scopeweave_server_shutdown_failed'], + ]); +}); + +test('runtime shutdown survives thrown close and broken failure telemetry', () => { + const signalTarget = new EventEmitter(); + const runtime = bindScopeWeaveRuntime({ + server: { + close() { + throw new Error('sensitive close failure'); + }, + }, + scheduler: { start: () => true, stop: () => true }, + signalTarget, + onShutdownFailure: () => { + throw new Error('telemetry unavailable'); + }, + }); + + assert.equal(runtime.shutdown(), true); +}); + +test('runtime detaches termination handlers when scheduler startup fails', () => { + const signalTarget = new EventEmitter(); + const startupError = new Error('scheduler failed to start'); + assert.throws( + () => bindScopeWeaveRuntime({ + server: { close: () => {} }, + scheduler: { + start() { + throw startupError; + }, + stop: () => true, + }, + signalTarget, + }), + (error) => error === startupError, + ); + assert.equal(signalTarget.listenerCount('SIGINT'), 0); + assert.equal(signalTarget.listenerCount('SIGTERM'), 0); +}); + +test('runtime defaults can bind and immediately unbind the real process signal target', () => { + const runtime = bindScopeWeaveRuntime({ + server: { close: (callback) => callback() }, + scheduler: { start: () => true, stop: () => true }, + }); + assert.equal(runtime.shutdown(), true); +}); \ No newline at end of file