Jobs, Queue, Cache & Observability
The cron scheduler and full job inventory, the (dormant) RabbitMQ queue, the Valkey cache strategy, and the OpenTelemetry → SigNoz pipeline.
Background work, caching, and telemetry. The headline: cron does the real work; the queue is provisioned but dormant.
Cron scheduler
Built on gocron v2 plus a custom ScheduledTask abstraction (scheduler/scheduler.go). Each concrete task embeds *BaseTask (the schedule string passed to NewBaseTask is the cron pattern). Every task is wrapped by TrackedTask (scheduler/tracked_task.go), which persists execution records to the job_executions table (all tracking is recover()-guarded so it can't break the job).
Execution loop (startScheduler, main.go): gocron.NewScheduler(), each job registered with gocron.CronJob(schedule, true) and WithSingletonMode(LimitModeReschedule) (prevents overlapping runs — critical for the many every-minute jobs). Event listeners feed metrics.GetCronMetrics().
time.Local is forced to UTC — all cron schedules are UTC. And several inline comments on schedules are stale; trust the cron string, not the comment (e.g. vendor-balance-sync comment says "every 3 hours" but runs */15).
Full cron inventory
| Job | Schedule (UTC) | Purpose | Feature gate |
|---|---|---|---|
export-processor | every min | Process pending data-export jobs | always |
export-cleanup | daily 02:00 | Delete old export files | always |
job-execution-cleanup | daily 03:00 | Purge old job_executions | always |
webhook-delivery | every min | Send outbound webhooks (orders, wallets) | always |
inventory-pump | every 6h | Pump voucher inventory DB→Valkey | vouchers |
pending-order-retry | every min | Retry failed/pending voucher orders | vouchers |
prefetch-processor | every 2 min | Process voucher prefetch jobs | vouchers |
import-processor | every min | Process product/inventory imports | vouchers |
import-cleanup | daily 00:00 | Clean old import jobs | vouchers |
vendor-catalog-sync | daily 02:00 | Sync vendor catalogs (snapshots) | vouchers |
vendor-catalog-cleanup | Sun 03:00 | Delete old catalog snapshots | vouchers |
vendor-catalog-retry | every min | Retry failed catalog syncs | vouchers |
g2a-reservation-expiry | every 5 min | Expire stale RESERVED G2A holds (30m TTL) | vouchers |
esim-installation-poll | every 5 min | Poll eSIM install/activation | esim |
esim-order-retry | every min | Retry PENDING eSIM orders | esim |
esim-expiry | daily 02:00 | Mark eSIM orders EXPIRED | esim |
vendor-balance-sync | every 15 min | Sync vendor wallet balances | vouchers OR topups OR esim |
fetch-shopify-orders | every min | Fetch new Shopify orders | shopify |
process-pending-shopify-orders | every min | Fulfil pending Shopify orders | shopify |
detect-shopify-cancellations | every 5 min | Detect Shopify cancellations | shopify |
payout-webhook-delivery | every min | Send payout status webhooks | payouts |
scheduled-payout-processor | every min | Process due scheduled payouts | payouts |
topup-order-retry | every min | Retry pending topup orders | topups |
Two constructors exist but are not registered (dormant): payout-processor, payout-status-sync.
go run main.go cron # run the scheduler (all registered)
go run main.go cron inventory-pump pending-order-retry # run specific jobs onlyQueue (RabbitMQ)
The queue/ package is a complete provider abstraction (Send/Receive/Disconnect), with a RabbitMQ implementation (durable topic exchange from QUEUE_EXCHANGE, routing key = channel, manual ack, W3C trace-context propagation on every message) and an in-memory one for tests. queue/worker.go defines the Processor interface consumers implement.
No workers are registered
processorRegistry is []queue.Processor{} (main.go:165) and nothing is appended. go run main.go worker starts zero consumers, and there are no queue.Send call sites in services//jobs/. RabbitMQ is provisioned and the plumbing is ready, but today all async work flows through cron, not the queue. Keep this in mind before debugging "missing" queue workers.
The processor/ package is not queue workers — it's synchronous import processors (INVENTORY, PRODUCTS, VENDOR_PRODUCTS, CLIENT_PRODUCT_DETAILS) driven by the import-processor cron.
Cache (Valkey)
cache/ exposes a rich Cache interface: KV (Retrieve/Store(key,val,ttl)/Remove/DeleteByPattern), lists, a FIFO queue, sorted sets (ZAdd/ZRangeByScore — used for inventory pools), and hashes. Implementations: cache_valkey.go (prod, uses valkeyotel for automatic tracing), cache_memory.go, cache_nil.go. Selected by CACHE_TYPE. Keys are namespaced by CACHE_PREFIX (sandbox: in sandbox). DeleteByPattern uses SCAN for production safety.
Strategy (convention):
| Data | TTL |
|---|---|
| Reference (countries, currencies) | 1 day + 10 min stale-while-revalidate |
| Catalog (products, categories) | 5 min |
| Dynamic / user data | none |
Stale-while-revalidate is a documented policy, not a coded primitive — TTLs are applied per call site via Store(..., ttl). Inventory pools use keys like pool:%d:%.2f:%s (product/denom/vendor) with a 24h TTL, rebuilt by inventory-pump.
Observability (OpenTelemetry → SigNoz)
Gated on OTEL_ENABLED (default false). When on, main.go init() sets up traces → metrics → instruments → logs, all via OTLP HTTP.
- Traces (
utils/tracer.go, globalutils.Tracer): OTLP HTTP toOTEL_EXPORTER_OTLP_ENDPOINT(defaultlocalhost:4318), auth headersignoz-access-tokenfromOTEL_EXPORTER_OTLP_TOKEN.AlwaysSample, batched. Registers W3C TraceContext + Baggage propagators, so an incomingtraceparent(e.g. from the Grasshopper worker) continues the same trace cross-service. - Metrics (
utils/metrics_exporter.go): OTLP HTTP;PeriodicReader(default 60s). - Logs (
utils/log_exporter.go): OTLP HTTP log exporter, batched.
All three signals go to an OpenTelemetry Collector (otel-collector-config.yaml) that also scrapes host metrics, then exports OTLP → SigNoz (SIGNOZ_ENDPOINT).
In production the collector and SigNoz run on the sandbox/telemetry server; ingest is fronted by otel.octopuscards.io (→ 127.0.0.1:4318) and the UI by telemetry.octopuscards.io (→ 127.0.0.1:8080). See Infrastructure and Operations.
zap ↔ OTel bridge
The base logger is the singleton's zap; retrieve it request/task-scoped via middleware.Logger(ctx). When OTel is enabled, an otelzap core is tee'd with the stdout core — logs go to both stdout and SigNoz. When off, stdout only.
Metrics with drill-down dimensions
Instruments (meter "octopus") carry client_id / product_id / vendor attributes for Grafana/SigNoz drill-down — a house rule that all business metrics must include client_id and product_id (Memory Appendix). Examples: InventoryAllocated/Depleted (product_id, vendor), VouchersIssued/Failed (vendor, product_id), G2ACodesDelivered (client_id, product_id), plus HTTP, orders, topups, payouts, wallet, and vendor-API latency series. Cron metrics (CronMetrics/CronJobInfo) track per-job last/next run, duration, running gauge, and status, persisted to cache.
Running locally
docker compose up -d brings up Postgres (5432), Valkey (6379), RabbitMQ (5672 + UI 15672), the OTel collector (4328 HTTP / 4327 gRPC → SigNoz), and MySQL (3306). Point OTEL_EXPORTER_OTLP_ENDPOINT at localhost:4328 to use the docker collector.
Wallet & Ledger
The client money layer every order and payout debits — wallets, the single-entry ledger, the atomic debit stored procedure, the refund paths, charges/FX, and prepaid voucher-burning. Includes precision and double-refund hazards.
Cron Catalog
Every scheduled task — the actual cron string (source of truth), whether it's registered, what it drives, batch/timeout, and the two implemented-but-unregistered payout crons. Plus the stale-comment schedule traps.