diff --git a/README.md b/README.md
index 9f466bb..ab3aed2 100644
--- a/README.md
+++ b/README.md
@@ -304,7 +304,29 @@ Throughput on a single instance (MacBook M3 Max, JDK 21 LTS, May 2026):
 | HTTP @ webhook latency 20 ms (sync sequential — parallel `sendAsync` planned) | ~38 msg/s | ~38 msg/s |
 | HTTP @ webhook latency 100 ms (sync sequential — parallel `sendAsync` planned) | ~9 msg/s | ~9 msg/s |
 
-Kafka throughput jumped 16-45× over the original sync-sequential baseline thanks to the `deliverBatch` fire-flush-await pattern. HTTP parallel `sendAsync` is next; multi-threaded scheduler scaling is in the roadmap.
+Kafka throughput jumped 16-45× over the original sync-sequential baseline thanks to the `deliverBatch` fire-flush-await pattern. HTTP parallel `sendAsync` is next.
+
+**Multi-threaded scheduler** (`OutboxSchedulerConfig.concurrency`, JDK 25, single Postgres+Kafka backend):
+
+| concurrency | speedup vs. concurrency=1 |
+|---|---|
+| 4  | **3.6×** |
+| 16 | **6.2×** |
+| 64 | **6.6×** (diminishing — see caveats below) |
+
+`concurrency=4` to `16` is the practical sweet spot — most of the available speedup is already
+captured there, with marginal gains beyond it on a single-instance backend. Default to platform
+threads (`workerExecutorFactory` default): in this benchmark's tested range (1-64 workers),
+switching to `virtualThreadPool` showed **no measurable advantage** over platform threads, even
+on a JEP 491 JDK (25) — contrary to the original hypothesis that virtual threads would win at
+concurrency=16+. Virtual threads only pay off when worker count vastly exceeds the platform pool
+size; at ≤64 workers there's no oversubscription for them to fix. Tuning rule of thumb:
+`concurrency × instances ≤ max_connections / 2` (row locks make cross-instance coordination free
+via `FOR UPDATE SKIP LOCKED`, but every worker holds a DB connection for its batch's duration).
+
+Full methodology, raw JMH results, before/after per change: [`benchmarks/`](benchmarks/), including
+[`results-postopt-KOJAK-77.md`](benchmarks/results-postopt-KOJAK-77.md) for the full concurrency
+breakdown and the reasoning behind the virtual-thread finding.
 
 ### MySQL: `rewriteBatchedStatements`
 
diff --git a/benchmarks/README.md b/benchmarks/README.md
index 5965ccf..6b110e6 100644
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -67,6 +67,14 @@ Single-entry `deliver()` calls with mocked I/O:
 Measures pure code overhead (JSON deserialization, record/request construction,
 exception classification). Useful as "did optimization X regress the hot path?" baseline.
 
+### Scheduler fan-out (`OutboxSchedulerConcurrencyBenchmark`)
+
+Drains a fixed backlog through `concurrency` parallel `processNext()` calls per round — one call
+per worker, each on its own transaction, mirroring `OutboxScheduler`'s internal fan-out (the
+scheduler's polling loop itself is still bypassed). `@Param executorType ∈ {platform, virtual}` ×
+`@Param concurrency ∈ {1, 4, 16, 64}`. Transport is Kafka `deliverBatch` — real Postgres + real
+Kafka via Testcontainers. See [`results-postopt-KOJAK-77.md`](results-postopt-KOJAK-77.md).
+
 ## How to read results
 
 Throughput benchmarks report **ops/s = msg/s** thanks to `@OperationsPerInvocation`.
@@ -96,8 +104,12 @@ investigate variability sources (background processes, thermal throttling, GC).
   is dominated by the target service's processing time + network — pick the value closest
   to your target. With `httpLatencyMs=100`, sequential delivery is bounded at
   `1000ms / 100ms = 10 msg/s/thread` regardless of library efficiency.
-- **Single-threaded scheduler.** Current `OutboxSchedulerConfig` does not expose `concurrency`.
-  Once that lands (planned), the throughput matrix will expand to `batchSize × concurrency`.
+- **Multi-threaded scheduler is single-instance, single-backend.** `OutboxSchedulerConcurrencyBenchmark`
+  fans out against one Postgres container and one Kafka broker; it measures the fan-out
+  mechanism's overhead, not unlimited horizontal DB/broker scaling. See
+  [`results-postopt-KOJAK-77.md`](results-postopt-KOJAK-77.md) for the concurrency=64 round-count
+  artifact (a single round drains the whole fixed backlog, so per-round fixed costs aren't
+  amortized the way they are at lower concurrency).
 
 ## Historical baselines
 
@@ -105,6 +117,7 @@ investigate variability sources (background processes, thermal throttling, GC).
   (sync sequential delivery, single-threaded scheduler)
 - [`results-postopt-KOJAK-75.md`](results-postopt-KOJAK-75.md) — batch UPDATE via JDBC
   `executeBatch()`, Postgres: 10.2×
+- [`results-postopt-KOJAK-77.md`](results-postopt-KOJAK-77.md) — multi-threaded scheduler concurrency knob
 - [`results-mysql-rewrite-batched-statements.md`](results-mysql-rewrite-batched-statements.md) —
   same optimization on MySQL: mechanism verified correct, but no net speedup measured at this
   batch size/hardware — read before assuming the Postgres multiplier transfers
diff --git a/benchmarks/results-postopt-KOJAK-77.md b/benchmarks/results-postopt-KOJAK-77.md
new file mode 100644
index 0000000..2611edc
--- /dev/null
+++ b/benchmarks/results-postopt-KOJAK-77.md
@@ -0,0 +1,104 @@
+# Multi-threaded scheduler with pluggable executor — Results (KOJAK-77)
+
+Measured on the same machine/config as [`results-kafka-deliverbatch.md`](results-kafka-deliverbatch.md):
+Postgres 16 + Kafka 3.8.1 via Testcontainers, full JMH config `fork=2, warmup=3 × 10s, iter=5 × 30s`
+(n=10 samples per benchmark). **JDK 25.0.2** this time (not the 21 LTS used for earlier docs in
+this directory) — notable because it means [JEP 491](https://openjdk.org/jeps/491) (removes
+`synchronized`-block carrier-thread pinning for virtual threads) is present; the "virtual threads
+lose to pinning on older JDKs" excuse does not apply to this run.
+
+`OutboxSchedulerConcurrencyBenchmark` drains a fixed backlog of 6,400 pending entries through
+`concurrency` parallel `OutboxProcessor.processNext(batchSize=100)` calls per round — one call per
+worker, each on its own transaction/connection, exactly mirroring `OutboxScheduler`'s internal
+fan-out (the scheduler's polling loop itself is bypassed, same rationale as the other throughput
+benchmarks in this directory). Kafka `deliverBatch` is the transport, since HTTP `deliverBatch`
+(KOJAK-74) isn't implemented yet and this benchmark is meant to isolate fan-out scaling, not
+transport I/O.
+
+## Headline numbers
+
+| concurrency | platform (ms/op) | virtual (ms/op) | platform msg/s | virtual msg/s | virtual vs platform |
+|---|---|---|---|---|---|
+| 1  | 0.357 ± 0.097 | 0.347 ± 0.070 | ~2,801 | ~2,882 | +3% (noise — CIs overlap heavily) |
+| 4  | 0.099 ± 0.003 | 0.104 ± 0.003 | ~10,101 | ~9,615 | **−5%** |
+| 16 | 0.058 ± 0.005 | 0.058 ± 0.002 | ~17,241 | ~17,241 | 0% (tied) |
+| 64 | 0.054 ± 0.002 | 0.058 ± 0.003 | ~18,519 | ~17,241 | **−7%** |
+
+Raw JSON: [`scheduler-concurrency.json`](scheduler-concurrency.json).
+
+Speedup vs. `concurrency=1, platform` baseline:
+
+| concurrency | platform speedup | virtual speedup |
+|---|---|---|
+| 4  | **3.6×** | 3.4× |
+| 16 | **6.2×** | 6.2× |
+| 64 | **6.6×** | 6.2× |
+
+## The ticket's hypothesis did not hold — reporting it straight
+
+The ticket's baseline expectations were:
+- `concurrency=4, platform`: target ~3-4× single-threaded — **confirmed** (3.6×).
+- `concurrency=16, virtual` should outperform `concurrency=16, platform` by 15-25% — **not
+  observed**. They're tied (0% difference, well within error bars).
+- `concurrency=64+, virtual` should be "the only viable option" — **not observed**. Platform is
+  actually ~7% *faster* than virtual at concurrency=64.
+
+At no tested concurrency level does virtual meaningfully beat platform; platform is same-or-ahead
+throughout. Since this run already has JEP 491 (JDK 25), the usual "virtual threads regress
+because of pre-JDK-24 pinning" explanation doesn't apply here. The more likely explanation:
+virtual threads' advantage shows up when task count vastly exceeds available platform threads
+(hundreds to thousands of concurrent blocking tasks contending for a small carrier pool) — this
+benchmark only ever runs *exactly* `concurrency` tasks against a platform pool sized to exactly
+`concurrency`, so there's no oversubscription for virtual threads to fix, and they pay their own
+(small) continuation-management overhead for no return. The ticket's "64+" ceiling likely needs to
+be an order of magnitude higher (hundreds to low thousands) before virtual threads' benefit would
+show up — genuinely testing that means decoupling worker count from DB connection limits, which
+is out of scope here.
+
+**Practical takeaway: default to platform threads (`OutboxSchedulerConfig`'s default) at every
+concurrency level tested (1-64).** `workerExecutorFactory` remains available for users who want to
+verify virtual threads at concurrency well beyond 64 for their own workload, but nothing in this
+data recommends switching the default.
+
+## Where the scaling actually plateaus
+
+Sublinear scaling from `concurrency=4` (3.6×) to `concurrency=16` (6.2×) to `concurrency=64`
+(6.6×) — far from the linear 4×/16×/64× the "multiplies throughput linearly" framing in the ticket
+might suggest. Two contributing factors:
+- **Shared, single-instance backend.** One Postgres container and one Kafka broker mean workers
+  increasingly contend for the same connection acceptance path and broker I/O as concurrency
+  rises — this benchmark measures the fan-out mechanism's overhead, not unlimited horizontal DB
+  scaling, and the "concurrency × instances ≤ `max_connections` / 2" rule of thumb below exists
+  precisely because of this.
+- **Round-count artifact at high concurrency.** With `batchSize=100` fixed, `concurrency=64` claims
+  all 6,400 entries in a *single* round — so its measurement is dominated by one-time per-round
+  costs (opening `concurrency` fresh connections, `claimPending` query planning) with no further
+  rounds to amortize them over. Lower concurrency values run many rounds, averaging out that fixed
+  cost. This means the `concurrency=64` number is a slight pessimistic outlier relative to a
+  steady-state, many-round production workload — a caveat worth keeping in mind rather than reading
+  64 as a hard ceiling.
+
+## Tuning recommendation
+
+- **`concurrency=4` to `concurrency=16`** is the practical sweet spot for a single Postgres+Kafka
+  backend on this hardware: most of the available speedup (3.6×-6.2×) is already captured, and
+  gains beyond 16 are marginal (6.2× → 6.6×) while consuming more DB connections.
+- Keep the default `defaultPlatformPool` executor. Nothing in this data supports switching to
+  `virtualThreadPool` at the concurrency levels tested.
+- **Rule of thumb documented on `OutboxSchedulerConfig`:** `concurrency × instances ≤
+  max_connections / 2` — cross-instance coordination is free (`FOR UPDATE SKIP LOCKED` handles
+  disjoint claims across both workers and instances with no app-level coordination), but every
+  worker across every instance holds a DB connection for the duration of its batch.
+
+## Verification context
+
+- Unit tests: `OutboxSchedulerConfigTest` (concurrency validation 1-256, `defaultPlatformPool`
+  thread naming, `virtualThreadPool` runs on virtual threads), `OutboxSchedulerTest` (fan-out
+  dispatches exactly `concurrency` workers per tick, `concurrency=1` never touches
+  `workerExecutorFactory`, one worker's exception doesn't block the others).
+- Integration test: `ConcurrentClaimTests` extended with a real `OutboxScheduler`
+  (`concurrency=4`) against both Postgres and MySQL — disjoint claims across workers *and* ticks,
+  zero delivery amplification.
+- `okapi-spring-boot`: `okapi.processor.concurrency` property wired through, with startup-failure
+  coverage for invalid values.
+- ktlint clean, full `./gradlew build` green (core, spring-boot, integration-tests).
diff --git a/benchmarks/scheduler-concurrency.json b/benchmarks/scheduler-concurrency.json
new file mode 100644
index 0000000..d29d8d0
--- /dev/null
+++ b/benchmarks/scheduler-concurrency.json
@@ -0,0 +1,532 @@
+[
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "1",
+            "executorType" : "platform"
+        },
+        "primaryMetric" : {
+            "score" : 0.3566704235731962,
+            "scoreError" : 0.0974459790021155,
+            "scoreConfidence" : [
+                0.25922444457108074,
+                0.4541164025753117
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.26312747039772727,
+                "50.0" : 0.3689720704123264,
+                "90.0" : 0.45276361384374997,
+                "95.0" : 0.45603878162946426,
+                "99.0" : 0.45603878162946426,
+                "99.9" : 0.45603878162946426,
+                "99.99" : 0.45603878162946426,
+                "99.999" : 0.45603878162946426,
+                "99.9999" : 0.45603878162946426,
+                "100.0" : 0.45603878162946426
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.45603878162946426,
+                    0.356075834140625,
+                    0.3919755042013889,
+                    0.4232871037723214,
+                    0.3602697251215278
+                ],
+                [
+                    0.37839303164930554,
+                    0.377674415703125,
+                    0.29365464715625,
+                    0.26312747039772727,
+                    0.2662077219602273
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "1",
+            "executorType" : "virtual"
+        },
+        "primaryMetric" : {
+            "score" : 0.34720389351516,
+            "scoreError" : 0.06952627072990189,
+            "scoreConfidence" : [
+                0.2776776227852581,
+                0.41673016424506193
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.30811317578125,
+                "50.0" : 0.33832400247395833,
+                "90.0" : 0.4573723407770648,
+                "95.0" : 0.4672377064732143,
+                "99.0" : 0.4672377064732143,
+                "99.9" : 0.4672377064732143,
+                "99.99" : 0.4672377064732143,
+                "99.999" : 0.4672377064732143,
+                "99.9999" : 0.4672377064732143,
+                "100.0" : 0.4672377064732143
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.33815907192708333,
+                    0.36858404951171875,
+                    0.3288113462152778,
+                    0.3421425101215278,
+                    0.4672377064732143
+                ],
+                [
+                    0.3099344863125,
+                    0.30811317578125,
+                    0.3203615924375,
+                    0.35020606335069443,
+                    0.3384889330208333
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "4",
+            "executorType" : "platform"
+        },
+        "primaryMetric" : {
+            "score" : 0.09862723203400736,
+            "scoreError" : 0.002815739354474301,
+            "scoreConfidence" : [
+                0.09581149267953307,
+                0.10144297138848166
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.09610727442095589,
+                "50.0" : 0.09840320409007353,
+                "90.0" : 0.1019284712601103,
+                "95.0" : 0.10213047217830883,
+                "99.0" : 0.10213047217830883,
+                "99.9" : 0.10213047217830883,
+                "99.99" : 0.10213047217830883,
+                "99.999" : 0.10213047217830883,
+                "99.9999" : 0.10213047217830883,
+                "100.0" : 0.10213047217830883
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.09976032856617648,
+                    0.09823047260110294,
+                    0.09805603017463235,
+                    0.09751775696691177,
+                    0.09611523820772058
+                ],
+                [
+                    0.10213047217830883,
+                    0.10011046299632353,
+                    0.09966834864889706,
+                    0.09857593557904412,
+                    0.09610727442095589
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "4",
+            "executorType" : "virtual"
+        },
+        "primaryMetric" : {
+            "score" : 0.10397616305457262,
+            "scoreError" : 0.002796758328300158,
+            "scoreConfidence" : [
+                0.10117940472627246,
+                0.10677292138287278
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.10130346124080883,
+                "50.0" : 0.10362827512408088,
+                "90.0" : 0.10692789718307676,
+                "95.0" : 0.10698128563419118,
+                "99.0" : 0.10698128563419118,
+                "99.9" : 0.10698128563419118,
+                "99.99" : 0.10698128563419118,
+                "99.999" : 0.10698128563419118,
+                "99.9999" : 0.10698128563419118,
+                "100.0" : 0.10698128563419118
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.10644740112304688,
+                    0.10289676816176471,
+                    0.10356370519301471,
+                    0.10539989581054687,
+                    0.10177409924632352
+                ],
+                [
+                    0.10698128563419118,
+                    0.10353915592830883,
+                    0.10369284505514706,
+                    0.10416301315257354,
+                    0.10130346124080883
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "16",
+            "executorType" : "platform"
+        },
+        "primaryMetric" : {
+            "score" : 0.05831040687920025,
+            "scoreError" : 0.005466810140024821,
+            "scoreConfidence" : [
+                0.05284359673917543,
+                0.06377721701922506
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.0537906276328125,
+                "50.0" : 0.057420186859786185,
+                "90.0" : 0.06411888988465073,
+                "95.0" : 0.06418689417534722,
+                "99.0" : 0.06418689417534722,
+                "99.9" : 0.06418689417534722,
+                "99.99" : 0.06418689417534722,
+                "99.999" : 0.06418689417534722,
+                "99.9999" : 0.06418689417534722,
+                "100.0" : 0.06418689417534722
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.0554868720703125,
+                    0.055249909484375,
+                    0.0554496171640625,
+                    0.05699982454769737,
+                    0.0537906276328125
+                ],
+                [
+                    0.06072154741776316,
+                    0.06350685126838235,
+                    0.06418689417534722,
+                    0.057840549171875,
+                    0.059871375859375
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "16",
+            "executorType" : "virtual"
+        },
+        "primaryMetric" : {
+            "score" : 0.05829309122861842,
+            "scoreError" : 0.0020862941345980135,
+            "scoreConfidence" : [
+                0.056206797094020405,
+                0.06037938536321644
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.0567883317109375,
+                "50.0" : 0.057635347,
+                "90.0" : 0.061058549313322365,
+                "95.0" : 0.0611839275,
+                "99.0" : 0.0611839275,
+                "99.9" : 0.0611839275,
+                "99.99" : 0.0611839275,
+                "99.999" : 0.0611839275,
+                "99.9999" : 0.0611839275,
+                "100.0" : 0.0611839275
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.057609984046875,
+                    0.0611839275,
+                    0.05911475393914474,
+                    0.0574009990078125,
+                    0.0567883317109375
+                ],
+                [
+                    0.059930145633223686,
+                    0.0583794749453125,
+                    0.057660709953125,
+                    0.0573724303359375,
+                    0.05749015521381579
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "64",
+            "executorType" : "platform"
+        },
+        "primaryMetric" : {
+            "score" : 0.05417698885355184,
+            "scoreError" : 0.002214712391078403,
+            "scoreConfidence" : [
+                0.051962276462473436,
+                0.05639170124463025
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.05227146646875,
+                "50.0" : 0.053629043937499996,
+                "90.0" : 0.05614647789638158,
+                "95.0" : 0.056150869309210524,
+                "99.0" : 0.056150869309210524,
+                "99.9" : 0.056150869309210524,
+                "99.99" : 0.056150869309210524,
+                "99.999" : 0.056150869309210524,
+                "99.9999" : 0.056150869309210524,
+                "100.0" : 0.056150869309210524
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.054760251625,
+                    0.05347714159375,
+                    0.053248665359375,
+                    0.05227146646875,
+                    0.0526747985234375
+                ],
+                [
+                    0.05378094628125,
+                    0.0559741119765625,
+                    0.05332468221726191,
+                    0.056106955180921056,
+                    0.056150869309210524
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.37",
+        "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 2,
+        "jvm" : "25.0.2-open",
+        "jvmArgs" : [
+            "-Xms8g",
+            "-Xmx8g",
+            "-XX:+UseG1GC",
+            "-Dliquibase.duplicateFileMode=WARN"
+        ],
+        "jdkVersion" : "25.0.2",
+        "vmName" : "OpenJDK 64-Bit Server VM",
+        "vmVersion" : "25.0.2+10-69",
+        "warmupIterations" : 3,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "30 s",
+        "measurementBatchSize" : 1,
+        "params" : {
+            "concurrency" : "64",
+            "executorType" : "virtual"
+        },
+        "primaryMetric" : {
+            "score" : 0.05779675429407895,
+            "scoreError" : 0.0025192637530646158,
+            "scoreConfidence" : [
+                0.05527749054101433,
+                0.060316018047143566
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.0557219973984375,
+                "50.0" : 0.05757087195867598,
+                "90.0" : 0.06081344351475694,
+                "95.0" : 0.060930749071180554,
+                "99.0" : 0.060930749071180554,
+                "99.9" : 0.060930749071180554,
+                "99.99" : 0.060930749071180554,
+                "99.999" : 0.060930749071180554,
+                "99.9999" : 0.060930749071180554,
+                "100.0" : 0.060930749071180554
+            },
+            "scoreUnit" : "ms/op",
+            "rawData" : [
+                [
+                    0.059757693506944445,
+                    0.056381873390625,
+                    0.056691587875,
+                    0.0557219973984375,
+                    0.056412676765625
+                ],
+                [
+                    0.059110762359375,
+                    0.060930749071180554,
+                    0.0577293069765625,
+                    0.05781845865625,
+                    0.05741243694078947
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    }
+]
+
+
diff --git a/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt b/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt
new file mode 100644
index 0000000..fca290f
--- /dev/null
+++ b/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt
@@ -0,0 +1,137 @@
+package com.softwaremill.okapi.benchmarks
+
+import com.softwaremill.okapi.benchmarks.support.KafkaBenchmarkSupport
+import com.softwaremill.okapi.benchmarks.support.PostgresBenchmarkSupport
+import com.softwaremill.okapi.core.OutboxEntryProcessor
+import com.softwaremill.okapi.core.OutboxMessage
+import com.softwaremill.okapi.core.OutboxProcessor
+import com.softwaremill.okapi.core.OutboxPublisher
+import com.softwaremill.okapi.core.OutboxSchedulerConfig
+import com.softwaremill.okapi.core.RetryPolicy
+import com.softwaremill.okapi.kafka.KafkaDeliveryInfo
+import com.softwaremill.okapi.kafka.KafkaMessageDeliverer
+import com.softwaremill.okapi.postgres.PostgresOutboxStore
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.openjdk.jmh.annotations.Benchmark
+import org.openjdk.jmh.annotations.BenchmarkMode
+import org.openjdk.jmh.annotations.Level
+import org.openjdk.jmh.annotations.Mode
+import org.openjdk.jmh.annotations.OperationsPerInvocation
+import org.openjdk.jmh.annotations.OutputTimeUnit
+import org.openjdk.jmh.annotations.Param
+import org.openjdk.jmh.annotations.Scope
+import org.openjdk.jmh.annotations.Setup
+import org.openjdk.jmh.annotations.State
+import org.openjdk.jmh.annotations.TearDown
+import java.time.Clock
+import java.util.UUID
+import java.util.concurrent.Callable
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.TimeUnit
+
+/**
+ * Measures scheduler fan-out throughput (KOJAK-77): each round dispatches [concurrency]
+ * parallel [OutboxProcessor.processNext] calls -- one per worker, each on its own JDBC
+ * transaction/connection, exactly mirroring what [com.softwaremill.okapi.core.OutboxScheduler]'s
+ * internal fan-out does -- and finds the platform-vs-virtual-thread breakeven point.
+ *
+ * The scheduler's [java.util.concurrent.ScheduledExecutorService] polling loop is bypassed
+ * (same rationale as [KafkaThroughputBenchmark]): we measure worker-pool throughput, not
+ * polling cadence. Kafka `deliverBatch` is the transport, since it's the one delivery path
+ * fast enough that the fan-out itself -- not delivery I/O -- is the bottleneck being measured
+ * (HTTP `deliverBatch` isn't implemented yet; KOJAK-74).
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+open class OutboxSchedulerConcurrencyBenchmark {
+
+    @Param("platform", "virtual")
+    var executorType: String = "platform"
+
+    @Param("1", "4", "16", "64")
+    var concurrency: Int = 1
+
+    private lateinit var postgres: PostgresBenchmarkSupport
+    private lateinit var kafka: KafkaBenchmarkSupport
+    private lateinit var producer: KafkaProducer<String, String>
+    private lateinit var publisher: OutboxPublisher
+    private lateinit var processor: OutboxProcessor
+    private lateinit var executor: ExecutorService
+    private lateinit var topic: String
+
+    @Setup(Level.Trial)
+    fun setupTrial() {
+        postgres = PostgresBenchmarkSupport().also { it.start() }
+        kafka = KafkaBenchmarkSupport().also { it.start() }
+        producer = kafka.createProducer()
+        topic = "bench-${UUID.randomUUID()}"
+
+        val clock = Clock.systemUTC()
+        val store = PostgresOutboxStore(postgres.jdbc)
+        publisher = OutboxPublisher(store, clock)
+        val deliverer = KafkaMessageDeliverer(producer)
+        val entryProcessor = OutboxEntryProcessor(deliverer, RetryPolicy(maxRetries = 0), clock)
+        processor = OutboxProcessor(store, entryProcessor)
+
+        executor = when (executorType) {
+            "virtual" -> OutboxSchedulerConfig.virtualThreadPool(concurrency)
+            else -> OutboxSchedulerConfig.defaultPlatformPool(concurrency)
+        }
+    }
+
+    @Setup(Level.Invocation)
+    fun setupInvocation() {
+        postgres.truncate()
+        val info = KafkaDeliveryInfo(topic = topic, partitionKey = "k")
+        postgres.jdbc.withTransaction {
+            repeat(TOTAL_ENTRIES) {
+                publisher.publish(OutboxMessage("bench.event", PAYLOAD), info)
+            }
+        }
+    }
+
+    @Benchmark
+    @OperationsPerInvocation(TOTAL_ENTRIES)
+    fun drainAll() {
+        var remaining = TOTAL_ENTRIES
+        while (remaining > 0) {
+            val futures = (1..concurrency).map {
+                executor.submit(Callable { postgres.jdbc.withTransaction { processor.processNext(BATCH_SIZE) } })
+            }
+            val processedThisRound = futures.sumOf { it.get() }
+            // A stalled round must fail loudly, not `break` silently -- @OperationsPerInvocation(TOTAL_ENTRIES)
+            // would otherwise have JMH divide elapsed time by a count of entries that were never actually
+            // processed, silently under-reporting the per-op cost.
+            check(processedThisRound > 0) {
+                "drainAll() stalled with $remaining/$TOTAL_ENTRIES entries left unprocessed " +
+                    "(concurrency=$concurrency, batchSize=$BATCH_SIZE, executorType=$executorType)"
+            }
+            remaining -= processedThisRound
+        }
+        check(remaining == 0) { "drainAll() finished with $remaining/$TOTAL_ENTRIES entries still unprocessed" }
+    }
+
+    @TearDown(Level.Trial)
+    fun teardown() {
+        executor.shutdown()
+        try {
+            if (!executor.awaitTermination(EXECUTOR_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                executor.shutdownNow()
+            }
+        } catch (e: InterruptedException) {
+            Thread.currentThread().interrupt()
+            executor.shutdownNow()
+        }
+        producer.close()
+        kafka.stop()
+        postgres.stop()
+    }
+
+    companion object {
+        const val TOTAL_ENTRIES = 6400
+        const val BATCH_SIZE = 100
+        private const val PAYLOAD = """{"orderId":"order-42","amount":100.50,"currency":"EUR"}"""
+        private const val EXECUTOR_TERMINATION_TIMEOUT_SECONDS = 5L
+    }
+}
diff --git a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt
index f2bbb2e..0f427e8 100644
--- a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt
+++ b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt
@@ -1,7 +1,12 @@
 package com.softwaremill.okapi.core
 
 import org.slf4j.LoggerFactory
+import java.util.concurrent.CancellationException
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
+import java.util.concurrent.Future
+import java.util.concurrent.RejectedExecutionException
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
@@ -9,9 +14,20 @@ import java.util.concurrent.atomic.AtomicBoolean
 /**
  * Standalone scheduler that periodically calls [OutboxProcessor.processNext].
  *
- * Each tick runs inside [transactionRunner]. The runner is required: without a surrounding
- * transaction, `FOR UPDATE SKIP LOCKED` releases its row lock at the end of the claim
- * SELECT and concurrent processor instances deliver the same entry multiple times.
+ * Each worker's call runs inside [transactionRunner]. The runner is required: without a
+ * surrounding transaction, `FOR UPDATE SKIP LOCKED` releases its row lock at the end of the
+ * claim SELECT and concurrent processor instances deliver the same entry multiple times.
+ *
+ * With [OutboxSchedulerConfig.concurrency] `> 1`, each tick fans out to that many workers on
+ * [OutboxSchedulerConfig.workerExecutorFactory]'s pool. `FOR UPDATE SKIP LOCKED` guarantees
+ * disjoint claims, but [outboxProcessor] (and its dependencies) must be safe for concurrent use.
+ * The tick waits for every worker before the next scheduled interval, so ticks never overlap.
+ * `concurrency = 1` (default) calls the batch inline, preserving single-worker semantics.
+ *
+ * At `concurrency > 1`, [outboxProcessor] -- and by extension its [OutboxStore], [MessageDeliverer],
+ * and any [OutboxProcessorListener] -- is invoked concurrently from multiple worker threads, so all
+ * of these must be safe for concurrent use. The store/deliverer implementations shipped in this
+ * project already are; a custom implementation must be too.
  *
  * Runs on a single daemon thread with explicit [start]/[stop] lifecycle.
  * [start] and [stop] are single-use -- the internal executor cannot be restarted after shutdown.
@@ -33,25 +49,135 @@ class OutboxScheduler @JvmOverloads constructor(
             Thread(r, "outbox-processor").apply { isDaemon = true }
         }
 
+    // Spun up lazily, only if concurrency > 1 -- accessed from both the scheduler thread (tick())
+    // and the caller thread (stop()), so this must use the default thread-safe SYNCHRONIZED mode.
+    // concurrency == 1 never touches this at all (see tick()), so it's never initialized then.
+    private val workers: Lazy<ExecutorService> = lazy { config.workerExecutorFactory(config.concurrency) }
+
     fun start() {
         check(!scheduler.isShutdown) { "OutboxScheduler cannot be restarted after stop()" }
         if (!running.compareAndSet(false, true)) return
-        logger.info("Outbox processor started [interval={}, batchSize={}]", config.interval, config.batchSize)
+        logger.info(
+            "Outbox processor started [interval={}, batchSize={}, concurrency={}]",
+            config.interval,
+            config.batchSize,
+            config.concurrency,
+        )
         scheduler.scheduleWithFixedDelay(::tick, 0L, config.interval.toMillis(), TimeUnit.MILLISECONDS)
     }
 
     fun stop() {
         if (!running.compareAndSet(true, false)) return
-        scheduler.shutdown()
-        if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
-            scheduler.shutdownNow()
+        shutdownAndAwait(scheduler)
+        if (workers.isInitialized()) {
+            shutdownAndAwait(workers.value)
         }
         logger.info("Outbox processor stopped")
     }
 
     fun isRunning(): Boolean = running.get()
 
+    /**
+     * Shuts down [executor] and waits up to [SHUTDOWN_TIMEOUT_SECONDS] for in-flight tasks to
+     * finish. If that times out, `shutdownNow()` forces it, then waits once more (same budget) so
+     * `stop()` doesn't return while a custom [OutboxSchedulerConfig.workerExecutorFactory]'s
+     * (potentially non-daemon) threads are still winding down -- logging if they still haven't by
+     * then, since nothing more can be done short of abandoning the executor. If the caller is
+     * interrupted at any point in this sequence, the flag is restored and [executor] is
+     * force-shut-down via `shutdownNow()` without a further wait, so `stop()` still completes
+     * deterministically and promptly honors the interrupt instead of leaking a partially-shut-down
+     * executor.
+     */
+    private fun shutdownAndAwait(executor: ExecutorService) {
+        executor.shutdown()
+        try {
+            if (executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) return
+            executor.shutdownNow()
+            if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+                logger.warn(
+                    "Executor (${executor.javaClass.name}) did not terminate after shutdownNow(); some worker threads may still be running",
+                )
+            }
+        } catch (e: InterruptedException) {
+            Thread.currentThread().interrupt()
+            logger.warn("Interrupted while awaiting executor (${executor.javaClass.name}) termination; forcing shutdownNow()", e)
+            executor.shutdownNow()
+        }
+    }
+
     private fun tick() {
+        if (config.concurrency == 1) {
+            // Skips workers/Lazy entirely -- not just an inline no-op path -- so this configuration
+            // truly pays no executor/synchronization overhead per tick, matching the class KDoc.
+            processBatch()
+            return
+        }
+        val pool = try {
+            workers.value
+        } catch (e: Exception) {
+            // workers.value runs OutboxSchedulerConfig.workerExecutorFactory on first access; a
+            // custom factory can fail (e.g. resource exhaustion). Uncaught, that would escape tick()
+            // and make scheduleWithFixedDelay suppress all future ticks. Lazy retries its
+            // initializer on a failed attempt, so the next tick tries workerExecutorFactory again.
+            // Note: this does not prevent an OutOfMemoryError if OS thread creation is exhausted --
+            // that's an Error, not an Exception, and is intentionally left uncaught here.
+            logger.error("Failed to initialize outbox worker pool, will retry at next scheduled interval", e)
+            return
+        }
+        val futures = (1..config.concurrency).mapNotNull { submitWorker(pool) }
+        futures.forEach(::awaitWorker)
+    }
+
+    /**
+     * Submits one worker's [processBatch] without letting [RejectedExecutionException] escape
+     * [tick] -- a custom [OutboxSchedulerConfig.workerExecutorFactory] may return an executor that
+     * can reject submissions (bounded queue, shutdown executor), and an uncaught exception here
+     * would make `scheduleWithFixedDelay` suppress all future ticks.
+     */
+    private fun submitWorker(pool: ExecutorService): Future<*>? = try {
+        pool.submit(::processBatch)
+    } catch (e: RejectedExecutionException) {
+        logger.error("Outbox worker submission rejected, will retry at next scheduled interval", e)
+        null
+    }
+
+    /**
+     * Awaits one worker's completion without letting anything escape [tick] -- an uncaught
+     * exception here would make `scheduleWithFixedDelay` suppress all future ticks, silently
+     * killing the scheduler. [processBatch] already catches [Exception] internally, so this
+     * only guards against an [Error] surfacing via [ExecutionException], a [CancellationException]
+     * if a custom [OutboxSchedulerConfig.workerExecutorFactory] cancels queued/running tasks (e.g.
+     * as a backpressure strategy), or an interrupt while awaiting.
+     *
+     * An [InterruptedException] only aborts the wait when [running] is already `false` -- i.e.
+     * when [stop] itself triggered the interrupt via `scheduler.shutdownNow()`. Any other
+     * interrupt is swallowed and the wait resumes, because bailing out early here would let the
+     * next scheduled tick start while this worker is still running, violating the class-level
+     * guarantee that ticks never overlap.
+     */
+    private fun awaitWorker(future: Future<*>) {
+        while (true) {
+            try {
+                future.get()
+                return
+            } catch (e: InterruptedException) {
+                if (!running.get()) {
+                    Thread.currentThread().interrupt()
+                    logger.warn("Interrupted while awaiting outbox worker during shutdown", e)
+                    return
+                }
+                // Not shutting down: keep waiting instead of letting the next tick overlap this worker.
+            } catch (e: ExecutionException) {
+                logger.error("Outbox worker failed unexpectedly, will retry at next scheduled interval", e.cause ?: e)
+                return
+            } catch (e: CancellationException) {
+                logger.error("Outbox worker was cancelled, will retry at next scheduled interval", e)
+                return
+            }
+        }
+    }
+
+    private fun processBatch() {
         try {
             transactionRunner.runInTransaction { outboxProcessor.processNext(config.batchSize) }
             logger.debug("Outbox processor tick completed [batchSize={}]", config.batchSize)
@@ -62,5 +188,6 @@ class OutboxScheduler @JvmOverloads constructor(
 
     companion object {
         private val logger = LoggerFactory.getLogger(OutboxScheduler::class.java)
+        private const val SHUTDOWN_TIMEOUT_SECONDS = 5L
     }
 }
diff --git a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt
index 5b6c41f..5a9d812 100644
--- a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt
+++ b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt
@@ -1,13 +1,57 @@
 package com.softwaremill.okapi.core
 
 import java.time.Duration
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
 
-data class OutboxSchedulerConfig(
+data class OutboxSchedulerConfig @JvmOverloads constructor(
     val interval: Duration = Duration.ofSeconds(1),
     val batchSize: Int = 10,
+    /**
+     * Number of parallel workers fanned out per scheduler tick. Each worker claims its own
+     * batch via [OutboxStore.claimPending]'s `FOR UPDATE SKIP LOCKED`, so workers never
+     * compete for the same rows. `1` (default) preserves the original single-worker tick
+     * with zero pooling overhead.
+     */
+    val concurrency: Int = 1,
+    /**
+     * Builds the [ExecutorService] used to run workers when [concurrency] > 1. Defaults to
+     * a fixed platform-thread pool ([defaultPlatformPool]). [virtualThreadPool] is available
+     * for workloads at concurrency well beyond what's benchmarked here; at the concurrency
+     * levels actually measured (up to 64, see `benchmarks/results-postopt-KOJAK-77.md`)
+     * platform threads were tied-or-ahead throughout, so nothing in that data recommends
+     * switching away from the default.
+     */
+    val workerExecutorFactory: (Int) -> ExecutorService = ::defaultPlatformPool,
 ) {
     init {
         require(!interval.isNegative && interval.toMillis() > 0) { "interval must be at least 1ms, got: $interval" }
         require(batchSize > 0) { "batchSize must be positive, got: $batchSize" }
+        require(concurrency in 1..256) { "concurrency must be between 1 and 256, got: $concurrency" }
+    }
+
+    companion object {
+        /** Fixed pool of [n] daemon platform threads, named `outbox-worker-N`. */
+        @JvmStatic
+        fun defaultPlatformPool(n: Int): ExecutorService {
+            require(n > 0) { "n must be positive, got: $n" }
+            val counter = AtomicInteger(0)
+            return Executors.newFixedThreadPool(n) { r ->
+                Thread(r, "outbox-worker-${counter.incrementAndGet()}").apply { isDaemon = true }
+            }
+        }
+
+        /**
+         * One virtual thread per submitted task. [n] is accepted for signature parity with
+         * [defaultPlatformPool] but otherwise unused — virtual thread creation is unbounded --
+         * still validated so a bad `concurrency` value fails the same way for either factory.
+         */
+        @JvmStatic
+        @Suppress("UNUSED_PARAMETER")
+        fun virtualThreadPool(n: Int): ExecutorService {
+            require(n > 0) { "n must be positive, got: $n" }
+            return Executors.newVirtualThreadPerTaskExecutor()
+        }
     }
 }
diff --git a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt
index da30e6d..4bb2c99 100644
--- a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt
+++ b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt
@@ -7,6 +7,8 @@ import java.time.Duration
 import java.time.Duration.ofMillis
 import java.time.Duration.ofNanos
 import java.time.Duration.ofSeconds
+import java.util.concurrent.Callable
+import java.util.concurrent.TimeUnit
 
 class OutboxSchedulerConfigTest : FunSpec({
 
@@ -14,12 +16,14 @@ class OutboxSchedulerConfigTest : FunSpec({
         val config = OutboxSchedulerConfig()
         config.interval shouldBe ofSeconds(1)
         config.batchSize shouldBe 10
+        config.concurrency shouldBe 1
     }
 
     test("accepts custom valid values") {
-        val config = OutboxSchedulerConfig(interval = ofMillis(500), batchSize = 50)
+        val config = OutboxSchedulerConfig(interval = ofMillis(500), batchSize = 50, concurrency = 8)
         config.interval shouldBe ofMillis(500)
         config.batchSize shouldBe 50
+        config.concurrency shouldBe 8
     }
 
     test("rejects zero interval") {
@@ -51,4 +55,66 @@ class OutboxSchedulerConfigTest : FunSpec({
             OutboxSchedulerConfig(interval = ofNanos(1))
         }
     }
+
+    test("accepts concurrency boundaries 1 and 256") {
+        OutboxSchedulerConfig(concurrency = 1).concurrency shouldBe 1
+        OutboxSchedulerConfig(concurrency = 256).concurrency shouldBe 256
+    }
+
+    test("rejects zero concurrency") {
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig(concurrency = 0)
+        }
+    }
+
+    test("rejects negative concurrency") {
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig(concurrency = -1)
+        }
+    }
+
+    test("rejects concurrency above 256") {
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig(concurrency = 257)
+        }
+    }
+
+    test("defaultPlatformPool runs submitted tasks and names its threads outbox-worker-N") {
+        val pool = OutboxSchedulerConfig.defaultPlatformPool(2)
+        try {
+            val threadNames = (1..2).map { pool.submit(Callable { Thread.currentThread().name }) }
+                .map { it.get(2, TimeUnit.SECONDS) }
+            threadNames.toSet() shouldBe setOf("outbox-worker-1", "outbox-worker-2")
+        } finally {
+            pool.shutdown()
+        }
+    }
+
+    test("defaultPlatformPool rejects zero or negative n") {
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig.defaultPlatformPool(0)
+        }
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig.defaultPlatformPool(-1)
+        }
+    }
+
+    test("virtualThreadPool runs submitted tasks on virtual threads") {
+        val pool = OutboxSchedulerConfig.virtualThreadPool(4)
+        try {
+            val isVirtual = pool.submit(Callable { Thread.currentThread().isVirtual }).get(2, TimeUnit.SECONDS)
+            isVirtual shouldBe true
+        } finally {
+            pool.shutdown()
+        }
+    }
+
+    test("virtualThreadPool rejects zero or negative n") {
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig.virtualThreadPool(0)
+        }
+        shouldThrow<IllegalArgumentException> {
+            OutboxSchedulerConfig.virtualThreadPool(-1)
+        }
+    }
 })
diff --git a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt
index 98af3ee..84543e2 100644
--- a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt
+++ b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt
@@ -1,14 +1,25 @@
 package com.softwaremill.okapi.core
 
+import ch.qos.logback.classic.Level
+import ch.qos.logback.classic.Logger
+import ch.qos.logback.classic.spi.ILoggingEvent
+import ch.qos.logback.core.read.ListAppender
 import io.kotest.assertions.throwables.shouldThrow
 import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.comparables.shouldBeGreaterThanOrEqualTo
 import io.kotest.matchers.shouldBe
+import org.slf4j.LoggerFactory
 import java.time.Duration.ofMillis
 import java.time.Duration.ofMinutes
+import java.util.concurrent.CancellationException
 import java.util.concurrent.CountDownLatch
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Future
+import java.util.concurrent.RejectedExecutionException
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.AtomicReference
 
 class OutboxSchedulerTest : FunSpec({
 
@@ -131,8 +142,549 @@ class OutboxSchedulerTest : FunSpec({
 
         txInvoked.get() shouldBe true
     }
+
+    test("concurrency > 1 fans out to that many workers per tick") {
+        val callCount = AtomicInteger(0)
+        val threadNames = java.util.Collections.synchronizedSet(mutableSetOf<String>())
+        val latch = CountDownLatch(4)
+        val processor = stubProcessor { _ ->
+            callCount.incrementAndGet()
+            threadNames += Thread.currentThread().name
+            latch.countDown()
+        }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(interval = ofMinutes(1), concurrency = 4),
+        )
+
+        scheduler.start()
+        latch.await(2, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+
+        callCount.get() shouldBe 4
+        threadNames shouldBe setOf("outbox-worker-1", "outbox-worker-2", "outbox-worker-3", "outbox-worker-4")
+    }
+
+    test("concurrency = 1 never invokes workerExecutorFactory (zero-overhead default path)") {
+        val factoryCalled = AtomicBoolean(false)
+        val latch = CountDownLatch(1)
+        val processor = stubProcessor { _ -> latch.countDown() }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(
+                interval = ofMillis(50),
+                concurrency = 1,
+                workerExecutorFactory = { n ->
+                    factoryCalled.set(true)
+                    OutboxSchedulerConfig.defaultPlatformPool(n)
+                },
+            ),
+        )
+
+        scheduler.start()
+        latch.await(2, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+
+        factoryCalled.get() shouldBe false
+    }
+
+    test("one worker's exception does not prevent the other workers from completing") {
+        val callCount = AtomicInteger(0)
+        val latch = CountDownLatch(4)
+        val processor = stubProcessor { _ ->
+            val count = callCount.incrementAndGet()
+            latch.countDown()
+            if (count == 1) throw RuntimeException("worker failure")
+        }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(interval = ofMinutes(1), concurrency = 4),
+        )
+
+        scheduler.start()
+        latch.await(2, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+
+        callCount.get() shouldBe 4
+    }
+
+    test("a spurious interrupt while running does not let ticks overlap (awaitWorker keeps waiting)") {
+        val workerStarted = CountDownLatch(2)
+        val releaseWorkers = CountDownLatch(1)
+        val processor = stubProcessor { _ ->
+            workerStarted.countDown()
+            releaseWorkers.await()
+        }
+        // Counts pool.submit() calls themselves, not worker executions: with a fixed 2-thread pool
+        // already saturated by the blocked first-round workers, an erroneous second tick's
+        // submissions would just sit queued behind them, never actually running -- so counting
+        // executions would miss the overlap entirely. Counting submissions catches it regardless.
+        val submitCount = AtomicInteger(0)
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(
+                interval = ofMillis(20),
+                concurrency = 2,
+                workerExecutorFactory = { n -> SubmitCountingExecutor(OutboxSchedulerConfig.defaultPlatformPool(n), submitCount) },
+            ),
+        )
+
+        try {
+            scheduler.start()
+            workerStarted.await(2, TimeUnit.SECONDS) shouldBe true
+            submitCount.get() shouldBe 2
+
+            // Interrupt the scheduler's own internal thread directly -- NOT via stop() -- to
+            // simulate a spurious interrupt while the scheduler is still meant to be running
+            // normally (running is still true). Without the fix, awaitWorker() would abandon the
+            // wait, tick() would return, and scheduleWithFixedDelay would start a new tick while
+            // these two workers are still blocked -- exactly the overlap the class forbids.
+            val schedulerThread = Thread.getAllStackTraces().keys.first { it.name == "outbox-processor" }
+            schedulerThread.interrupt()
+
+            // Several intervals' worth of margin for a buggy early-return to manifest as extra
+            // submissions before the original two workers are ever released.
+            Thread.sleep(200)
+            submitCount.get() shouldBe 2
+
+            releaseWorkers.countDown()
+        } finally {
+            releaseWorkers.countDown()
+            scheduler.stop()
+        }
+    }
+
+    test("RejectedExecutionException submitting a worker does not kill the scheduler") {
+        val concurrency = 2
+        // workerExecutorFactory only runs on the scheduler's background thread on the first
+        // tick(), so a plain lateinit var raced with this test thread's polling -- an
+        // AtomicReference makes the handoff safe.
+        val rejectingExecutorRef = AtomicReference<AlwaysRejectingExecutor?>(null)
+        val processor = stubProcessor { _ -> }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(
+                interval = ofMillis(20),
+                concurrency = concurrency,
+                workerExecutorFactory = { n ->
+                    AlwaysRejectingExecutor(OutboxSchedulerConfig.defaultPlatformPool(n)).also { rejectingExecutorRef.set(it) }
+                },
+            ),
+        )
+
+        scheduler.start()
+        // Without the fix, the first rejection escapes tick() and scheduleWithFixedDelay
+        // suppresses all further ticks, so submitAttempts would get stuck at `concurrency` (one
+        // tick's worth). Observing it climb past that proves later ticks still ran.
+        awaitCondition { (rejectingExecutorRef.get()?.submitAttempts?.get() ?: 0) > concurrency }
+        scheduler.stop()
+
+        rejectingExecutorRef.get()!!.submitAttempts.get() shouldBeGreaterThanOrEqualTo concurrency
+    }
+
+    test("CancellationException awaiting a worker does not kill the scheduler") {
+        val concurrency = 2
+        // workerExecutorFactory only runs on the scheduler's background thread on the first
+        // tick(), so a plain lateinit var raced with this test thread's polling -- an
+        // AtomicReference makes the handoff safe.
+        val cancellingExecutorRef = AtomicReference<AlwaysCancellingExecutor?>(null)
+        val processor = stubProcessor { _ -> }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(
+                interval = ofMillis(20),
+                concurrency = concurrency,
+                workerExecutorFactory = { n ->
+                    AlwaysCancellingExecutor(OutboxSchedulerConfig.defaultPlatformPool(n)).also { cancellingExecutorRef.set(it) }
+                },
+            ),
+        )
+
+        scheduler.start()
+        // Without the fix, the first CancellationException escapes tick() via awaitWorker()'s
+        // future.get() and scheduleWithFixedDelay suppresses all further ticks, so
+        // cancelledSubmissions would get stuck at `concurrency` (one tick's worth). Observing it
+        // climb past that proves later ticks still ran.
+        awaitCondition { (cancellingExecutorRef.get()?.cancelledSubmissions?.get() ?: 0) > concurrency }
+        scheduler.stop()
+
+        cancellingExecutorRef.get()!!.cancelledSubmissions.get() shouldBeGreaterThanOrEqualTo concurrency
+    }
+
+    test("workerExecutorFactory failure during lazy init does not kill the scheduler, and retries next tick") {
+        val factoryCallCount = AtomicInteger(0)
+        val latch = CountDownLatch(1)
+        val processor = stubProcessor { _ -> latch.countDown() }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(
+                interval = ofMillis(20),
+                concurrency = 2,
+                workerExecutorFactory = { n ->
+                    // Fails on the very first attempt (simulating e.g. resource exhaustion), then
+                    // succeeds -- Lazy retries its initializer after a failed attempt, so this
+                    // proves both that tick() survives the failure and that the next tick retries.
+                    if (factoryCallCount.incrementAndGet() == 1) {
+                        throw RuntimeException("simulated worker pool creation failure")
+                    }
+                    OutboxSchedulerConfig.defaultPlatformPool(n)
+                },
+            ),
+        )
+
+        scheduler.start()
+        // Without the fix, the first factory failure escapes tick() and scheduleWithFixedDelay
+        // suppresses all further ticks, so this would never reach processBatch and time out.
+        latch.await(2, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+
+        factoryCallCount.get() shouldBeGreaterThanOrEqualTo 2
+    }
+
+    test("interrupting the caller during stop() restores the flag and still shuts down without throwing") {
+        val processingStarted = CountDownLatch(1)
+        val releaseProcessing = CountDownLatch(1)
+        val processor = stubProcessor { _ ->
+            processingStarted.countDown()
+            releaseProcessing.await()
+        }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(interval = ofMillis(50)),
+        )
+
+        scheduler.start()
+        processingStarted.await(2, TimeUnit.SECONDS) shouldBe true
+
+        var stopThrew = false
+        var interruptedAfterReturn = false
+        val stopper = Thread {
+            try {
+                scheduler.stop()
+            } catch (_: Throwable) {
+                stopThrew = true
+            }
+            interruptedAfterReturn = Thread.currentThread().isInterrupted
+        }
+        stopper.start()
+        awaitBlocked(stopper)
+        stopper.interrupt()
+        awaitTermination(stopper)
+
+        stopThrew shouldBe false
+        interruptedAfterReturn shouldBe true
+    }
+
+    test("stop() waits again after shutdownNow() forces a slow-to-terminate worker pool to finish") {
+        val processor = stubProcessor { _ -> }
+        val executorRef = AtomicReference<FakeSlowShutdownExecutor?>(null)
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(
+                interval = ofMinutes(1),
+                concurrency = 2,
+                workerExecutorFactory = { n ->
+                    FakeSlowShutdownExecutor(OutboxSchedulerConfig.defaultPlatformPool(n), terminatesAfterShutdownNow = true)
+                        .also { executorRef.set(it) }
+                },
+            ),
+        )
+
+        scheduler.start()
+        awaitCondition { executorRef.get() != null }
+        scheduler.stop()
+
+        // Without the fix, stop() returns as soon as the first awaitTermination() reports false,
+        // never calling shutdownNow() -- awaitTerminationCallCount stuck at 1 -- and never
+        // confirming the forced shutdown actually finished. Observing a second call proves the
+        // follow-up wait happened.
+        executorRef.get()!!.awaitTerminationCallCount.get() shouldBe 2
+    }
+
+    test("stop() logs a warning but still returns if a worker pool never terminates even after shutdownNow()") {
+        val processor = stubProcessor { _ -> }
+        val executorRef = AtomicReference<FakeSlowShutdownExecutor?>(null)
+
+        val schedulerLogger = LoggerFactory.getLogger(OutboxScheduler::class.java) as Logger
+        val warnLogged = CountDownLatch(1)
+        val appender = object : ListAppender<ILoggingEvent>() {
+            override fun append(event: ILoggingEvent) {
+                super.append(event)
+                if (event.level == Level.WARN) warnLogged.countDown()
+            }
+        }.apply { start() }
+        schedulerLogger.addAppender(appender)
+
+        try {
+            val scheduler = OutboxScheduler(
+                outboxProcessor = processor,
+                transactionRunner = noOpTransactionRunner(),
+                config = OutboxSchedulerConfig(
+                    interval = ofMinutes(1),
+                    concurrency = 2,
+                    workerExecutorFactory = { n ->
+                        FakeSlowShutdownExecutor(OutboxSchedulerConfig.defaultPlatformPool(n), terminatesAfterShutdownNow = false)
+                            .also { executorRef.set(it) }
+                    },
+                ),
+            )
+
+            scheduler.start()
+            awaitCondition { executorRef.get() != null }
+            scheduler.stop()
+
+            executorRef.get()!!.awaitTerminationCallCount.get() shouldBe 2
+            warnLogged.await(2, TimeUnit.SECONDS) shouldBe true
+        } finally {
+            schedulerLogger.detachAppender(appender)
+        }
+    }
+
+    test("stop() before start() is a no-op") {
+        val processor = stubProcessor { _ -> }
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(interval = ofMinutes(1)),
+        )
+
+        scheduler.stop() // must return, not throw, despite never having started
+
+        scheduler.isRunning() shouldBe false
+    }
+
+    test("calling stop() a second time is a no-op") {
+        val latch = CountDownLatch(1)
+        val processor = stubProcessor { _ -> latch.countDown() }
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(interval = ofMillis(50)),
+        )
+
+        scheduler.start()
+        latch.await(2, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+        scheduler.stop() // running is already false -- must short-circuit, not re-run shutdown logic
+
+        scheduler.isRunning() shouldBe false
+    }
+
+    test(
+        "interrupting the caller during stop() with an active worker pool logs a warning from " +
+            "awaitWorker and force-shuts-down the worker pool too, all without throwing",
+    ) {
+        // Both workers block (simulating a stuck delivery), so the scheduler's own thread is stuck
+        // inside awaitWorker()'s future.get() when stop() is called. A single interrupt on the
+        // stop()-caller cascades through both executors: it forces scheduler.shutdownNow(), which
+        // interrupts the scheduler thread -- awaitWorker() must catch that itself (not hang, not
+        // crash) and log a warning. That restores the caller's own interrupt flag, so when stop()
+        // moves on to the workers pool's shutdownAndAwait, its awaitTermination sees the flag still
+        // set and immediately treats it as interrupted too, force-shutting-down the worker pool
+        // without a second explicit interrupt.
+        //
+        // The stub retries its own latch wait on interrupt rather than propagating it: if it let
+        // the worker pool's shutdownNow() interrupt complete its Future normally, that would race
+        // against the scheduler thread's own interrupt (from scheduler.shutdownNow()) -- and on a
+        // contended machine, the worker's Future can win, so awaitWorker()'s future.get() returns
+        // normally instead of throwing and no warning is logged (flaky failure at `warnLogged.await`
+        // below). Making the stub interrupt-proof forces the Future to stay incomplete until
+        // releaseWorkers.countDown() below, so the scheduler thread's own interrupt is the only way
+        // awaitWorker() can ever unblock.
+        val workerStarted = CountDownLatch(2)
+        val releaseWorkers = CountDownLatch(1)
+        val processor = stubProcessor { _ ->
+            workerStarted.countDown()
+            var interrupted = false
+            while (true) {
+                try {
+                    releaseWorkers.await()
+                    break
+                } catch (_: InterruptedException) {
+                    interrupted = true
+                }
+            }
+            if (interrupted) Thread.currentThread().interrupt()
+        }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = noOpTransactionRunner(),
+            config = OutboxSchedulerConfig(interval = ofMinutes(1), concurrency = 2),
+        )
+
+        val schedulerLogger = LoggerFactory.getLogger(OutboxScheduler::class.java) as Logger
+        val warnLogged = CountDownLatch(1)
+        val appender = object : ListAppender<ILoggingEvent>() {
+            override fun append(event: ILoggingEvent) {
+                super.append(event)
+                if (event.level == Level.WARN) warnLogged.countDown()
+            }
+        }.apply { start() }
+        schedulerLogger.addAppender(appender)
+
+        try {
+            scheduler.start()
+            workerStarted.await(2, TimeUnit.SECONDS) shouldBe true
+
+            var stopThrew = false
+            val stopper = Thread {
+                try {
+                    scheduler.stop()
+                } catch (_: Throwable) {
+                    stopThrew = true
+                }
+            }
+            stopper.start()
+
+            awaitBlocked(stopper)
+            stopper.interrupt()
+            awaitTermination(stopper)
+
+            warnLogged.await(2, TimeUnit.SECONDS) shouldBe true
+            stopThrew shouldBe false
+        } finally {
+            releaseWorkers.countDown() // let the still-blocked, interrupt-proof workers exit
+            schedulerLogger.detachAppender(appender)
+        }
+    }
 })
 
+/**
+ * Always rejects submissions, tracking how many attempts were made -- used to prove
+ * [OutboxScheduler] doesn't die when [OutboxSchedulerConfig.workerExecutorFactory] returns an
+ * executor that rejects.
+ */
+private class AlwaysRejectingExecutor(private val delegate: ExecutorService) : ExecutorService by delegate {
+    val submitAttempts = AtomicInteger(0)
+
+    override fun submit(task: Runnable): Future<*> {
+        submitAttempts.incrementAndGet()
+        throw RejectedExecutionException("worker pool full (test double)")
+    }
+
+    override fun close() {
+        delegate.close()
+    }
+}
+
+/**
+ * Cancels every submitted task's [Future] immediately, tracking how many were cancelled -- used to
+ * prove [OutboxScheduler] doesn't die when a worker's future surfaces a [CancellationException]
+ * (e.g. a custom [OutboxSchedulerConfig.workerExecutorFactory] cancelling stale tasks as a
+ * backpressure strategy).
+ */
+private class AlwaysCancellingExecutor(private val delegate: ExecutorService) : ExecutorService by delegate {
+    val cancelledSubmissions = AtomicInteger(0)
+
+    override fun submit(task: Runnable): Future<*> {
+        val future = delegate.submit(task)
+        future.cancel(true)
+        cancelledSubmissions.incrementAndGet()
+        return future
+    }
+
+    override fun close() {
+        delegate.close()
+    }
+}
+
+/**
+ * Fakes a slow-to-terminate executor: [awaitTermination] reports "didn't terminate" on its first
+ * call regardless of the real delegate's state (keeping the test fast, not tied to any real
+ * timeout), then [terminatesAfterShutdownNow] on subsequent calls -- simulating an executor whose
+ * tasks only actually stop once `shutdownNow()` forcibly interrupts them. Used to prove
+ * [OutboxScheduler.stop] waits again for termination after forcing `shutdownNow()`, instead of
+ * returning as soon as the first wait times out.
+ */
+private class FakeSlowShutdownExecutor(
+    private val delegate: ExecutorService,
+    private val terminatesAfterShutdownNow: Boolean,
+) : ExecutorService by delegate {
+    val awaitTerminationCallCount = AtomicInteger(0)
+    private val shutdownNowCalled = AtomicBoolean(false)
+
+    override fun awaitTermination(timeout: Long, unit: TimeUnit): Boolean {
+        if (awaitTerminationCallCount.incrementAndGet() == 1) return false
+        return shutdownNowCalled.get() && terminatesAfterShutdownNow
+    }
+
+    override fun shutdownNow(): MutableList<Runnable> {
+        shutdownNowCalled.set(true)
+        return delegate.shutdownNow()
+    }
+
+    override fun close() {
+        delegate.close()
+    }
+}
+
+/** Counts [submit] calls (queued or not) without altering delegate behavior otherwise. */
+private class SubmitCountingExecutor(
+    private val delegate: ExecutorService,
+    private val submitCount: AtomicInteger,
+) : ExecutorService by delegate {
+    override fun submit(task: Runnable): Future<*> {
+        submitCount.incrementAndGet()
+        return delegate.submit(task)
+    }
+
+    override fun close() {
+        delegate.close()
+    }
+}
+
+private fun awaitCondition(timeoutMs: Long = 2_000, condition: () -> Boolean) {
+    val deadline = System.currentTimeMillis() + timeoutMs
+    while (!condition()) {
+        check(System.currentTimeMillis() < deadline) { "Condition not met within ${timeoutMs}ms" }
+        Thread.sleep(5)
+    }
+}
+
+/**
+ * Polls (rather than sleeping a fixed duration) until [thread] is parked waiting, so the test
+ * isn't sensitive to CI scheduling jitter -- interrupting before the thread actually blocks would
+ * make it miss the wait entirely.
+ */
+private fun awaitBlocked(thread: Thread, timeoutMs: Long = 5_000) {
+    val deadline = System.currentTimeMillis() + timeoutMs
+    while (thread.state != Thread.State.WAITING && thread.state != Thread.State.TIMED_WAITING) {
+        check(System.currentTimeMillis() < deadline) { "Thread never blocked (state=${thread.state})" }
+        Thread.sleep(5)
+    }
+}
+
+/**
+ * A timed [Thread.join] only establishes a happens-before edge for the joined thread's writes if
+ * it returns because the thread actually terminated, not because the timeout elapsed. Confirming
+ * termination and then joining again unconditionally (returns immediately on an already-dead
+ * thread) closes that gap.
+ */
+private fun awaitTermination(thread: Thread, timeoutMs: Long = 5_000) {
+    thread.join(timeoutMs)
+    check(!thread.isAlive) { "Thread did not terminate within ${timeoutMs}ms" }
+    thread.join()
+}
+
 private fun stubProcessor(onProcessNext: (Int) -> Unit): OutboxProcessor {
     val store = object : OutboxStore {
         override fun persist(entry: OutboxEntry) = entry
diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt
index 55f1fd6..73142d2 100644
--- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt
+++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt
@@ -6,9 +6,12 @@ import com.softwaremill.okapi.core.OutboxEntryProcessor
 import com.softwaremill.okapi.core.OutboxId
 import com.softwaremill.okapi.core.OutboxMessage
 import com.softwaremill.okapi.core.OutboxProcessor
+import com.softwaremill.okapi.core.OutboxScheduler
+import com.softwaremill.okapi.core.OutboxSchedulerConfig
 import com.softwaremill.okapi.core.OutboxStatus
 import com.softwaremill.okapi.core.OutboxStore
 import com.softwaremill.okapi.core.RetryPolicy
+import com.softwaremill.okapi.core.TransactionRunner
 import com.softwaremill.okapi.test.support.JdbcConnectionProvider
 import com.softwaremill.okapi.test.support.RecordingMessageDeliverer
 import io.kotest.assertions.withClue
@@ -18,6 +21,7 @@ import io.kotest.matchers.maps.shouldContain
 import io.kotest.matchers.shouldBe
 import java.sql.Connection
 import java.time.Clock
+import java.time.Duration
 import java.time.Instant
 import java.time.ZoneOffset
 import java.util.concurrent.CompletableFuture
@@ -165,4 +169,49 @@ fun FunSpec.concurrentClaimTests(
             counts shouldContain (OutboxStatus.PENDING to 0L)
         }
     }
+
+    test("[$dbName] OutboxScheduler with concurrency=4 workers processes disjoint batches without amplification") {
+        val fixedClock = Clock.fixed(Instant.parse("2024-01-01T00:00:00Z"), ZoneOffset.UTC)
+
+        // 80 entries, batchSize=10, concurrency=4 -> 40 claimed per tick, so it takes 2 ticks
+        // to drain everything. This exercises both the multi-worker fan-out within a single tick
+        // AND repeated ticks, unlike the single-shot manual-thread test above.
+        val entryCount = 80
+        jdbc.withTransaction {
+            (0 until entryCount).forEach { i -> store.persist(createTestEntry(i)) }
+        }
+
+        val recorder = RecordingMessageDeliverer()
+        val entryProcessor = OutboxEntryProcessor(recorder, RetryPolicy(maxRetries = 0), fixedClock)
+        val processor = OutboxProcessor(store, entryProcessor)
+
+        val transactionRunner = object : TransactionRunner {
+            override fun <T> runInTransaction(block: () -> T): T =
+                jdbc.withTransaction(transactionIsolation = Connection.TRANSACTION_READ_COMMITTED) { block() }
+        }
+
+        val scheduler = OutboxScheduler(
+            outboxProcessor = processor,
+            transactionRunner = transactionRunner,
+            config = OutboxSchedulerConfig(interval = Duration.ofMillis(50), batchSize = 10, concurrency = 4),
+        )
+
+        scheduler.start()
+        val deadline = System.currentTimeMillis() + 30_000
+        while (recorder.deliveryCount() < entryCount && System.currentTimeMillis() < deadline) {
+            Thread.sleep(50)
+        }
+        scheduler.stop()
+
+        recorder.assertNoAmplification()
+        withClue("Expected exactly $entryCount unique deliveries from a concurrency=4 scheduler, got ${recorder.deliveryCount()}") {
+            recorder.deliveryCount() shouldBe entryCount
+        }
+
+        val counts = jdbc.withTransaction { store.countByStatuses() }
+        withClue("DB state after scheduler run: $counts") {
+            counts shouldContain (OutboxStatus.DELIVERED to entryCount.toLong())
+            counts shouldContain (OutboxStatus.PENDING to 0L)
+        }
+    }
 }
diff --git a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt
index e575aa6..9d9dfc9 100644
--- a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt
+++ b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt
@@ -205,6 +205,7 @@ class OutboxAutoConfiguration(
             config = OutboxSchedulerConfig(
                 interval = props.interval,
                 batchSize = props.batchSize,
+                concurrency = props.concurrency,
             ),
         )
     }
diff --git a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt
index 5c1f077..f7a8fb3 100644
--- a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt
+++ b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt
@@ -8,10 +8,16 @@ data class OutboxProcessorProperties(
     val interval: Duration = Duration.ofSeconds(1),
     val batchSize: Int = 10,
     val maxRetries: Int = 5,
+    /**
+     * Number of parallel workers fanned out per scheduler tick, each claiming its own batch
+     * via `FOR UPDATE SKIP LOCKED`. See [com.softwaremill.okapi.core.OutboxSchedulerConfig].
+     */
+    val concurrency: Int = 1,
 ) {
     init {
         require(!interval.isNegative && interval.toMillis() > 0) { "interval must be at least 1ms" }
         require(batchSize > 0) { "batchSize must be positive" }
         require(maxRetries >= 0) { "maxRetries must be >= 0" }
+        require(concurrency in 1..256) { "concurrency must be between 1 and 256" }
     }
 }
diff --git a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt
index 56c7f86..fcd5239 100644
--- a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt
+++ b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt
@@ -50,12 +50,14 @@ class OutboxProcessorAutoConfigurationTest : FunSpec({
                 "okapi.processor.interval=500ms",
                 "okapi.processor.batch-size=20",
                 "okapi.processor.max-retries=3",
+                "okapi.processor.concurrency=4",
             )
             .run { ctx ->
                 val props = ctx.getBean(OutboxProcessorProperties::class.java)
                 props.interval shouldBe ofMillis(500)
                 props.batchSize shouldBe 20
                 props.maxRetries shouldBe 3
+                props.concurrency shouldBe 4
             }
     }
 
@@ -65,6 +67,7 @@ class OutboxProcessorAutoConfigurationTest : FunSpec({
             props.interval shouldBe ofSeconds(1)
             props.batchSize shouldBe 10
             props.maxRetries shouldBe 5
+            props.concurrency shouldBe 1
         }
     }
 
@@ -92,6 +95,14 @@ class OutboxProcessorAutoConfigurationTest : FunSpec({
             }
     }
 
+    test("invalid concurrency triggers startup failure") {
+        contextRunner
+            .withPropertyValues("okapi.processor.concurrency=0")
+            .run { ctx ->
+                ctx.startupFailure.shouldNotBeNull()
+            }
+    }
+
     test("stop(callback) invokes callback AND actually halts the scheduler") {
         contextRunner.run { ctx ->
             val scheduler = ctx.getBean(OutboxProcessorScheduler::class.java)