Production Best Practices & Operational Runbook

-- Checkpoints with Optimistic Concurrency Control
CREATE TABLE IF NOT EXISTS siglo_checkpoints (
    stack_id VARCHAR(255) PRIMARY KEY,
    namespace VARCHAR(255),
    state JSONB NOT NULL,
    version BIGINT NOT NULL DEFAULT 1,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_siglo_checkpoints_ns ON siglo_checkpoints(namespace);

-- Durable wait records with row-level locking indexes
CREATE TABLE IF NOT EXISTS siglo_wait_records (
    correlation_key VARCHAR(255) PRIMARY KEY,
    stack_id VARCHAR(255) NOT NULL,
    namespace VARCHAR(255),
    event_name VARCHAR(255),
    status VARCHAR(50) NOT NULL DEFAULT 'WAITING',
    continuation_step_index INT NOT NULL DEFAULT 0,
    version VARCHAR(50),
    data JSONB,
    locked_by VARCHAR(255),
    locked_at TIMESTAMPTZ,
    resume_payload JSONB,
    resumed_by VARCHAR(255),
    resumed_at TIMESTAMPTZ,
    expires_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_siglo_wait_records_due ON siglo_wait_records(status, expires_at);

-- Idempotency store with settlement status
CREATE TABLE IF NOT EXISTS siglo_idempotency_records (
    key VARCHAR(255) PRIMARY KEY,
    status VARCHAR(50) NOT NULL,
    result JSONB,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    expires_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_siglo_idempotency_expires ON siglo_idempotency_records(expires_at);

import { flow, OptimisticLockException } from '@siglo/core';

async function runWithOCCRetry<T>(workflow: any, input: any, maxAttempts = 3): Promise<T> {
    for (let attempt = 1; attempt <= maxAttempts; attempt++) {
        try {
            return await workflow.run(input, { durable: true });
        } catch (error) {
            if (error instanceof OptimisticLockException && attempt < maxAttempts) {
                const backoffMs = Math.floor(Math.random() * 50 * Math.pow(2, attempt));
                await new Promise(res => setTimeout(res, backoffMs));
                continue;
            }
            throw error;
        }
    }
    throw new Error('OCC retry exhausted');
}

import { PostgresWaitPoller } from '@siglo/scheduler';
import { dbPool } from './db';
import { workflowRegistry } from './workflows';

const poller = new PostgresWaitPoller({
    client: dbPool,
    pollIntervalMs: 1000,
    batchSize: 50,
    // lockTimeoutMs must be >= 3x the max expected duration of steps after wait
    lockTimeoutMs: 60000,
    flowLookup: (version, stackId) => workflowRegistry.get(stackId),
    onResumed: rec => console.info(`Workflow ${rec.stackId} resumed successfully`),
    onFailed: (rec, err) => console.error(`Resumption failed for ${rec.stackId}:`, err),
});

// Start poller loop
poller.start();

// Graceful container shutdown (Kubernetes SIGTERM)
process.on('SIGTERM', async () => {
    console.info('Stopping scheduler poller...');
    await poller.stop();
    await dbPool.end();
    process.exit(0);
});

import http from 'http';
import { createSchedulerWebhookHandler } from '@siglo/scheduler';
import { workflowRegistry } from './workflows';

const webhookHandler = createSchedulerWebhookHandler({
    secretToken: process.env.SCHEDULER_SECRET_TOKEN,
    flowLookup: (workflowId, stackId) => workflowRegistry.get(stackId),
});

const server = http.createServer(webhookHandler);
server.listen(8080);

import { Step } from '@siglo/core';
import { PostgresIdempotencyStore } from '@siglo/postgres';

const idempotencyStore = new PostgresIdempotencyStore({ client: dbPool });

export const chargeCardStep = Step.create('chargeCard', async (ctx: any) => {
    return await paymentGateway.charge({ amount: ctx.amount, customerId: ctx.customerId });
}).idempotent(ctx => `charge:${ctx.orderId}`, {
    store: idempotencyStore,
    ttlMs: 86400000, // 24-hour claim retention
    settleTimeoutMs: 3000, // Spin-wait up to 3000ms for in-flight request to settle
    pollIntervalMs: 50, // Check state every 50ms
});

import { flow, Step } from '@siglo/core';

const partnerStep = Step.create('callPartnerWebhook', async (ctx: any) => {
    const res = await fetch(ctx.webhookUrl, { method: 'POST', body: JSON.stringify(ctx.payload) });
    return await res.json();
}).allowNetwork(['api.partner.com', 'hooks.slack.com']);

export const secureWebhookFlow = flow(partnerStep).networkPolicy({
    defaultPolicy: 'deny',
    allowedHosts: ['api.partner.com', 'hooks.slack.com'],
    allowLocalhost: false,
});

flowInstance.networkPolicy({
    allowUnixSockets: ['/var/run/custom-app.sock'],
});

import { Step, registerCompensationHandler } from '@siglo/core';

// 1. Register named compensations once at application startup
registerCompensationHandler('billing:refundCharge', async (output, input, ctx) => {
    await stripe.refunds.create({ charge: ctx.chargeId });
});

registerCompensationHandler('inventory:releaseHold', async (output, input, ctx) => {
    await inventoryService.release(ctx.reservationId);
});

// 2. Reference named compensations in your steps
export const reserveInventoryStep = Step.create('reserveInventory', async (ctx: any) => {
    const res = await inventoryService.reserve(ctx.itemId, ctx.quantity);
    return { reservationId: res.id };
}).compensate('inventory:releaseHold');

import { SQSClient, SendMessageCommand } from '@aws-sdk/client-sqs';

const sqs = new SQSClient({ region: 'us-east-1' });

const deadLetterQueue = {
    send: async (entry: any) => {
        await sqs.send(
            new SendMessageCommand({
                QueueUrl: process.env.COMPENSATION_DLQ_URL,
                MessageBody: JSON.stringify(entry),
            }),
        );
    },
};

// Execute best-effort saga rollback with DLQ sink:
await orderFlow.compensate('stack-id-to-rollback', {
    mode: 'best-effort',
    deadLetterQueue,
});

npm exec --package=@siglo/cli -- siglo reconcile --stale-threshold=10m --action=compensate

import { createMonitorServer } from '@siglo/monitor';
import { z } from 'zod';
import { orderFlow } from './workflows';

// Enforce strict schema on workflow resume payloads (read by @siglo/monitor)
(orderFlow as any).resumeSchema = z
    .object({
        approved: z.boolean(),
        approverId: z.string().uuid(),
        comments: z.string().max(500).optional(),
    })
    .strict(); // .strict() rejects unknown keys with validation errors

const monitorServer = await createMonitorServer({
    port: 4000,
    workflows: {
        orderFlow,
    },
    // Disable remote resumptions in production (read-only monitoring)
    allowRemoteResume: process.env.NODE_ENV !== 'production',
    auth: {
        apiKey: process.env.MONITOR_API_KEY,
        adminKey: process.env.MONITOR_ADMIN_KEY,
    },
    security: {
        // Redact PII and secrets in SSE, WebSockets, and GET endpoints
        redactedKeys: ['password', 'secret', 'token', 'apiKey', 'creditCard', 'ssn', 'authorization'],
    },
    cors: {
        origin: ['https://internal-devtools.yourdomain.com'],
    },
});