From 7ebfa4fd30dbcbde1402067f8b5d99efa7e315a6 Mon Sep 17 00:00:00 2001 From: Michael Tarassov Date: Sun, 23 Aug 2026 14:35:26 +0500 Subject: [PATCH] =?UTF-8?q?feat(jobs):=20transactional=20outbox=20?= =?UTF-8?q?=E2=80=94=20opt-in=20atomic=20event=20dispatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After-response dispatch (4-arg with_repo_errors) is best-effort by design: a process dying between the DB commit and Jobs::submit loses the event. For must-not-lose events (money mail, reconciled webhooks) add the outbox: - migrations/010_outbox.sql: outbox table (kind, jsonb payload, claimed_at, attempts, last_error) + partial index over the unclaimed backlog - src/jobs/Outbox.hpp/.cpp: Outbox::enqueue(txn, kind, payload) INSERTs in the caller's open transaction (commit = event durable, rollback = gone); Outbox::drain(batch) claims via UPDATE ... FOR UPDATE SKIP LOCKED, relays to Jobs::submit, DELETEs on success, releases with attempts+1 + last_error on failure; stale claims (dead drainer) re-claimable after kStaleClaimSec — at-least-once end to end - Core schedules the drain every outbox.drain_interval_sec seconds (OUTBOX_DRAIN_INTERVAL_SEC, default 0 = off — the pattern is opt-in; existing email/webhook flows keep their after-response path unchanged) - docs: CONVENTIONS gotcha 20 (outbox vs after-response decision + 3-line billing-fork example), CONFIG.md row, module-deps edge jobs -> database - tests/integration/test_outbox.cpp: commit->drain->job-in-Redis, rollback invisibility, failed-submit retry accounting, concurrent-drain exactly-once partition --- config/config.json | 3 + config/config.sample.json | 3 + docs/CONFIG.md | 1 + docs/CONVENTIONS.md | 9 ++ docs/module-deps.txt | 1 + migrations/010_outbox.sql | 38 +++++ src/core/Core.cpp | 26 ++++ src/core/Core.hpp | 6 + src/jobs/Outbox.cpp | 105 ++++++++++++++ src/jobs/Outbox.hpp | 93 ++++++++++++ tests/integration/test_outbox.cpp | 226 ++++++++++++++++++++++++++++++ 11 files changed, 511 insertions(+) create mode 100644 migrations/010_outbox.sql create mode 100644 src/jobs/Outbox.cpp create mode 100644 src/jobs/Outbox.hpp create mode 100644 tests/integration/test_outbox.cpp diff --git a/config/config.json b/config/config.json index a3d9491..5d77df6 100644 --- a/config/config.json +++ b/config/config.json @@ -198,6 +198,9 @@ "dlq_metric_refresh_sec": "${JOBS_DLQ_METRIC_REFRESH_SEC:-10}", "queue_metric_refresh_sec": "${JOBS_QUEUE_METRIC_REFRESH_SEC:-10}" }, + "outbox": { + "drain_interval_sec": "${OUTBOX_DRAIN_INTERVAL_SEC:-0}" + }, "content": { "enabled": "${CONTENT_ENABLED:-false}" }, diff --git a/config/config.sample.json b/config/config.sample.json index 1ec3d47..b672bbf 100644 --- a/config/config.sample.json +++ b/config/config.sample.json @@ -198,6 +198,9 @@ "dlq_metric_refresh_sec": "${JOBS_DLQ_METRIC_REFRESH_SEC:-10}", "queue_metric_refresh_sec": "${JOBS_QUEUE_METRIC_REFRESH_SEC:-10}" }, + "outbox": { + "drain_interval_sec": "${OUTBOX_DRAIN_INTERVAL_SEC:-0}" + }, "content": { "enabled": "${CONTENT_ENABLED:-false}" }, diff --git a/docs/CONFIG.md b/docs/CONFIG.md index a5d8425..0a62011 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -211,6 +211,7 @@ For URL components: `REDIS_HOST`, `REDIS_PORT`. | `JOBS_VISIBILITY_TIMEOUT_SEC` | `jobs.visibility_timeout_sec` | int | `0` | Processing lease: a job whose worker dies is re-queued after this many seconds. `0` disables leases (legacy behaviour) | | `JOBS_DLQ_METRIC_REFRESH_SEC` | `jobs.dlq_metric_refresh_sec` | int | `10` | Exports `jobs_dlq_depth{type="..."}` plus an aggregate `type="_total"` | | `JOBS_QUEUE_METRIC_REFRESH_SEC` | `jobs.queue_metric_refresh_sec` | int | `10` | Same bookkeeping for the waiting queue: `jobs_queue_depth{type="..."}` plus `type="_total"` | +| `OUTBOX_DRAIN_INTERVAL_SEC` | `outbox.drain_interval_sec` | int | `0` | Transactional outbox (`src/jobs/Outbox.hpp`): how often the API pod relays `outbox` table rows to the job queue. `0` (default) disables draining — the pattern is opt-in; rows written via `Outbox::enqueue` sit in Postgres until a deploy enables this. Needs `jobs.enabled=true`. | | `DB_REPLICA_LAG_METRIC_REFRESH_SEC` | `database.replica_lag_metric_refresh_sec` | int | `15` | Refresh interval for the `db_replica_lag_seconds` gauge. Only registered when read replicas are configured (primary has no replay timestamp). | ## Content diff --git a/docs/CONVENTIONS.md b/docs/CONVENTIONS.md index 3fe12d2..e85b7bd 100644 --- a/docs/CONVENTIONS.md +++ b/docs/CONVENTIONS.md @@ -164,6 +164,14 @@ These don't fall out of reading the code; they were learned the hard way. 17. **Multi-table test cleanup goes through `TestHelpers::wipe_app_data()`.** It is the one place that knows the FK order (users CASCADE before extra roles; posts/audit_log/used_tokens ride one TRUNCATE) and re-seeds migration 001's two roles. Don't hand-roll TRUNCATE/DELETE sequences in fixtures. 18. **Never accumulate with a self-referencing upsert** (`INSERT ... ON CONFLICT (k) DO UPDATE SET x = t.x + EXCLUDED.x`) through `Database::execute_write`. In a downstream fork every second-or-later write computed as though the existing row's value were 0 — invisible for positive deltas, fatal for negative ones (`CHECK` violations on refunds). CI evidence proved the conflicting row WAS detected and visible earlier in the same transaction, yet the self-referencing SET still read it as absent; the root cause was never found (forensics: site fork, commit b676430). Safe idiom: `INSERT ... ON CONFLICT DO NOTHING` (materialize the row so it can be locked) → `SELECT ... FOR UPDATE` → compute the new total in C++ → plain `UPDATE`. 19. **After-response side effects go through the 4-arg `with_repo_errors` overload** (email dispatch, enqueue, webhook fire): `after_fn` runs once the guarded block completed, OUTSIDE the catch ladder, so a throwing side effect can never fire `callback` a second time. Stash results into a `std::optional` captured by reference if `after_fn` needs them (learned downstream: a receipt-email dispatch inside the guarded lambda double-fired the callback). +20. **Events that must not be lost go through the transactional outbox** (`src/jobs/Outbox.hpp`, migration 010, opt-in via `OUTBOX_DRAIN_INTERVAL_SEC`). The after-response hook above is best-effort BY DESIGN: a process dying between the DB commit and `Jobs::submit` loses the event, and a failed submit is only logged. Fine for a re-requestable confirm-email; not fine for money paths, where "wallet debited but the receipt/webhook never fired" is a truth defect someone repairs by hand. There, write the event in the SAME transaction as the ledger write — commit makes both durable, rollback erases both — and the periodic drain relays it to the job queue (at-least-once; handlers must tolerate a duplicate, which Jobs' retry/lease paths already require). A billing-fork receipt dispatch is three lines inside the existing wallet transaction: + ```cpp + // inside Database::execute_transaction([&](auto& txn) { ... ledger write ... + if (result.credited) // same dedupe gate BillingEmails::receipt documents + Jobs::Outbox::enqueue(txn, Email::SendEmail::kJobType, + {{"to", user.email}, {"subject", subject}, {"text", text}}); + ``` + The template's own email flows deliberately stay on the after-response path (bit-for-bit unchanged behaviour) — the outbox is the upgrade a fork opts specific call sites into, not a global rewire. ### Frontend gotchas @@ -193,5 +201,6 @@ The scaffolding scripts are the entry points: | Check spec ↔ code drift | `./scripts/check-openapi-drift.sh` | | Validate Helm render | `make helm-validate` | | Worked CRUD example | `docs/EXAMPLES.md` | +| Must-not-lose event dispatch (outbox vs after-response) | `src/jobs/Outbox.hpp` (doc comment is the decision guide; gotcha 20 above has the worked example) | | Money / ledger / payment-provider reference | `src/billing/Wallet.hpp` (billing module: append-only ledger, idempotent capture/refund/spend — spend is reference-keyed via migration 009's partial unique index, no HTTP endpoint (charging is fork domain), integer-only money; best-effort emails in `src/email/BillingEmails.hpp`) | | ADRs / architecture decisions | `docs/adr/` | diff --git a/docs/module-deps.txt b/docs/module-deps.txt index b9f8651..6bc3356 100644 --- a/docs/module-deps.txt +++ b/docs/module-deps.txt @@ -59,6 +59,7 @@ email -> repositories email -> security email -> utils jobs -> cache +jobs -> database jobs -> email jobs -> observability jobs -> utils diff --git a/migrations/010_outbox.sql b/migrations/010_outbox.sql new file mode 100644 index 0000000..0c48bbd --- /dev/null +++ b/migrations/010_outbox.sql @@ -0,0 +1,38 @@ +-- Migration 010: transactional outbox — opt-in atomic event dispatch. +-- +-- Applied in numeric order on boot. The MigrationRunner wraps this file in ONE +-- transaction under an advisory lock — do NOT add BEGIN/COMMIT. Idempotent DDL. +-- +-- The after-response hook (HandlerSupport.hpp 4-arg with_repo_errors) fires a +-- side effect AFTER the DB commit: a process that dies between the commit and +-- Jobs::submit loses the event forever. For events that must not be lost +-- (money mail, webhooks about paid state), the row is written HERE, inside the +-- SAME transaction as the domain write, and a periodic drain task relays it to +-- the Redis job queue afterwards (src/jobs/Outbox.hpp). Commit → the event is +-- durable; rollback → the event never existed. Delivery is at-least-once. +-- +-- Columns: +-- kind the Jobs type Outbox::drain() will submit ("email.send", +-- "webhook.deliver", any registered handler type) +-- payload the exact Jobs::submit payload, opaque to the outbox +-- claimed_at NULL = ready for drain; non-NULL = a drainer claimed it +-- (re-claimable after Outbox::kStaleClaimSec so a drainer that +-- died between claim and submit can't strand the row forever) +-- attempts failed submit attempts so far (grows without bound — retry +-- forever is correct for "must not be lost"; alert on it) +-- last_error what the most recent failed submit threw +CREATE TABLE IF NOT EXISTS outbox ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + kind TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + claimed_at TIMESTAMPTZ, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT NOT NULL DEFAULT '' +); + +-- Drain order is FIFO over the unclaimed backlog; the partial index keeps the +-- claim query cheap no matter how many claimed-in-flight rows exist. +CREATE INDEX IF NOT EXISTS idx_outbox_unclaimed + ON outbox (created_at) + WHERE claimed_at IS NULL; diff --git a/src/core/Core.cpp b/src/core/Core.cpp index dfc1d4f..a7c63ef 100644 --- a/src/core/Core.cpp +++ b/src/core/Core.cpp @@ -30,6 +30,7 @@ #include "database/Migrations.hpp" #include "email/Mailer.hpp" #include "jobs/Jobs.hpp" +#include "jobs/Outbox.hpp" #include "messaging/Messaging.hpp" #include "observability/Observability.hpp" #include "security/Auth.hpp" @@ -599,6 +600,31 @@ void Application::init_jobs_(Config::AppConfig& cfg) { Jobs::get().set_visibility_timeout(visibility); register_dlq_metric_(cfg); register_queue_depth_metric_(cfg); + register_outbox_drain_(cfg); +} + +void Application::register_outbox_drain_(Config::AppConfig& cfg) { + // Tasks is only up in server mode (worker_main runs its own loop), and the + // drain needs both stores: rows come from Postgres, jobs land in Redis. + if (!Tasks::is_initialized() || !Database::is_initialized() || !Jobs::is_initialized()) + return; + const int interval_sec = cfg.get("outbox.drain_interval_sec", "OUTBOX_DRAIN_INTERVAL_SEC", 0); + if (interval_sec <= 0) + return; // opt-in: the outbox table sits inert until a deploy enables draining + spdlog::info("Transactional outbox drain enabled: every {}s", interval_sec); + Tasks::schedule_recurring("outbox_drain", std::chrono::seconds(interval_sec), [] { + if (!Database::is_initialized() || !Jobs::is_initialized()) + return; + try { + auto stats = Jobs::Outbox::drain(); + if (stats.failed > 0) + spdlog::warn("outbox drain: {} row(s) failed to submit (will retry)", stats.failed); + } catch (const std::exception& e) { + // DB hiccup — rows stay put, the next tick retries. Losing a tick + // never loses an event; that's the whole point of the table. + spdlog::warn("outbox drain failed: {}", e.what()); + } + }); } void Application::register_health_check(std::string name, HealthFn probe, bool critical) { diff --git a/src/core/Core.hpp b/src/core/Core.hpp index acb1267..b557c14 100644 --- a/src/core/Core.hpp +++ b/src/core/Core.hpp @@ -152,6 +152,12 @@ class Application { // so without a reaper the table + its index grow monotonically. static void register_token_reaper_(); + // Schedules the transactional-outbox drain (Jobs::Outbox::drain) every + // outbox.drain_interval_sec seconds. Opt-in: the default 0 schedules + // nothing, so the outbox table sits inert unless a deploy turns it on. + // Server-mode only (Tasks isn't initialized in the worker). + static void register_outbox_drain_(Config::AppConfig& cfg); + // Registers the subsystem probes the template ships with. Services // that add their own modules can call Core::get().register_health_check // at any point after Core::initialize() returns. diff --git a/src/jobs/Outbox.cpp b/src/jobs/Outbox.cpp new file mode 100644 index 0000000..7804e8a --- /dev/null +++ b/src/jobs/Outbox.cpp @@ -0,0 +1,105 @@ +/** + * @file Outbox.cpp + * @brief Body for src/jobs/Outbox.hpp — compiled once into app_core (ADR 0003 + * as amended 2026-08-22). The only jobs TU that includes the Database + * layer (edge declared in docs/module-deps.txt). + */ + +#include "jobs/Outbox.hpp" + +#include +#include +#include + +#include + +#include "database/Database.hpp" +#include "jobs/Jobs.hpp" + +namespace Jobs { +namespace Outbox { + +namespace { + +struct PendingRow { + std::string id; + std::string kind; + std::string payload; // jsonb as text; parsed at submit time + int attempts = 0; +}; + +} // namespace + +DrainStats drain(int batch, int reclaim_sec) { + DrainStats stats; + if (batch <= 0) + return stats; + + // 1. Claim a batch in a SHORT transaction — no Redis I/O while holding row + // locks. SKIP LOCKED makes concurrent drainers partition the backlog + // instead of blocking or double-claiming; the committed claimed_at then + // hides the rows from later passes until they're finalized (or the + // claim goes stale — reclaim_sec — because this process died here). + auto rows = Database::get().execute_write([&](auto& txn) { + std::vector out; + auto r = txn.exec_params( + "UPDATE outbox SET claimed_at = now() " + "WHERE id IN (SELECT id FROM outbox " + " WHERE claimed_at IS NULL " + " OR claimed_at < now() - make_interval(secs => $2::int) " + " ORDER BY created_at " + " LIMIT $1 " + " FOR UPDATE SKIP LOCKED) " + "RETURNING id, kind, payload, attempts", + batch, + reclaim_sec); + out.reserve(r.size()); + for (const auto& row : r) { + out.push_back({row["id"].template as(), + row["kind"].template as(), + row["payload"].template as(), + row["attempts"].template as()}); + } + return out; + }); + if (rows.empty()) + return stats; + + // 2. Relay each row to the job queue. A per-row failure (Redis down, Jobs + // not initialized, unparseable payload) is recorded, never thrown — + // one poisoned row must not stall the rest of the batch. + std::vector submitted_ids; + std::vector> failed; // id -> error + for (const auto& row : rows) { + try { + Jobs::get().submit(row.kind, json::parse(row.payload)); + submitted_ids.push_back(row.id); + } catch (const std::exception& e) { + spdlog::warn( + "outbox: submit of {} (kind={}, attempt {}) failed: {}", row.id, row.kind, row.attempts + 1, e.what()); + failed.emplace_back(row.id, e.what()); + } + } + + // 3. Finalize: drop the relayed rows, release the failed ones for the next + // pass. If THIS transaction is lost (process death, DB outage), the + // claims go stale and every row — including already-submitted ones — is + // redelivered after reclaim_sec: at-least-once, as documented. + Database::get().execute_write([&](auto& txn) { + for (const auto& id : submitted_ids) + txn.exec_params("DELETE FROM outbox WHERE id = $1", id); + for (const auto& [id, err] : failed) + txn.exec_params( + "UPDATE outbox SET claimed_at = NULL, attempts = attempts + 1, last_error = $2 WHERE id = $1", id, err); + return 0; + }); + + stats.submitted = static_cast(submitted_ids.size()); + stats.failed = static_cast(failed.size()); + if (stats.submitted > 0 || stats.failed > 0) + spdlog::debug("outbox: drained batch — {} submitted, {} failed", stats.submitted, stats.failed); + return stats; +} + +} // namespace Outbox +} // namespace Jobs diff --git a/src/jobs/Outbox.hpp b/src/jobs/Outbox.hpp new file mode 100644 index 0000000..b93b0d9 --- /dev/null +++ b/src/jobs/Outbox.hpp @@ -0,0 +1,93 @@ +/** + * @file Outbox.hpp + * @brief Transactional outbox: atomic "domain write + event" dispatch to the + * Jobs queue. Opt-in — nothing drains until `outbox.drain_interval_sec` + * (OUTBOX_DRAIN_INTERVAL_SEC) is set > 0. + * + * ## When to use the outbox vs the after-response hook + * + * The default dispatch discipline is the after-response hook — the 4-arg + * `Api::with_repo_errors` overload (api/HandlerSupport.hpp): commit the DB + * write, send the response, then fire the side effect (Jobs::submit, email, + * webhook). That is best-effort BY DESIGN: a process that dies between the DB + * commit and the submit loses the event, and a failed submit is only logged. + * For a confirm-email link the user can re-request, that trade-off is right — + * don't add an outbox row, a table and a drain hop for events whose loss is + * an inconvenience. + * + * Use the outbox when losing the event corrupts truth someone relies on: + * - money paths: a receipt/refund notice for a ledger write that DID happen + * (src/email/BillingEmails.hpp discipline — the wallet write is durable, + * so the notice about it must eventually go out too); + * - webhooks that downstream systems reconcile against ("payment captured"); + * - any event where "DB committed but nobody was told" needs a human to + * notice and repair by hand. + * + * Mechanics: `enqueue(txn, kind, payload)` INSERTs into the `outbox` table + * INSIDE the caller's open transaction — commit makes the event durable + * atomically with the domain write, rollback erases both. A periodic task + * (`Outbox::drain`, scheduled by Core when the interval is > 0) claims + * unclaimed rows (`FOR UPDATE SKIP LOCKED` — concurrent drainers never double- + * claim) and relays each to `Jobs::submit(kind, payload)`: the row is DELETEd + * on success, or released (attempts+1, last_error, claimed_at back to NULL) + * on failure so the next drain retries it. Delivery is therefore + * AT-LEAST-ONCE: a drainer dying between submit and delete redelivers after + * kStaleClaimSec — job handlers for outbox kinds must tolerate a duplicate + * (the same contract Jobs' retry/visibility-timeout paths already impose). + * + * Existing flows are NOT routed through here — the after-response paths keep + * their exact behaviour; the outbox is the opt-in upgrade for the call sites + * that need it. Worked example: docs/CONVENTIONS.md gotcha 20. + */ + +#pragma once + +#include + +#include "jobs/Job.hpp" + +namespace Jobs { +namespace Outbox { + +/// A claim older than this is considered abandoned (the drainer died between +/// claiming and finalizing) and becomes drainable again — the self-healing +/// half of the at-least-once contract. +inline constexpr int kStaleClaimSec = 300; + +/** + * @brief Record an event in the SAME transaction as the caller's domain + * write. @p txn is the `auto& txn` every repository lambda receives + * from Database::execute_write / execute_transaction + * (Database::detail::TracingTxn — only exec/exec_params are used). + * @p kind must be a job type a worker handles (jobs/BuiltinHandlers.cpp + * or a fork-registered handler); @p payload is passed to Jobs::submit + * verbatim. Throws on SQL failure — which rolls the caller's + * transaction back, exactly the atomicity the pattern promises. + */ +template +void enqueue(Txn& txn, const std::string& kind, const json& payload) { + txn.exec_params("INSERT INTO outbox (kind, payload) VALUES ($1, $2::jsonb)", kind, payload.dump()); +} + +/// What one drain pass did — returned for tests/observability; the periodic +/// task just logs it. +struct DrainStats { + long submitted = 0; ///< rows relayed to Jobs::submit and deleted + long failed = 0; ///< rows whose submit threw (attempts bumped, retried next pass) +}; + +/** + * @brief Relay up to @p batch pending outbox rows to the Jobs queue. + * Claim (UPDATE ... WHERE claimed_at IS NULL ... FOR UPDATE SKIP + * LOCKED RETURNING) → Jobs::submit per row → one finalize transaction + * (DELETE the submitted, release the failed). Safe to call from + * several processes at once. Throws only when the claim/finalize + * transactions themselves fail (DB down) — per-row submit failures are + * recorded on the row, never thrown. + * @param reclaim_sec claims older than this are treated as abandoned and + * re-claimed (defaults to kStaleClaimSec). + */ +DrainStats drain(int batch = 100, int reclaim_sec = kStaleClaimSec); + +} // namespace Outbox +} // namespace Jobs diff --git a/tests/integration/test_outbox.cpp b/tests/integration/test_outbox.cpp new file mode 100644 index 0000000..35fedd3 --- /dev/null +++ b/tests/integration/test_outbox.cpp @@ -0,0 +1,226 @@ +/** + * @file test_outbox.cpp + * @brief Integration tests for the transactional outbox (src/jobs/Outbox.hpp + * + migration 010): enqueue rides the caller's Postgres transaction, + * drain relays committed rows to the Redis job queue. + * + * The suite pins the four properties that ARE the pattern: + * 1. commit → drain submits the job (payload intact); + * 2. rollback → the event never existed (atomicity with the domain write); + * 3. a failed submit is recorded (attempts/last_error), released, and a + * later drain retries it to success; + * 4. concurrent drains partition the backlog (SKIP LOCKED) — every row + * submitted exactly once, none lost. + * + * Requires live Postgres (migrations on — 010 creates the table) and Redis. + */ + +#include +#include +#include +#include + +#include + +#include + +#include "database/Database.hpp" +#include "jobs/Jobs.hpp" +#include "jobs/Outbox.hpp" +#include "test_helpers.hpp" + +using json = nlohmann::json; + +namespace { + +class OutboxTest : public TestHelpers::CoreBackedTest { +protected: + // Outbox kinds double as job-queue names; TearDown drains exactly these. + static constexpr const char* kKinds[] = {"outboxq", "outboxq_rb", "outboxq_fail", "outboxq_conc"}; + + std::string config_file_name() const override { return "outbox_test_config.json"; } + + void config_overrides(json& cfg) override { + cfg["logging"]["name"] = "outbox_test"; + cfg["logging"]["file"] = "logs/outbox_test.log"; + cfg["observability"]["service_name"] = "outbox_svc"; + cfg["database"]["migrations_enabled"] = true; + cfg["database"]["migrations_dir"] = "migrations"; + cfg["jobs"]["enabled"] = true; + cfg["jobs"]["result_ttl"] = 3600; + cfg["jobs"]["max_retries"] = 2; + // Deliberately NOT setting outbox.drain_interval_sec: the default 0 + // must schedule nothing (opt-in contract) — every drain below is an + // explicit call, so a stray background tick would break the counts. + } + + // pick() needs a blocking client whose socket_timeout outlives the BRPOP + // argument (same rationale as test_jobs.cpp). + void post_init() override { + Jobs::get().init_blocking_client( + TestHelpers::redis_host(), TestHelpers::redis_port(), /*brpop_timeout_sec=*/5, /*password=*/""); + } + + void SetUp() override { + TestHelpers::CoreBackedTest::SetUp(); + if (::testing::Test::IsSkipped()) + return; + truncate_outbox(); + } + + void TearDown() override { + if (Database::is_initialized()) { + try { + truncate_outbox(); + } catch (...) {} + } + if (Cache::is_initialized()) + TestHelpers::drain_jobs({std::begin(kKinds), std::end(kKinds)}); + TestHelpers::CoreBackedTest::TearDown(); + } + + static void truncate_outbox() { + Database::get().execute_write([](auto& txn) { + txn.exec("TRUNCATE TABLE outbox"); + return 0; + }); + } + + static long outbox_count() { + return Database::get().execute_read_primary( + [](auto& txn) { return txn.exec("SELECT count(*) FROM outbox")[0][0].template as(); }); + } +}; + +TEST_F(OutboxTest, EnqueueCommitDrainSubmitsJob) { + Database::get().execute_transaction([](auto& txn) { + Jobs::Outbox::enqueue(txn, "outboxq", {{"n", 1}, {"who", "outbox"}}); + return 0; + }); + ASSERT_EQ(outbox_count(), 1); + + auto stats = Jobs::Outbox::drain(); + EXPECT_EQ(stats.submitted, 1); + EXPECT_EQ(stats.failed, 0); + // The relayed row is gone — a second drain must find nothing (no dup). + EXPECT_EQ(outbox_count(), 0); + auto again = Jobs::Outbox::drain(); + EXPECT_EQ(again.submitted, 0); + + // The job actually reached the Redis queue, payload intact. + auto picked = Jobs::get().pick({"outboxq"}, 2, "w1"); + ASSERT_TRUE(picked); + EXPECT_EQ(picked->type, "outboxq"); + EXPECT_EQ(picked->payload.at("n").get(), 1); + EXPECT_EQ(picked->payload.at("who").get(), "outbox"); + Jobs::get().complete(picked->id, {{"ok", true}}); +} + +TEST_F(OutboxTest, RolledBackEnqueueIsInvisible) { + // The atomicity that IS the pattern: the domain write and the event share + // one transaction, so a rollback erases both. + EXPECT_THROW(Database::get().execute_transaction([](auto& txn) -> int { + Jobs::Outbox::enqueue(txn, "outboxq_rb", {{"n", 2}}); + throw std::runtime_error("simulated domain-write failure after enqueue"); + }), + std::runtime_error); + + EXPECT_EQ(outbox_count(), 0); + auto stats = Jobs::Outbox::drain(); + EXPECT_EQ(stats.submitted, 0); + EXPECT_EQ(stats.failed, 0); + + // Nothing reached Redis either. + const auto depth = Jobs::get().queue_depth_by_type(); + EXPECT_EQ(depth.count("outboxq_rb") ? depth.at("outboxq_rb") : 0, 0); +} + +TEST_F(OutboxTest, FailedSubmitBumpsAttemptsAndLaterDrainDelivers) { + Database::get().execute_transaction([](auto& txn) { + Jobs::Outbox::enqueue(txn, "outboxq_fail", {{"n", 3}}); + return 0; + }); + + // Take the job queue down: submit now throws, which must be RECORDED on + // the row — never lost, never thrown out of drain. + Jobs::shutdown(); + auto s1 = Jobs::Outbox::drain(); + EXPECT_EQ(s1.submitted, 0); + EXPECT_EQ(s1.failed, 1); + + struct RowState { + int attempts; + std::string last_error; + bool released; + }; + auto state1 = Database::get().execute_read_primary([](auto& txn) { + auto r = txn.exec("SELECT attempts, last_error, claimed_at IS NULL AS released FROM outbox"); + return RowState{r[0]["attempts"].template as(), + r[0]["last_error"].template as(), + r[0]["released"].template as()}; + }); + EXPECT_EQ(state1.attempts, 1); + EXPECT_FALSE(state1.last_error.empty()); + EXPECT_TRUE(state1.released) << "a failed row must be released (claimed_at NULL) so the next drain retries it"; + + // Still down: the retry happens immediately on the next drain and keeps + // counting. + auto s2 = Jobs::Outbox::drain(); + EXPECT_EQ(s2.failed, 1); + auto attempts2 = Database::get().execute_read_primary( + [](auto& txn) { return txn.exec("SELECT attempts FROM outbox")[0][0].template as(); }); + EXPECT_EQ(attempts2, 2); + + // Queue back up → the event finally goes out. Crash-shaped outages heal + // the same way; nothing was lost meanwhile. + Jobs::initialize(/*result_ttl=*/3600, /*max_retries=*/2); + Jobs::get().init_blocking_client( + TestHelpers::redis_host(), TestHelpers::redis_port(), /*brpop_timeout_sec=*/5, /*password=*/""); + auto s3 = Jobs::Outbox::drain(); + EXPECT_EQ(s3.submitted, 1); + EXPECT_EQ(s3.failed, 0); + EXPECT_EQ(outbox_count(), 0); + + auto picked = Jobs::get().pick({"outboxq_fail"}, 2, "w1"); + ASSERT_TRUE(picked); + EXPECT_EQ(picked->payload.at("n").get(), 3); + Jobs::get().complete(picked->id, {{"ok", true}}); +} + +TEST_F(OutboxTest, ConcurrentDrainsSubmitEveryRowExactlyOnce) { + constexpr int kRows = 40; + Database::get().execute_transaction([](auto& txn) { + for (int i = 0; i < kRows; ++i) + Jobs::Outbox::enqueue(txn, "outboxq_conc", {{"n", i}}); + return 0; + }); + ASSERT_EQ(outbox_count(), kRows); + + const auto before = Jobs::get().queue_depth_by_type(); + const long base = before.count("outboxq_conc") ? before.at("outboxq_conc") : 0; + + // Two drainers race over the same backlog in small batches. SKIP LOCKED + // must make them partition it: no row claimed twice, none skipped forever. + std::atomic submitted{0}; + auto drainer = [&] { + for (int pass = 0; pass < 10; ++pass) { + auto stats = Jobs::Outbox::drain(/*batch=*/5); + submitted += stats.submitted; + EXPECT_EQ(stats.failed, 0); + } + }; + std::thread t1(drainer), t2(drainer); + t1.join(); + t2.join(); + + EXPECT_EQ(submitted.load(), kRows); + EXPECT_EQ(outbox_count(), 0); + + // Redis agrees: exactly kRows jobs queued — a duplicate claim would + // overshoot, a lost row would undershoot. + const auto after = Jobs::get().queue_depth_by_type(); + ASSERT_TRUE(after.count("outboxq_conc")); + EXPECT_EQ(after.at("outboxq_conc"), base + kRows); +} + +} // namespace