Background jobs¶

Async work on dev2.mycashless.com runs in two complementary layers.

Layer	Workload	Trigger
`cashless-cron`	One curl pod per tick that POSTs into the cm_v2 admin API.	Kubernetes `CronJob` (`batch/v1`).
`cashless-worker`	Long-running pod with four supervised Python loops.	`Deployment` `replicas: 1` + Bash supervisor (`backgrounds/entrypoint.sh`).
`webhook-drain`	One-shot Python invocation that drains `webhook_dispatch`.	Kubernetes `CronJob` shipped with `cashless-backend-v2`.
`cashless-db-backup`	`mysqldump` + GCS upload of the `cm1` Cloud SQL instance.	Kubernetes `CronJob` shipped with `cashless-db-backup`.

There is no RQ / Celery / Dramatiq

services.md describes cashless-worker as an "RQ / Redis" queue worker for historical reasons. It is not. The four worker scripts each poll a MySQL table on a fixed sleep interval; no broker, no Queue.enqueue(), no fan-out workers. Redis is used today only for the auth-client claims cache, slowapi rate-limiting storage, and the RFC-draft Idempotency-Key cache — not as a job queue.

Scheduled jobs (Kubernetes CronJobs)¶

Every CronJob resource on the cluster, verified by reading the manifest that creates it:

Name	Schedule	Purpose	Source
`refresh-global-devices`	`0 /3 * *` (UTC)	curl POST to `/v2/cronjobs/refresh-global-devices/trigger`. Inside cm_v2 recomputes `DeviceRegistry.used_events / used_days / app_version / last_used_*days` from per-event `Device` rows. Direct port of legacy GAE `GET /api/cron/global/devices`.	`services/cashless-cron/.../cronjobs.yaml` + `values.yaml`
`cashless-backend-v2-webhook-drain`	`/1 * * *`	`python -m cm_v2.tools.drain_webhooks` — pulls up to 50 `webhook_dispatch` rows with `status=pending` and `next_attempt_at<=NOW()`, fires one 3s HTTP POST each, applies the backoff schedule (1s, 5s, 30s, 5m, 30m → dead). `FOR UPDATE SKIP LOCKED` so parallel runs don't contend.	`backend/cashless-backend/.../webhook-drain-cronjob.yaml` + `drain_webhooks.py`
`cashless-db-backup`	`0 3 * * *` (UTC)	`mysqldump` of `cm1` via the in-pod Cloud SQL Auth Proxy sidecar → gzip → `gsutil cp` to `gs://<bucket>/cashless/daily/`. 30-day retention pruned at end of run.	`services/cashless-db-backup/.../cronjob.yaml` + `values.yaml`

Shared CronJob hygiene¶

Every CronJob in the cluster sets:

concurrencyPolicy: Forbid — a long run skips the next tick rather than overlap.
startingDeadlineSeconds: 60–600 — slip budget if the controller is busy.
activeDeadlineSeconds ceiling — hard kill so a stuck pod doesn't sit forever.
ttlSecondsAfterFinished — auto-delete Job pods after 10min–24h.
successfulJobsHistoryLimit / failedJobsHistoryLimit — keep the last 1–3 pods around for kubectl logs post-mortem.
Non-root pod (UID 1000 or 65532), read-only rootfs, all caps dropped, seccomp RuntimeDefault.

The full per-entry override surface for cashless-cron is documented in services/cashless-cron/README.md.

Cron parity status¶

The legacy stack scheduled 14 endpoints across cashless-backend v1/v2 and personalaccount-api v2. The post-Phase-2 state, audited on 2026-05-06:

Status	Count	Notes
REWIRED	6	5 manual-trigger endpoints on `pa_v4` (`POST /v4/cron/...`) + `refresh-global-devices` + `webhook-drain`.
MISSING (gaps)	4	Stripe auto-refund, Stripe auto-capture, Mercado Pago auto-refund, Stripe PI cancel. All DP-blocked on processor-SDK integration.
INTENTIONALLY RETIRED	4	Order/Sale-to-Transaction migration crons (data pre-migrated in cashless-models v2); legacy GAE `cron.yaml` (replaced by Helm-managed `CronJob`s).

The five rewired pa_v4 jobs (/v4/cron/remove-expired-verifications, /expire-unpaid-transfers, /expire-inprogress-transfers, /usercode/logged-in, /usercode/transfer, /transaction/auto-complete) are not yet on a scheduled trigger — they exist as authenticated POST endpoints invokable on demand. Adding them to cashless-cron is mechanical (see Adding a new scheduled job) but pending product sign-off on cadence.

Full breakdown: .planning/cron-parity.md.

The long-running worker (`cashless-worker`)¶

A singleton Deployment (replicas: 1, strategy: Recreate, PDB minAvailable: 1) running the cashless-worker image built from backend/cashless-backend/backgrounds/. The container PID 1 is entrypoint.sh, a Bash supervisor that re-launches each child after a 5s delay if it crashes:

run_with_restart notification.py 5 &
run_with_restart nfc_logs.py 5 &
run_with_restart sale_logs.py 5 &
run_with_restart transaction_logs.py 5 &
wait

Each child is an independent polling loop. There is no broker — the "queue" is a MySQL table read with SELECT ... LIMIT <PER_PAGE> on each tick.

Script	Polled table → sink	Tick	Idempotency strategy
`notification.py`	`notification` → FCM + Mailgun + Twilio.	60s loop	Atomic `UPDATE notification SET delivered=True WHERE id IN (...)` after dispatch.
`nfc_logs.py`	`nfc_data` (`synced=False`) → MongoDB (`cm.nfclogs`).	30s idle / 0.1s busy	`synced=True` flip after the MongoDB write.
`sale_logs.py`	`sale_log` + `sale_item_log` → `sale` + `sale_item`.	30s idle / 0.05s busy	Collision resolution by latest non-cancellation `uid`; `is_active=False` on the log row.
`transaction_logs.py`	`transaction_log` + `transaction_item_log` → `transaction` + `transaction_item`.	continuous	`SELECT ... FOR UPDATE SKIP LOCKED` + MySQL `GET_LOCK(<uid>)` belt-and-braces against double-promotion across instances.

Producer → consumer¶

flowchart LR
    POS([mPOS / dispatch_app]) -- HTTP --> CM[cashless-backend-v2]
    ADMIN([Admin UI]) -- HTTP --> CM
    CM -- INSERT --> NL[(notification)]
    CM -- INSERT --> NFC[(nfc_data)]
    CM -- INSERT --> SL[(sale_log)]
    CM -- INSERT --> TL[(transaction_log)]
    CM -- INSERT --> WD[(webhook_dispatch)]

    NL --> N(notification.py)
    NFC --> NF(nfc_logs.py)
    SL --> SLO(sale_logs.py)
    TL --> TLO(transaction_logs.py)
    WD --> DRAIN(drain_webhooks.py - CronJob)

    N --> FCM[FCM / Mailgun / Twilio]
    NF --> MONGO[(MongoDB cm.nfclogs)]
    SLO --> SALE[(sale / sale_item)]
    TLO --> TXN[(transaction / transaction_item)]
    DRAIN --> THIRDPARTY[Company-ticket partner HTTP endpoint]

Worker invariants¶

Singleton. Running two replicas would double every FCM push, every SMS, and every email. The chart enforces replicas: 1 and uses strategy: Recreate so a rollout fully terminates the old pod before the new one starts. Do not bump replicas without first sharding the polling queries by id MOD N.
MODELS_FIELD_ENCRYPTION_KEY must be present (mounted via ExternalSecrets from cashless-prod-models-field-encryption-key) — the worker reads Customer.clabe_account / Customer.id_number (Fernet-encrypted columns) and will raise FieldEncryptionKeyMissingError on any notification touching a PAUser join otherwise.
No HTTP surface. No probes, no Service. K8s only learns of death via process exit; the in-pod Bash supervisor restarts crashed children in 5s.

Idempotency caveat — notification.py

The notification table only carries a single delivered: bool column. If the FCM batch dispatch partially succeeds and then the process is SIGKILLed before the UPDATE ... SET delivered=True commit, the next tick will redeliver to every device in the batch — there is no per-device delivery ledger. The F9 P0 defense-in-depth content scan (notification.py::send_batch_notifications) drops the entire batch if any hard-PII is detected in the title or body, but it does not address replay. Treat the worker as at-least-once for push, email, and SMS.

The Redis substrate¶

Redis backs caching and rate-limiting only — never job state. Connection details live in cashless-redis-creds.REDIS_URL, written by the cashless-redis wrapper chart and consumed via envFrom: secretRef.

Logical use	Where	Key shape
Claims cache (auth)	`cm_v2/auth_client.py` + `pa_v4/auth_client.py`	`cashless:claims:<sha256(token)>` with 60s TTL.
Rate limiting (slowapi)	`cashless-auth`, `cm_v2`, `pa_v4`	slowapi internal (`LIMITS/...`); writes via `storage_uri`.
Token revocation blocklist	`cashless-auth`	`cashless:token:revoked:<jti>`, TTL = refresh-token lifetime.
`Idempotency-Key` cache (HTTP)	`cm_v2/idempotency.py` + sibling middleware	DB-backed (`idempotency_key` table, migration 018) — not Redis.

A single in-cluster master backs everything; architecture: standalone, 8Gi PVC, maxmemory 384mb with allkeys-lru eviction (values.yaml). No DB-index sharding — every caller writes to db 0.

Idempotency¶

Surface	Mechanism
HTTP `POST/PATCH/PUT/DELETE` on cm_v2 / pa_v4	RFC-draft `Idempotency-Key` header. Same key + same body hash → cached response replayed for 24h. Different body → 422 `idempotency_key_request_mismatch`. Stored in the `idempotency_key` MySQL table.
Cron trigger (cm_v2 admin)	Cron pod sends `Idempotency-Key: <UTC-timestamp>-<entry-name>` per tick. cm_v2 has not wired the cache for the trigger endpoint yet — the client side is already paying its dues so it's a one-line server change when needed.
Webhook dispatch	`enqueue_company_ticket_webhook` (`cm_v2/services/webhooks.py`) short-circuits when a `WebhookDispatch` row for the `transfer_uid` already exists. Drain itself uses `FOR UPDATE SKIP LOCKED` so parallel runners never re-attempt the same row.
Payment-webhook finaliser	`register_charge_result` in `pa_v4/services/reload/charge.py` upserts the `OnlinePayment` row; redelivery for an already-finalised charge short-circuits before any side-effect runs. The atomicity guarantee is documented in `pa_v4/CLAUDE.md`.
Transaction-log promotion	`transaction_logs.py` holds a MySQL `GET_LOCK(<uid>, 10)` for the duration of each promotion so two instances never both flip the same `uid`.
Notification dispatch	None at the device level — see warning above. The flag is per-`Notification`, not per-`(notification, device)`.

Observability¶

Every worker uses structlog configured by backgrounds/log_config.py — JSON output on stdout, picked up by the GKE logging agent. Each event carries a worker= tag (notifications / nfc / sale / transaction).

You want to see…	Where
Is the worker pod alive?	`kubectl -n cashless get deploy cashless-worker` (replicas 1/1).
Last 200 lines of one supervised child	`kubectl -n cashless logs deploy/cashless-worker \\| grep '"worker": "transactions"'`
Did the last cron tick fire?	`kubectl -n cashless get job -l cashless.media/cron-entry=refresh-global-devices --sort-by=.metadata.creationTimestamp`
Why is the cron pod failing?	`kubectl -n cashless logs -l cashless.media/cron-entry=<name> --tail=200`
Webhook backlog	`SELECT status, COUNT(*) FROM webhook_dispatch GROUP BY status;` (`pending` + `next_attempt_at > NOW()` is the queue depth).
Notification backlog	`SELECT COUNT(*) FROM notification WHERE delivered = 0 AND deliver_time <= UTC_TIMESTAMP();`
Sale / transaction log backlog	`SELECT COUNT() FROM sale_log WHERE is_active = 1;` / `SELECT COUNT() FROM transaction_log WHERE is_active = 1;`

There is no separate dashboard for queue depth yet — backlog SQL above is the source of truth. OpenTelemetry traces from cm_v2 / pa_v4 cover the producer side (the INSERT that enqueues a row); worker spans are not yet emitted.

TBD — worker tracing

Adding OTLP-emitting context to each polling tick is on the post-launch list. Until then, correlate by notification.id / transfer.uid / transaction.uid strings in structlog and producer-side traces.

Adding a new scheduled job¶

A new K8s CronJob that hits a cm_v2 admin endpoint:

Register it in apiserver/cm_v2/routers/cronjobs.py::_CRONJOBS (name, description, enabled=True, documented schedule). POST /v2/cronjobs/{name}/trigger 404s until the name is in this list.

Append an entry to entries: in services/cashless-cron/charts/cashless-cron/values.yaml:

entries:
  - name: my-new-job          # MUST match the cm_v2 registry name
    schedule: "*/15 * * * *"
    description: "..."
    activeDeadlineSeconds: 300

Commit on cashless-v2. ArgoCD syncs the new CronJob.

A new long-running polling worker (rarely the right move — prefer a CronJob invoking a one-shot Python entrypoint, e.g. cm_v2.tools.drain_webhooks):

Drop a module under backend/cashless-backend/backgrounds/ with a main() that takes db_url, iterations, session_factory, and sleep keyword args (mirror sale_logs.py).
Add run_with_restart your_script.py 5 & to backgrounds/entrypoint.sh.
Bump the worker image and the chart pointer in deploy/argocd/values/cashless-worker.yaml.
Validate locally with iterations=1 before pushing — the entrypoint restart loop hides crash-on-startup behind a 5s backoff.

There is no broker to register with, no queue name to claim, and no worker pool to scale. Adding a job is "add a polling loop or add a CronJob" — that is the whole surface today.