​

Accounting Job Queue System

​

Architecture

​

Components

1. Queue Table (accounting_work_queue): PostgreSQL table storing job state
2. Queue Functions (@cona/core): Functions to enqueue and manage jobs
3. Temporal Activities: Fetch and batch process jobs
4. Temporal Workflow: Continuous polling loop that processes jobs
5. Temporal Worker: Dedicated worker pool for the queue

​

Job States

• pending: Job is ready to be processed
• leased: Job has been claimed by a worker
• completed: Job successfully processed
• failed: Job failed after max retries
• cancelled: Job cancelled (e.g., document deleted)

​

Usage

​

Enqueuing Jobs

import { enqueueAccountingJob } from "@cona/core";

const result = await enqueueAccountingJob({
  organizationId: "org_123",
  documentId: "doc_456",
  versionHint: BigInt(1),
  readyAt: new Date(), // Optional, defaults to now
});

​

Document Readiness

await tx.documents.update({
  where: { id: documentId },
  data: {
    accounting_ready_at: new Date(),
    accounting_ready_version: { increment: 1 },
  },
});

​

Retry Behavior

• Attempt 1: 5 minutes
• Attempt 2: 15 minutes (5 × 3¹)
• Attempt 3: 45 minutes (5 × 3²)
• Attempt 4: 2 hours 15 minutes (5 × 3³)
• Attempt 5: 6 hours (5 × 3⁴, capped)
• Attempt 6+: 6 hours (capped)

​

Performance

​

Batch Processing

• Batch size: 50 jobs per poll
• Concurrency: 3 jobs processed in parallel
• Throughput: ~22 jobs/minute average, ~50 jobs/minute peak

​

Adaptive Polling

• Jobs found: Immediately poll again (zero sleep, minimal latency)
• No jobs found: Sleep 10 seconds before next poll (avoid DB load)

​

Monitoring

​

Key Metrics

• Pending Jobs: Count of jobs in pending state
• Processing Time: Time from ready_at to completion
• Failure Rate: Percentage of jobs reaching failed state
• Stale Jobs: Jobs in leased state > 15 minutes

​

SQL Queries

-- Pending jobs count
SELECT COUNT(*) FROM accounting_work_queue WHERE state = 'pending';

-- Failed jobs requiring attention
SELECT * FROM accounting_work_queue
WHERE state = 'failed'
ORDER BY updated_at DESC;

-- Stale leased jobs
SELECT * FROM accounting_work_queue
WHERE state = 'leased'
  AND last_attempted_at < NOW() - INTERVAL '15 minutes';

-- Processing stats
SELECT state, COUNT(*) FROM accounting_work_queue GROUP BY state;

​

Configuration

​

Environment Variables

• ACCOUNTING_QUEUE_BATCH_SIZE: Number of jobs to fetch per poll (default: 50)
• ACCOUNTING_QUEUE_POLL_INTERVAL_SECONDS: Seconds between polls when idle (default: 10)
• ACCOUNTING_QUEUE_RETRY_BASE_SECONDS: Base retry delay in seconds (default: 300)

​

Worker Configuration

const accountingQueueWorker = await Worker.create({
  taskQueue: ACCOUNTING_QUEUE_TASK_QUEUE,
  maxConcurrentActivityTaskExecutions: 3,
  maxConcurrentWorkflowTaskExecutions: 5,
  // ...
});

​

Error Handling

​

Transient Errors

​

Document State Errors

• Document deleted: Job cancelled immediately
• Document not ready (accounting_ready_at is null): Job rescheduled for 30 seconds
• Version mismatch: Job rescheduled with updated version (5 second delay)

​

Permanent Errors

1. Investigation of the last_error field
2. Manual correction of the underlying issue
3. Resetting the job to pending state:

import { markAccountingJobPending } from "@cona/core";

await markAccountingJobPending({
  organizationId: "org_123",
  documentId: "doc_456",
  versionHint: BigInt(currentVersion),
  delaySeconds: 0,
});

​

Stale Job Recovery

WHERE (
  (state = 'pending' AND ready_at <= NOW())
  OR (state = 'leased' AND last_attempted_at < NOW() - INTERVAL '15 minutes')
)

​

Troubleshooting

​

Jobs Not Processing

1. Check worker is running: ps aux | grep temporal-worker
2. Check Temporal UI for workflow status at https://cloud.temporal.io
3. Check database for pending jobs count

​

High Failure Rate

1. Review last_error field in failed jobs
2. Check database connection pool availability
3. Check for schema mismatches or missing object types
4. Verify accounting rules are configured

​

Growing Queue

1. Increase ACCOUNTING_QUEUE_BATCH_SIZE (current: 50)
2. Scale worker instances horizontally
3. Check for slow createFinancialEntries calls
4. Verify database performance

​

Starting the Queue Worker

temporal workflow start \
  --task-queue accounting-queue \
  --type syncAccountingQueueWorkflow \
  --workflow-id accounting-queue-worker-production \
  --input '{"batchSize": 50, "pollIntervalSeconds": 10}'

​

Implementation Details

​

Database Schema

model accounting_work_queue {
  id                String               @id @default(cuid())
  created_at        DateTime             @default(now())
  updated_at        DateTime             @updatedAt
  org_id            String
  document_id       String
  version_hint      BigInt
  ready_at          DateTime             @default(now())
  state             ACCOUNTING_JOB_STATE @default(pending)
  run_count         Int                  @default(0)
  last_attempted_at DateTime?
  last_error        String?

  organization organization @relation(fields: [org_id], references: [id])
  document     documents    @relation(fields: [document_id], references: [id])

  @@unique([org_id, document_id])
  @@index([state, ready_at])
  @@index([org_id])
}

​

Key Files

• Schema: packages/database/prisma/schema.prisma
• Core Queue: packages/core/src/accounting/queue/
• Activities: packages/temporal-workflows/src/activities/accounting/
• Workflows: packages/temporal-workflows/src/workflows/accounting/
• Worker: apps/temporal-workers/src/worker.ts

​

Migration from Old System

• ✅ Survives worker crashes (durable in PostgreSQL)
• ✅ Supports safe concurrent processing (FOR UPDATE SKIP LOCKED)
• ✅ Provides observability (query queue state)
• ✅ Handles retries with exponential backoff
• ✅ Decouples document creation from accounting processing
• ✅ Enables horizontal scaling