diff --git a/.gitignore b/.gitignore
index 93dfca86..fcf2acb6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,4 +15,11 @@ run/
 minecraft/
 
 # Mkdocs
-site/
\ No newline at end of file
+site/
+
+# Never commit credentials. The publish script reads a token from $GITHUB_TOKEN or
+# from a .secrets/gh_token file; this keeps either out of the repository if one is
+# accidentally placed inside it.
+.secrets/
+gh_token
+*_token
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 00000000..a7798fbc
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,158 @@
+# Changelog
+
+## 1.3.25-reopt.1 (2026-10-05)
+
+Reoptimization pass over upstream Ledger 1.3.25 for MC 26.3 Fabric.
+This entry describes the current version. Changes introduced since reopt.1 are
+listed first, followed by a cumulative summary of everything this build carries
+relative to upstream.
+
+### Changed since reopt.2
+
+- **Real JDBC batch insert (ingest).** Exposed's `batchInsert` was issuing one
+  statement per row - a 13,500-action burst produced 13,500 individual
+  `INSERT INTO actions` statements. On the same schema and rows a genuine JDBC
+  batch costs 23 us/row and a single statement 44 us/row, against 310-500 us/row
+  on the live path, so statement machinery dominated the insert. Inserts now go
+  through a prepared statement on the transaction's own connection with
+  addBatch/executeBatch. Values still come from Exposed column types via
+  `valueToDB`, so the stored representation is unchanged - verified against an
+  archive written by the ORM path, 500/500 samples byte-identical.
+- **Burst-aware queue scheduling.** The writer waited the full `batchDelay`
+  (default 500 ms) on every pass even with work already queued: 6.1 s of an 8.7 s
+  ingest window was spent waiting against 2.6 s of writes. It now skips the wait
+  only while the previous pass filled a complete batch (i.e. while the queue is
+  being flooded), and returns to the configured cadence as soon as a pass comes up
+  short. Trickle traffic still batches exactly as before; an idle server does not
+  become one transaction per action.
+- **Rollback/restore batch size 1,000 -> 5,000 rows.** Fewer read and commit round
+  trips for the same undo: 3.86/4.16 s became 1.99/2.03 s on a 13,500-block
+  rollback (about 2x). Zero "can't keep up" warnings at either setting, so this
+  does not cost responsiveness - the per-tick budget still yields mid-batch.
+
+Combined effect on a 13,500-action workload against reopt.1 (medians of three
+runs, fresh terrain each time):
+
+| | reopt.1 | reopt.3 |
+|---|---|---|
+| drain window | 4.534 s | 3.622 s |
+| drain rate | 2,977 rows/s | 3,727 rows/s |
+| rollback | 4.01 s | 2.01 s |
+
+The drain window contains a fixed 3.0 s settle detector, so the write work behind
+it went from about 1.53 s to 0.62 s.
+
+### Evaluated and rejected (measured, not assumed)
+
+- **SQLite foreign keys off** - a controlled transaction-cost test showed FK
+  checks cost 5.9 -> 5.7 ms per 1,000-row transaction (3%). Not worth weakening
+  referential integrity for.
+- **SQLite `page_size`** - cannot be changed on an existing database; PRAGMA
+  page_size is silently ignored once the file exists (every tested value collapsed
+  back to 4,096). Not a lever this build can pull. The same test confirmed VACUUM
+  alone reclaims about 4.7% of the file, which `/ledger compact` already exposes.
+- **Dropping the legacy `time` TEXT column** (~13% more space) - deliberately not
+  implemented. Exposed cannot map an optional column, so supporting both schemas
+  would mean duplicating the table definition and branching the read path used by
+  every search and rollback; combined with the change being irreversible on user
+  databases, that trade is not worth 13% of disk under a
+  stability-first priority. `time_ms` already makes `time` redundant for querying.
+
+### Changed since reopt.1
+
+- **Integer time index (main storage win).** `actions.time` stores TEXT
+  (`'2026-10-05 13:22:21.702'`, ~23 bytes). An index over that key cost ~35 bytes
+  per row - measured at 19.6% of the entire database, the largest non-table object.
+  A new `time_ms` column holds the same instant as epoch milliseconds and carries
+  the index instead:
+  - time index: **35.5 -> 16.4 B/row (-54%)**
+  - whole database, like-for-like (both fresh, neither vacuumed):
+    **181.44 -> 167.2 B/row (-7.9%)**
+  - an independent 500,000-row synthetic measurement reproduces -53% index size and
+    -11.5% file size
+  Note on the two sizes: running `/ledger compact` after the migration reclaims the
+  pages the dropped index leaves on the freelist, which takes an existing database
+  down by roughly 12%. Part of that comes from VACUUM itself - which any index-drop
+  would also produce - so the honest attribution for this change is the -7.9%
+  like-for-like figure above.
+  The TEXT column is kept and written alongside, so display strings and any external
+  tooling reading `time` are unaffected.
+- **Automatic, resumable migration.** On startup the column is added, indexed and
+  backfilled in id-ordered batches; once no sentinel values remain the superseded
+  TEXT index is dropped. Every value is verified to be an exact millisecond match for
+  its TEXT source, distinct counts are preserved, and no rows are lost
+  (`bench/test_time_migration.py`, 8/8 checks). If a backfill cannot finish, queries
+  transparently fall back to the TEXT column, so a partially migrated database still
+  returns correct results. With `updateSchema = false` the migration does not run at
+  all and the plugin stays on the original schema.
+- **SQLite connection tuning.** `temp_store=MEMORY`, 16 MiB page cache, 256 MiB
+  `mmap_size`, 64 MiB `journal_size_limit`, applied through `SQLiteConfig` so they
+  reach every pooled connection. Verified applied via JDBC, not assumed.
+- **Adaptive rollback/restore tick budget.** The fixed 25 ms budget is replaced by one
+  derived from the server's smoothed tick time (35 ms idle, 5 ms when ticks are already
+  at capacity). Scope note: on an idle benchmark server this changes nothing
+  measurable - a rollback was measured completing with zero yields, so the budget was
+  never its bottleneck. The benefit is on a loaded server, which a benchmark cannot
+  show. A previously recorded claim that a large share of rollback wall time went into
+  `delay(1)` was wrong and has been removed from the source comments.
+
+### Evaluated and rejected
+
+- **Composite `(time_ms, id)` and `(time_ms, rolled_back, id)` indexes.** Once the time
+  key became an 8-byte integer these became affordable in principle, so they were
+  measured at 500,000 rows. Both still trigger `USE TEMP B-TREE FOR ORDER BY` for
+  Ledger's `ORDER BY id DESC LIMIT n` queries, giving no query improvement (the
+  differences seen were within noise) while adding 3.0% / 3.7% to the file. Not
+  adopted.
+
+### Cumulative summary vs upstream 1.3.25
+
+- **MC 26.3 block-state NBT key rename - handled upstream, not here.** An earlier
+  revision of this branch carried its own fix for this; upstream fixed it
+  independently in #402 (merged as 22705ac), so that commit has been dropped when
+  this branch was rebased onto 1.3.25 and the code is upstream's. Credit for the
+  fix belongs there. It is listed only to explain why this branch no longer
+  touches `NbtUtils.kt`.
+- **Fixed - rollback keyset off-by-one** that could exclude the newest action row.
+- **Block-state dictionary encoding** - identical state strings are stored once in
+  `block_states` and referenced by integer id. Measured on a stairs workload, it saves
+  31.3% (55.8 bytes per row) versus storing the literal state text.
+- **`/ledger compact`** - batched migration of legacy text states into the dictionary
+  plus `VACUUM`. Also the way to reclaim the space freed by the index change above.
+- **Streaming rollback/restore** - keyset pagination with per-batch progress commits.
+- **COUNT(*) result cache** (30 s TTL).
+- **SQLite tuning** - WAL journal mode, `synchronous=NORMAL`, 10 s busy timeout.
+- **Faster shutdown drain** with `NonCancellable` batch writes.
+- **Reverted composite indexes** that an intermediate revision of this branch added;
+  they inflated the index footprint by 127% for no query benefit, and a startup
+  migration now repairs databases that carry them.
+
+### Verification (live MC 26.3 Fabric server)
+
+- Boot, setblock/search/rollback/restore/status/compact all run with zero exceptions.
+- Rollback -> restore round-trip preserves exact block state
+  (`execute if block ... oak_stairs[facing=north,half=top]` passes after restore).
+- Dictionary round-trip and `compact` behaviour verified against a live database.
+- Time migration: 8/8 checks on a real server boot.
+- Index-repair migration: 4/4 checks on a database carrying the intermediate layout.
+- Benchmarks A (13,500 uniform placements) and B (stateful stairs) both report **zero
+  errors and exactly 13,500 rows** for every run.
+
+### Benchmarked against CoreProtect 24.1 (MC 26.3)
+
+See `ledger-vs-coreprotect-性能实测报告.html` at the project root. Medians of three
+runs, 13,500 block placements:
+
+| | Ledger reopt.2 | Ledger reopt.1 | CoreProtect (DuckDB) |
+|---|---|---|---|
+| bytes / row | 167.2 | 181.4 | 107.7 |
+| ingest (rows/s) | 1,917 | 1,957 | 2,156 |
+| lookup | 185 ms | 159 ms | 153 ms |
+| rollback | 3.47 s | 3.95 s | 0.61 s |
+
+**Only the bytes-per-row row is a real, reproducible result** - it comes from the
+schema itself and is independent of machine load. Every other row moved by more
+between campaigns than between the two builds: in an earlier campaign reopt.1
+ingested at 1,959 rows/s and rolled back in 2.63-2.69 s, while in the final one it
+did 1,957 rows/s and 3.95 s. Ingest, lookup and rollback are therefore reported as
+*no measurable change*, not as improvements.
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/commands/LedgerCommand.kt b/src/main/kotlin/com/github/quiltservertools/ledger/commands/LedgerCommand.kt
index 4abc30c9..88610d1d 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/commands/LedgerCommand.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/commands/LedgerCommand.kt
@@ -1,6 +1,7 @@
 package com.github.quiltservertools.ledger.commands
 
 import com.github.quiltservertools.ledger.api.ExtensionManager
+import com.github.quiltservertools.ledger.commands.subcommands.CompactCommand
 import com.github.quiltservertools.ledger.commands.subcommands.InspectCommand
 import com.github.quiltservertools.ledger.commands.subcommands.PageCommand
 import com.github.quiltservertools.ledger.commands.subcommands.PlayerCommand
@@ -51,6 +52,9 @@ fun registerCommands(dispatcher: Dispatcher) {
 
     rootNode.addChild(PurgeCommand.build())
 
+    // Reoptimization: migrate legacy text block states to dictionary encoding + vacuum
+    rootNode.addChild(CompactCommand.build())
+
     rootNode.addChild(PlayerCommand.build())
 
     ExtensionManager.commands.forEach {
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/CompactCommand.kt b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/CompactCommand.kt
new file mode 100644
index 00000000..19b72a6a
--- /dev/null
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/CompactCommand.kt
@@ -0,0 +1,80 @@
+package com.github.quiltservertools.ledger.commands.subcommands
+
+import com.github.quiltservertools.ledger.Ledger
+import com.github.quiltservertools.ledger.commands.BuildableCommand
+import com.github.quiltservertools.ledger.config.SearchSpec
+import com.github.quiltservertools.ledger.config.config
+import com.github.quiltservertools.ledger.database.DatabaseManager
+import com.github.quiltservertools.ledger.utility.Context
+import com.github.quiltservertools.ledger.utility.LiteralNode
+import com.github.quiltservertools.ledger.utility.TextColorPallet
+import com.github.quiltservertools.ledger.utility.literal
+import kotlinx.coroutines.launch
+import me.lucko.fabric.api.permissions.v0.Permissions
+import net.minecraft.commands.Commands.literal
+import net.minecraft.network.chat.Component
+
+/**
+ * Reoptimization: /ledger compact
+ *
+ * Migrates legacy text block_state / old_block_state columns into the
+ * block_states dictionary (int references) and then reclaims the freed disk
+ * space with VACUUM (SQLite) / OPTIMIZE TABLE (MySQL).
+ *
+ * Safe to run on a live server:
+ *  - reads and writes go through the single database thread
+ *  - each row's text -> ref swap happens atomically in one UPDATE, so
+ *    concurrent searches always see a valid state (either the text column or
+ *    the ref column)
+ *  - new writes already use the dictionary and are skipped automatically
+ */
+object CompactCommand : BuildableCommand {
+    private const val PROGRESS_INTERVAL_MS = 5_000L
+
+    override fun build(): LiteralNode = literal("compact")
+        .requires(Permissions.require("ledger.commands.purge", config[SearchSpec.purgePermissionLevel]))
+        .executes { runCompact(it) }
+        .build()
+
+    private fun runCompact(ctx: Context): Int {
+        val source = ctx.source
+        source.sendSuccess(
+            { Component.translatable("text.ledger.compact.starting").setStyle(TextColorPallet.secondary) },
+            true,
+        )
+        Ledger.launch {
+            var lastReport = System.currentTimeMillis()
+            val migrated = DatabaseManager.compactBlockStates(
+                batchSize = 5000,
+                onProgress = { done, total ->
+                    val now = System.currentTimeMillis()
+                    if (now - lastReport >= PROGRESS_INTERVAL_MS) {
+                        lastReport = now
+                        source.sendSuccess(
+                            {
+                                Component.translatable(
+                                    "text.ledger.compact.progress",
+                                    done.toString().literal().setStyle(TextColorPallet.secondary),
+                                    total.toString().literal().setStyle(TextColorPallet.secondary),
+                                ).setStyle(TextColorPallet.primary)
+                            },
+                            false,
+                        )
+                    }
+                },
+            )
+            DatabaseManager.vacuumDatabase()
+
+            source.sendSuccess(
+                {
+                    Component.translatable(
+                        "text.ledger.compact.complete",
+                        migrated.toString().literal().setStyle(TextColorPallet.secondary),
+                    ).setStyle(TextColorPallet.primary)
+                },
+                true,
+            )
+        }
+        return 1
+    }
+}
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/PageCommand.kt b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/PageCommand.kt
index f09b3c81..d6f3f8ef 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/PageCommand.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/PageCommand.kt
@@ -37,8 +37,8 @@ object PageCommand : BuildableCommand {
                     source,
                     results,
                     Component.translatable(
-                        "text.ledger.header.search"
-                    ).setStyle(TextColorPallet.primary)
+                        "text.ledger.header.search",
+                    ).setStyle(TextColorPallet.primary),
                 )
             }
 
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RestoreCommand.kt b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RestoreCommand.kt
index b165dbed..f032bca0 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RestoreCommand.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RestoreCommand.kt
@@ -12,6 +12,7 @@ import com.github.quiltservertools.ledger.utility.MessageUtils
 import com.github.quiltservertools.ledger.utility.TextColorPallet
 import com.github.quiltservertools.ledger.utility.launchMain
 import com.github.quiltservertools.ledger.utility.literal
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import me.lucko.fabric.api.permissions.v0.Permissions
 import net.minecraft.commands.Commands
@@ -31,9 +32,13 @@ object RestoreCommand : BuildableCommand {
         params.ensureSpecific()
         Ledger.launch {
             MessageUtils.warnBusy(source)
-            val actions = DatabaseManager.selectRestore(params)
 
-            if (actions.isEmpty()) {
+            // Reoptimization: restore only targets previously rolled-back actions,
+            // so no upper bound is needed - restoring logs new actions, but those
+            // have rolledBack=false and the batch filter requires rolledBack=true.
+            val total = DatabaseManager.countActionsFor(params, rolledBack = true)
+
+            if (total == 0L) {
                 source.sendFailure(Component.translatable("error.ledger.command.no_results"))
                 return@launch
             }
@@ -42,7 +47,7 @@ object RestoreCommand : BuildableCommand {
                 {
                     Component.translatable(
                         "text.ledger.restore.start",
-                        actions.size.toString().literal().setStyle(TextColorPallet.secondary),
+                        total.toString().literal().setStyle(TextColorPallet.secondary),
                     ).setStyle(TextColorPallet.primary)
                 },
                 true,
@@ -50,16 +55,62 @@ object RestoreCommand : BuildableCommand {
 
             context.source.level.launchMain {
                 val fails = HashMap<String, Int>()
-                val actionIds = HashSet<Int>()
-                for (action in actions) {
-                    if (!action.restore(context.source.server)) {
-                        fails[action.identifier] = fails.getOrPut(action.identifier) { 0 } + 1
-                    } else {
-                        actionIds.add(action.id)
+                var applied = 0L
+                var batchIndex = 0
+                var cursor = 0
+
+                while (true) {
+                    // Keyset-paginated DB read: at most ROLLBACK_BATCH_SIZE rows in memory.
+                    val actions = DatabaseManager.selectRestoreBatch(params, cursor, ROLLBACK_BATCH_SIZE)
+                    if (actions.isEmpty()) break
+
+                    // Re-checked each batch: the server's tick load can change mid-restore.
+                    val tickBudgetNs = rollbackTickBudgetNs(context.source.server)
+
+                    var batchStart = System.nanoTime()
+                    val actionIds = HashSet<Int>(actions.size)
+
+                    for (action in actions) {
+                        if (!action.restore(context.source.server)) {
+                            fails[action.identifier] = fails.getOrPut(action.identifier) { 0 } + 1
+                        } else {
+                            actionIds.add(action.id)
+                            applied++
+                        }
+
+                        // Adaptive tick budget: commit partial progress and yield the
+                        // rest of the tick once we exceed the time budget.
+                        if (System.nanoTime() - batchStart > tickBudgetNs) {
+                            if (actionIds.isNotEmpty()) {
+                                DatabaseManager.restoreActions(actionIds)
+                                actionIds.clear()
+                            }
+                            delay(1) // one Minecraft tick
+                            batchStart = System.nanoTime()
+                        }
+                    }
+
+                    // Per-batch commit of the restored flags (crash-safe).
+                    if (actionIds.isNotEmpty()) {
+                        DatabaseManager.restoreActions(actionIds)
+                    }
+
+                    // Oldest-first ordering: the last row of the batch is the next cursor.
+                    cursor = actions.last().id
+                    batchIndex++
+
+                    if (batchIndex % ROLLBACK_PROGRESS_INTERVAL == 0) {
+                        source.sendSuccess(
+                            {
+                                Component.translatable(
+                                    "text.ledger.restore.progress",
+                                    applied.toString().literal().setStyle(TextColorPallet.secondary),
+                                    total.toString().literal().setStyle(TextColorPallet.secondary),
+                                ).setStyle(TextColorPallet.primary)
+                            },
+                            false,
+                        )
                     }
-                }
-                Ledger.launch {
-                    DatabaseManager.restoreActions(actionIds)
                 }
 
                 for (entry in fails.entries) {
@@ -77,7 +128,7 @@ object RestoreCommand : BuildableCommand {
                     {
                         Component.translatable(
                             "text.ledger.restore.finish",
-                            actions.size,
+                            applied.toString(),
                         ).setStyle(TextColorPallet.primary)
                     },
                     true,
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RollbackCommand.kt b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RollbackCommand.kt
index cf69d8c2..bf524999 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RollbackCommand.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/commands/subcommands/RollbackCommand.kt
@@ -12,10 +12,78 @@ import com.github.quiltservertools.ledger.utility.MessageUtils
 import com.github.quiltservertools.ledger.utility.TextColorPallet
 import com.github.quiltservertools.ledger.utility.launchMain
 import com.github.quiltservertools.ledger.utility.literal
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import me.lucko.fabric.api.permissions.v0.Permissions
 import net.minecraft.commands.Commands
 import net.minecraft.network.chat.Component
+import net.minecraft.server.MinecraftServer
+
+// Reoptimization: streaming rollback/restore tuning (shared with RestoreCommand).
+// Previously the whole result set was loaded into memory and applied in one
+// monolithic main-thread loop, freezing the server on large rollbacks. Now:
+//  - DB batches are read via keyset pagination (max ROLLBACK_BATCH_SIZE rows in memory)
+//  - the main thread works at most one tick budget before yielding (see below)
+//  - rolled-back flags are committed per batch, so progress survives a crash
+/**
+ * Reoptimization: rows read and committed per rollback/restore step.
+ *
+ * Raised from 1,000 to 5,000 after measuring the trade-off on a 13,500-block
+ * rollback: 1,000 took 3.86/4.16 s and 5,000 took 1.99/2.03 s, i.e. about twice as
+ * fast, because the per-read and per-commit overhead is paid three times instead of
+ * fourteen. Both batch sizes produced zero "can't keep up" warnings, so the larger
+ * batch does not cost server responsiveness - the per-tick work budget below still
+ * splits a batch and yields mid-way.
+ *
+ * Worst-case memory is bounded by about 5,000 materialised actions.
+ */
+internal const val ROLLBACK_BATCH_SIZE = 5000
+internal const val ROLLBACK_PROGRESS_INTERVAL = 5 // batches between progress messages
+
+// Reoptimization: adaptive tick budget thresholds. Smoothed tick durations (milliseconds
+// of a 50 ms tick) that select a per-tick work budget (nanoseconds).
+private const val SATURATED_TICK_MS = 50f
+private const val BUSY_TICK_MS = 35f
+private const val WARM_TICK_MS = 20f
+private const val MIN_TICK_BUDGET_NS = 5_000_000L
+private const val BUSY_TICK_BUDGET_NS = 15_000_000L
+private const val WARM_TICK_BUDGET_NS = 25_000_000L
+private const val IDLE_TICK_BUDGET_NS = 35_000_000L
+
+/**
+ * Reoptimization: the per-tick work budget is sized from the server's current tick
+ * load instead of being a fixed 25 ms.
+ *
+ * Honest scope note: on a *benchmark* server (13,500-block rollback, nothing else
+ * running) this changes almost nothing. Measured with the adaptive budget, a rollback
+ * completes with zero `delay(1)` yields in ~3.1 s of which essentially all of it is
+ * the block writes themselves - the budget was never the bottleneck there. Earlier
+ * profiling notes claiming a large share of the wall time went into `delay(1)` were
+ * wrong and have been removed.
+ *
+ * The point of the change is the *loaded* case, which a benchmark cannot show: a
+ * fixed 25 ms budget behaves identically whether the server is idle or already
+ * struggling at 45 ms ticks, because the budget cannot see the difference. Reading
+ * the smoothed tick time lets a rollback back off automatically when the server is
+ * under load, and use more of an idle tick when it is not. The feedback is
+ * self-limiting: a rollback that stretches ticks shrinks its own budget next tick.
+ *
+ * @return nanoseconds of main-thread work allowed per tick
+ */
+internal fun rollbackTickBudgetNs(server: MinecraftServer): Long {
+    val smoothedTickMs = server.currentSmoothedTickTime
+    return when {
+        // Server already at or past a full tick: barely touch the main thread.
+        smoothedTickMs >= SATURATED_TICK_MS -> MIN_TICK_BUDGET_NS
+
+        smoothedTickMs >= BUSY_TICK_MS -> BUSY_TICK_BUDGET_NS
+
+        smoothedTickMs >= WARM_TICK_MS -> WARM_TICK_BUDGET_NS
+
+        // Idle: use most of the spare time in the tick, so a rollback finishes fast.
+        else -> IDLE_TICK_BUDGET_NS
+    }
+}
 
 object RollbackCommand : BuildableCommand {
     override fun build(): LiteralNode = Commands.literal("rollback")
@@ -31,9 +99,15 @@ object RollbackCommand : BuildableCommand {
         params.ensureSpecific()
         Ledger.launch {
             MessageUtils.warnBusy(source)
-            val actions = DatabaseManager.selectRollback(params)
 
-            if (actions.isEmpty()) {
+            // Reoptimization: exclusive upper bound taken before any changes are applied,
+            // so actions logged by this rollback itself (ids >= bound) can never be
+            // picked up by later batches of the same rollback. +1 because the keyset
+            // predicate is strictly less-than and the newest row must be included.
+            val upperBound = DatabaseManager.currentMaxActionId() + 1
+            val total = DatabaseManager.countActionsFor(params, rolledBack = false)
+
+            if (total == 0L) {
                 source.sendFailure(Component.translatable("error.ledger.command.no_results"))
                 return@launch
             }
@@ -42,7 +116,7 @@ object RollbackCommand : BuildableCommand {
                 {
                     Component.translatable(
                         "text.ledger.rollback.start",
-                        actions.size.toString().literal().setStyle(TextColorPallet.secondary),
+                        total.toString().literal().setStyle(TextColorPallet.secondary),
                     ).setStyle(TextColorPallet.primary)
                 },
                 true,
@@ -50,16 +124,62 @@ object RollbackCommand : BuildableCommand {
 
             context.source.level.launchMain {
                 val fails = HashMap<String, Int>()
-                val actionIds = HashSet<Int>()
-                for (action in actions) {
-                    if (!action.rollback(context.source.server)) {
-                        fails[action.identifier] = fails.getOrPut(action.identifier) { 0 } + 1
-                    } else {
-                        actionIds.add(action.id)
+                var applied = 0L
+                var batchIndex = 0
+                var cursor = upperBound
+
+                while (true) {
+                    // Keyset-paginated DB read: at most ROLLBACK_BATCH_SIZE rows in memory.
+                    val actions = DatabaseManager.selectRollbackBatch(params, cursor, ROLLBACK_BATCH_SIZE)
+                    if (actions.isEmpty()) break
+
+                    // Re-checked each batch: the server's tick load can change mid-rollback.
+                    val tickBudgetNs = rollbackTickBudgetNs(context.source.server)
+
+                    var batchStart = System.nanoTime()
+                    val actionIds = HashSet<Int>(actions.size)
+
+                    for (action in actions) {
+                        if (!action.rollback(context.source.server)) {
+                            fails[action.identifier] = fails.getOrPut(action.identifier) { 0 } + 1
+                        } else {
+                            actionIds.add(action.id)
+                            applied++
+                        }
+
+                        // Adaptive tick budget: commit partial progress and yield the
+                        // rest of the tick once we exceed the time budget.
+                        if (System.nanoTime() - batchStart > tickBudgetNs) {
+                            if (actionIds.isNotEmpty()) {
+                                DatabaseManager.rollbackActions(actionIds)
+                                actionIds.clear()
+                            }
+                            delay(1) // one Minecraft tick
+                            batchStart = System.nanoTime()
+                        }
+                    }
+
+                    // Per-batch commit of the rolled-back flags (crash-safe).
+                    if (actionIds.isNotEmpty()) {
+                        DatabaseManager.rollbackActions(actionIds)
+                    }
+
+                    // Newest-first ordering: the last row of the batch is the next cursor.
+                    cursor = actions.last().id
+                    batchIndex++
+
+                    if (batchIndex % ROLLBACK_PROGRESS_INTERVAL == 0) {
+                        source.sendSuccess(
+                            {
+                                Component.translatable(
+                                    "text.ledger.rollback.progress",
+                                    applied.toString().literal().setStyle(TextColorPallet.secondary),
+                                    total.toString().literal().setStyle(TextColorPallet.secondary),
+                                ).setStyle(TextColorPallet.primary)
+                            },
+                            false,
+                        )
                     }
-                }
-                Ledger.launch {
-                    DatabaseManager.rollbackActions(actionIds)
                 }
 
                 for (entry in fails.entries) {
@@ -77,7 +197,7 @@ object RollbackCommand : BuildableCommand {
                     {
                         Component.translatable(
                             "text.ledger.rollback.finish",
-                            actions.size,
+                            applied.toString(),
                         ).setStyle(TextColorPallet.primary)
                     },
                     true,
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/database/ActionQueueService.kt b/src/main/kotlin/com/github/quiltservertools/ledger/database/ActionQueueService.kt
index 508d8b27..4fec317f 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/database/ActionQueueService.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/database/ActionQueueService.kt
@@ -5,14 +5,31 @@ import com.github.quiltservertools.ledger.actions.ActionType
 import com.github.quiltservertools.ledger.config.DatabaseSpec
 import com.github.quiltservertools.ledger.utility.ticks
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 import java.util.concurrent.LinkedBlockingQueue
 
 object ActionQueueService {
+    // Reoptimization: shutdown drain uses larger chunks - there is no tick
+    // pressure while stopping, so fewer transactions finish the drain faster.
+    private const val SHUTDOWN_BATCH_MULTIPLIER = 4
+    private const val MIN_SHUTDOWN_BATCH_SIZE = 5000
+
     private val queue = LinkedBlockingQueue<ActionType>()
     private lateinit var job: Job
 
+    /**
+     * Reoptimization: passes since the last time a batch was filled.
+     *
+     * 0 means actions are arriving at least as fast as they are written (burst mode:
+     * drain without waiting). Any higher value means the server has caught up, so the
+     * configured batchDelay applies again.
+     */
+    @Volatile
+    private var consecutiveShortPasses = 0
+
     val size: Int get() = queue.size
 
     fun start() {
@@ -29,24 +46,70 @@ object ActionQueueService {
 
     suspend fun drainAll() {
         job.cancel()
+        // Reoptimization: at shutdown there is no tick pressure - flush in larger
+        // chunks (4x the runtime batch size, at least 5000) so the queue drains
+        // with fewer transactions and the server stops waiting less time.
+        val shutdownBatchSize = maxOf(
+            Ledger.config[DatabaseSpec.batchSize] * SHUTDOWN_BATCH_MULTIPLIER,
+            MIN_SHUTDOWN_BATCH_SIZE,
+        )
         while (queue.isNotEmpty()) {
-            drainBatch()
+            drainBatch(shutdownBatchSize)
         }
     }
 
-    private suspend fun drainBatch() {
+    /**
+     * Writes one batch.
+     *
+     * @return true when a full [size] batch was taken, i.e. the queue is being fed at
+     *         least as fast as it is drained
+     */
+    private suspend fun drainBatch(size: Int = Ledger.config[DatabaseSpec.batchSize]): Boolean {
         val batch = mutableListOf<ActionType>()
-        queue.drainTo(batch, Ledger.config[DatabaseSpec.batchSize])
+        queue.drainTo(batch, size)
+        val saturated = batch.size >= size
 
-        DatabaseManager.logActionBatch(batch)
+        // Reoptimization: NonCancellable so a shutdown-time cancellation can never
+        // abort a half-written batch. The rows were already removed from the queue,
+        // so a cancelled write would silently drop logged actions.
+        withContext(NonCancellable) {
+            DatabaseManager.logActionBatch(batch)
+        }
+        return saturated
     }
 
     private suspend fun prepareNextBatch() {
         job = Ledger.launch {
-            if (queue.size < Ledger.config[DatabaseSpec.batchSize]) {
-                delay(Ledger.config[DatabaseSpec.batchDelay].ticks)
+            val batchSize = Ledger.config[DatabaseSpec.batchSize].coerceAtLeast(1)
+            val maxDelayTicks = Ledger.config[DatabaseSpec.batchDelay].coerceAtLeast(1)
+
+            // Reoptimization: drain at full speed while the queue is being flooded, and
+            // keep the configured cadence once it is not.
+            //
+            // Measured on a 13,500-action burst: the unconditional `delay(batchDelay)`
+            // burned 6.1 s of an 8.7 s ingest window while the writes themselves took
+            // 2.6 s per pass - the scheduler, not the database, was the bottleneck,
+            // because every pass paid the full 500 ms even with work already queued.
+            //
+            // burstMode means the previous pass filled a complete batch, i.e. actions
+            // arrive faster than they are written. In that state a partial queue is the
+            // tail of a flood, so waiting for it would only add latency. Once a pass
+            // comes up short the server is caught up and the original wait applies, so
+            // trickle traffic (a few actions at a time) still batches exactly as
+            // upstream does - this deliberately does not turn an idle server into one
+            // small transaction per action.
+            val burstMode = consecutiveShortPasses == 0
+            if (queue.size < batchSize && !burstMode) {
+                delay(maxDelayTicks.ticks)
+            }
+
+            if (queue.isNotEmpty()) {
+                // Logging happens in the called transaction; see drainBatch.
+                val saturated = drainBatch(batchSize)
+                // saturated => still flooded, keep burstMode; otherwise count the misses
+                // so a quiet server settles onto the configured delay.
+                consecutiveShortPasses = if (saturated) 0 else consecutiveShortPasses + 1
             }
-            if (queue.isNotEmpty()) drainBatch()
             prepareNextBatch()
         }
     }
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseCacheService.kt b/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseCacheService.kt
index 969a4d95..ff6ae7cb 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseCacheService.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseCacheService.kt
@@ -17,4 +17,10 @@ object DatabaseCacheService {
     val playerKeys: BiMap<UUID, Int> = HashBiMap.create()
 
     val playernameKeys: BiMap<String, Int> = HashBiMap.create()
+
+    /**
+     * Reoptimization: dictionary-encoded block states (state string <-> block_states.id).
+     * Keeps the actions table rows narrow (int ref instead of a full state string per row).
+     */
+    val blockStateKeys: BiMap<String, Int> = HashBiMap.create()
 }
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseManager.kt b/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseManager.kt
index bb6a5f58..e765f837 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseManager.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/database/DatabaseManager.kt
@@ -21,10 +21,16 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.newSingleThreadContext
+import kotlinx.coroutines.withContext
 import net.minecraft.core.BlockPos
 import net.minecraft.resources.Identifier
 import net.minecraft.server.players.NameAndId
 import org.jetbrains.exposed.v1.core.Column
+import org.jetbrains.exposed.v1.core.EqOp
+import org.jetbrains.exposed.v1.core.GreaterOp
+import org.jetbrains.exposed.v1.core.IntegerColumnType
+import org.jetbrains.exposed.v1.core.LessOp
+import org.jetbrains.exposed.v1.core.LiteralOp
 import org.jetbrains.exposed.v1.core.Op
 import org.jetbrains.exposed.v1.core.SortOrder
 import org.jetbrains.exposed.v1.core.SqlLogger
@@ -47,6 +53,7 @@ import org.jetbrains.exposed.v1.dao.Entity
 import org.jetbrains.exposed.v1.dao.EntityClass
 import org.jetbrains.exposed.v1.dao.IntEntityClass
 import org.jetbrains.exposed.v1.jdbc.Database
+import org.jetbrains.exposed.v1.jdbc.JdbcTransaction
 import org.jetbrains.exposed.v1.jdbc.Query
 import org.jetbrains.exposed.v1.jdbc.SchemaUtils
 import org.jetbrains.exposed.v1.jdbc.andWhere
@@ -75,6 +82,41 @@ const val MIN_RETRY_DELAY = 1000L
 const val MAX_RETRY_DELAY = 300_000L
 private const val MAX_EXTRA_DATA_BYTES = 65_535
 
+// Reoptimization: TTL for the search count cache
+private const val COUNT_CACHE_TTL_MS = 30_000L
+private const val COUNT_CACHE_MAX_ENTRIES = 256
+
+// Reoptimization: SQLite tuning (WAL + NORMAL + generous busy timeout)
+private const val SQLITE_BUSY_TIMEOUT_MS = 10_000
+
+// Reoptimization: rows converted per statement while backfilling actions.time_ms.
+// Small enough that each commit is short, large enough to finish a typical
+// database in a handful of statements.
+private const val TIME_MS_BACKFILL_BATCH = 20_000
+
+/**
+ * Reoptimization: the exact column list and order used by [DatabaseManager.insertActions].
+ * The two legacy TEXT state columns are omitted on purpose - new rows store the
+ * dictionary reference instead, so they stay NULL, and SQLite fills them in.
+ */
+private const val INSERT_ACTIONS_SQL =
+    "INSERT INTO actions (action_id, \"time\", time_ms, x, y, z, object_id, old_object_id, " +
+        "world_id, block_state_ref, old_block_state_ref, \"source\", player_id, extra_data, " +
+        "rolled_back) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
+
+// Reoptimization: SQLite connection tuning. The cache size is expressed in KiB and is
+// negative to mean "KiB rather than pages"; the other two are byte counts.
+private const val SQLITE_CACHE_SIZE_KIB = -16384
+private const val SQLITE_MMAP_SIZE_BYTES = "268435456"
+private const val SQLITE_JOURNAL_SIZE_LIMIT_BYTES = 67108864
+
+// Reoptimization: keyset-pagination helpers. SqlExpressionBuilder is deprecated (ERROR-level)
+// in Exposed 1.0.0, so comparison predicates for cursor pagination are built directly.
+private fun intLiteral(value: Int) = LiteralOp(IntegerColumnType(), value)
+
+// Reoptimization: kept as a single facade object so the fork stays drop-in with
+// upstream call sites; splitting it would ripple through every caller.
+@Suppress("LargeClass")
 object DatabaseManager {
 
     // These values are initialised late to allow the database to be created at server start,
@@ -84,6 +126,26 @@ object DatabaseManager {
         get() = database.dialect.name
 
     private val cache = DatabaseCacheService
+
+    /**
+     * Reoptimization: true once the `time_ms` column exists. The insert path only
+     * writes it when true, so a database running with `updateSchema = false` (where
+     * the migration never runs) keeps working on the original schema.
+     */
+    @Volatile
+    private var timeMsColumnPresent = false
+
+    /**
+     * Reoptimization: true once every row carries an epoch-millisecond `time_ms`.
+     * While false (i.e. a backfill is still in progress) time filters use the legacy
+     * TEXT column so results stay correct on a partially migrated database.
+     */
+    @Volatile
+    private var timeMsReady = false
+
+    // Reoptimization: kept so vacuumDatabase() can open an autocommit connection
+    // (VACUUM cannot run inside a transaction).
+    private var compactDataSource: DataSource? = null
     private var databaseContext = Dispatchers.IO + CoroutineName("Ledger Database")
     private val ledgerLogger = object : SqlLogger {
         override fun log(context: StatementContext, transaction: Transaction) {
@@ -94,9 +156,12 @@ object DatabaseManager {
     @OptIn(ExperimentalCoroutinesApi::class, DelicateCoroutinesApi::class)
     fun setup(dataSource: DataSource?) {
         if (dataSource == null) {
-            database = Database.connect(getDefaultDatasource())
+            val default = getDefaultDatasource()
+            compactDataSource = default
+            database = Database.connect(default)
             databaseContext = newSingleThreadContext("Ledger Database")
         } else {
+            compactDataSource = dataSource
             database = Database.connect(dataSource)
         }
     }
@@ -106,6 +171,26 @@ object DatabaseManager {
         return SQLiteDataSource(
             SQLiteConfig().apply {
                 setJournalMode(SQLiteConfig.JournalMode.WAL)
+                // Reoptimization: WAL + NORMAL is the standard high-throughput combo.
+                // Commits no longer fsync per transaction (only at checkpoints),
+                // which massively reduces batch write latency without risking corruption.
+                setSynchronous(SQLiteConfig.SynchronousMode.NORMAL)
+                setBusyTimeout(SQLITE_BUSY_TIMEOUT_MS)
+
+                // Reoptimization: read-path tuning. These are applied by SQLiteConfig to
+                // every connection the pool opens, so they survive reconnects (unlike a
+                // one-off PRAGMA statement, which would only affect a single connection).
+                //  - temp_store=MEMORY: ORDER BY / GROUP BY spills stay in RAM instead of
+                //    hitting disk, which matters for the large sorts a rollback does.
+                //  - cache_size: 16 MiB page cache (the value is in KiB, negative = KiB).
+                //  - mmap_size: 256 MiB of memory-mapped I/O, so page reads during
+                //    lookups avoid a copy into the page cache.
+                //  - journal_size_limit: cap the WAL at 64 MiB after a checkpoint so a
+                //    long bulk insert cannot leave an unbounded -wal file behind.
+                setTempStore(SQLiteConfig.TempStore.MEMORY)
+                setCacheSize(SQLITE_CACHE_SIZE_KIB)
+                setPragma(SQLiteConfig.Pragma.MMAP_SIZE, SQLITE_MMAP_SIZE_BYTES)
+                setJournalSizeLimit(SQLITE_JOURNAL_SIZE_LIMIT_BYTES)
             },
         ).apply {
             url = "jdbc:sqlite:$dbFilepath"
@@ -121,17 +206,264 @@ object DatabaseManager {
             Tables.ObjectIdentifiers,
             Tables.Sources,
             Tables.Worlds,
+            Tables.BlockStates,
         )
+
+        // Reoptimization: make sure the dictionary-encoding reference columns exist.
+        // ALTER TABLE ADD COLUMN (nullable, no default rewrite) is instant on both
+        // SQLite and MySQL 8+ INSTANT, and fully backwards compatible.
+        ensureBlockStateColumns()
+
         if (config[DatabaseSpec.updateSchema]) {
             try {
-                exec("CREATE INDEX IF NOT EXISTS actions_time ON actions(time)")
+                // Reoptimization: ensureTimeMs replaces the old TEXT `actions_time` index
+                // with an INTEGER `actions_time_ms` one, so the legacy
+                // `CREATE INDEX IF NOT EXISTS actions_time` that used to live here is gone -
+                // re-asserting it on every start would undo the migration.
+                ensureTimeMs()
+                ensurePerformanceIndexes()
             } catch (e: java.sql.SQLException) {
-                logWarn("Could not create actions_time index (MySQL 8.0.12+ required if using MySQL): ${e.message}")
+                logWarn("Could not run schema migration: ${e.message}")
             }
         }
         logInfo("Tables created")
     }
 
+    /**
+     * Reoptimization: adds block_state_ref / old_block_state_ref to the actions table
+     * when missing (metadata-driven, idempotent).
+     */
+    private fun JdbcTransaction.ensureBlockStateColumns() {
+        val columns = existingColumnNames("actions")
+        if ("block_state_ref" !in columns) {
+            exec("ALTER TABLE actions ADD COLUMN block_state_ref INTEGER")
+        }
+        if ("old_block_state_ref" !in columns) {
+            exec("ALTER TABLE actions ADD COLUMN old_block_state_ref INTEGER")
+        }
+    }
+
+    private fun JdbcTransaction.existingColumnNames(table: String): Set<String> {
+        val isSQLite = databaseType.equals("SQLite", ignoreCase = true)
+        val sql = if (isSQLite) "PRAGMA table_info($table)" else "SHOW COLUMNS FROM $table"
+        val column = if (isSQLite) "name" else "Field"
+        val names = mutableSetOf<String>()
+        exec(sql) { rs ->
+            while (rs.next()) names.add(rs.getString(column))
+        }
+        return names
+    }
+
+    /**
+     * Reoptimization: integer time index.
+     *
+     * The `time` column stores TEXT ("2026-10-05 13:22:21.702"). An index over it costs
+     * ~35 bytes per row - measured at 19.6% of the whole database on a 13,500-row
+     * workload, making it the single largest object after the table itself. The same
+     * instant as epoch milliseconds needs ~15 bytes, so `time_ms` carries the index and
+     * every time filter uses it.
+     *
+     * Migration steps, all idempotent and resumable:
+     *   1. add `time_ms` (INTEGER NOT NULL DEFAULT 0) if missing - instant in SQLite
+     *   2. create the integer index if missing
+     *   3. backfill every row still at the sentinel 0, in id-ordered batches
+     *   4. once no sentinel rows remain, drop the superseded TEXT index
+     *
+     * Until step 3 completes the flag [timeMsReady] stays false and queries fall back
+     * to the TEXT column, so a partially migrated database still returns correct
+     * results and the backfill simply resumes on the next start.
+     */
+    private fun JdbcTransaction.ensureTimeMs() {
+        val columns = existingColumnNames("actions")
+
+        if ("time_ms" !in columns) {
+            exec("ALTER TABLE actions ADD COLUMN time_ms INTEGER NOT NULL DEFAULT 0")
+            logInfo("Added actions.time_ms column (integer time index)")
+        }
+        timeMsColumnPresent = true
+
+        val isSQLite = databaseType.equals("SQLite", ignoreCase = true)
+        val indexes = mutableSetOf<String>()
+        if (isSQLite) {
+            exec("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='actions'") { rs ->
+                while (rs.next()) indexes.add(rs.getString("name"))
+            }
+        }
+        if ("actions_time_ms" !in indexes) {
+            exec("CREATE INDEX actions_time_ms ON actions(time_ms)")
+            logInfo("Created actions_time_ms index")
+        }
+
+        // Backfill: strftime is not used because it drops the fractional part, and
+        // julianday keeps millisecond accuracy well within double precision here.
+        // Verified against the TEXT values on a 13,500-row database: exact match on
+        // every row, 1:1 distinct mapping, ordering preserved.
+        var remaining = countSentinelTimeRows()
+        if (remaining > 0L) {
+            logInfo("Backfilling time_ms for $remaining actions (one-off migration)")
+            while (remaining > 0L) {
+                val issued = backfillTimeMsBatch(TIME_MS_BACKFILL_BATCH)
+                val now = if (issued) countSentinelTimeRows() else remaining
+                // Stop when the statement could not be issued, or when a pass converted
+                // nothing (defensive: never spin on a database we cannot make progress on).
+                val progressed = issued && now < remaining
+                if (!progressed) {
+                    logWarn("time_ms backfill stopped early; affected rows stay on the TEXT fallback")
+                }
+                remaining = if (progressed) now else 0L
+            }
+            logInfo("time_ms backfill finished, $remaining rows left unconverted")
+        }
+
+        timeMsReady = countSentinelTimeRows() == 0L
+
+        // The TEXT index is dead weight once every query uses time_ms.
+        if (timeMsReady && "actions_time" in indexes) {
+            try {
+                if (isSQLite) {
+                    exec("DROP INDEX IF EXISTS actions_time")
+                } else {
+                    exec("DROP INDEX actions_time ON actions")
+                }
+                logInfo("Dropped superseded TEXT time index (replaced by actions_time_ms)")
+                // The dropped index leaves its pages on the freelist; only VACUUM returns
+                // them to the filesystem. /ledger compact does exactly that.
+                logInfo("Run \"/ledger compact\" to reclaim the freed index space on disk")
+            } catch (e: java.sql.SQLException) {
+                logWarn("Could not drop actions_time (harmless, leaving in place): ${e.message}")
+            }
+        }
+    }
+
+    private fun JdbcTransaction.countSentinelTimeRows(): Long {
+        var count = 0L
+        exec("SELECT COUNT(*) FROM actions WHERE time_ms = 0") { rs ->
+            if (rs.next()) count = rs.getLong(1)
+        }
+        return count
+    }
+
+    /**
+     * Converts one id-ordered slice of legacy rows. The threshold is the smallest
+     * sentinel id plus the batch size, so every still-unconverted row below it is
+     * written in a single statement.
+     *
+     * @return true when a statement was issued (i.e. sentinel rows still exist), false
+     *         when there is nothing left to do
+     */
+    private fun JdbcTransaction.backfillTimeMsBatch(batchSize: Int): Boolean {
+        var upper = 0
+        exec("SELECT id FROM actions WHERE time_ms = 0 ORDER BY id LIMIT 1") { rs ->
+            if (rs.next()) upper = rs.getInt(1)
+        }
+        if (upper == 0) return false
+
+        exec(
+            "UPDATE actions SET time_ms = " +
+                "CAST(ROUND((julianday(time) - 2440587.5) * 86400000.0) AS INTEGER) " +
+                "WHERE time_ms = 0 AND id < ${upper + batchSize}",
+        )
+        return true
+    }
+
+    /**
+     * Reoptimization: makes sure the upstream index set is present and removes the
+     * composite (dimension, time) indexes that an earlier revision of this branch
+     * created. Ledger stores timestamps as TEXT, so each extra index column costs
+     * ~23 bytes per row; measured over 13,500 rows those composites grew the index
+     * footprint from ~1.45 MB to ~3.29 MB (+78% of the whole database) while
+     * producing no measurable lookup improvement, so they were reverted.
+     *
+     * Metadata-driven, idempotent and safe to re-run. Only executed when the user
+     * opts in via updateSchema.
+     */
+    private fun JdbcTransaction.ensurePerformanceIndexes() {
+        val isSQLite = databaseType.equals("SQLite", ignoreCase = true)
+
+        fun existingIndexes(): Set<String> {
+            val names = mutableSetOf<String>()
+            if (isSQLite) {
+                exec("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='actions'") { rs ->
+                    while (rs.next()) names.add(rs.getString("name"))
+                }
+            } else {
+                exec("SHOW INDEX FROM actions") { rs ->
+                    while (rs.next()) names.add(rs.getString("Key_name"))
+                }
+            }
+            return names
+        }
+
+        val existing = existingIndexes()
+
+        // Upstream index set. Exposed creates these via the table definition; they are
+        // re-asserted here so databases that were created by an intermediate build
+        // (which dropped them in favour of composites) are repaired on startup.
+        val wantedIndexes = listOf(
+            "actions_action_id" to "CREATE INDEX actions_action_id ON actions(action_id)",
+            "actions_object_id" to "CREATE INDEX actions_object_id ON actions(object_id)",
+            "actions_old_object_id" to "CREATE INDEX actions_old_object_id ON actions(old_object_id)",
+            "actions_player_id" to "CREATE INDEX actions_player_id ON actions(player_id)",
+            "actions_by_location" to "CREATE INDEX actions_by_location ON actions(x, y, z, world_id)",
+        )
+        for ((name, ddl) in wantedIndexes) {
+            if (name !in existing) {
+                try {
+                    exec(ddl)
+                } catch (e: java.sql.SQLException) {
+                    logWarn("Could not create index $name: ${e.message}")
+                }
+            }
+        }
+
+        // Drop the composite indexes an earlier revision of this branch introduced.
+        val superseded = listOf(
+            "actions_player_time",
+            "actions_action_time",
+            "actions_object_time",
+            "actions_old_object_time",
+        )
+        for (name in superseded) {
+            if (name in existing) {
+                try {
+                    if (isSQLite) {
+                        exec("DROP INDEX IF EXISTS $name")
+                    } else {
+                        exec("DROP INDEX $name ON actions")
+                    }
+                } catch (e: java.sql.SQLException) {
+                    logWarn("Could not drop index $name (harmless, leaving in place): ${e.message}")
+                }
+            }
+        }
+
+        // An earlier revision also widened actions_by_location to (world, x, z, time).
+        // Rebuild it in the upstream shape when the wider form is detected.
+        if (isSQLite && "actions_by_location" in existing) {
+            var storedSql: String? = null
+            try {
+                exec(
+                    "SELECT sql FROM sqlite_master WHERE type='index' AND name='actions_by_location'",
+                ) { rs ->
+                    if (rs.next()) storedSql = rs.getString(1)
+                }
+            } catch (e: java.sql.SQLException) {
+                logWarn("Could not read actions_by_location definition: ${e.message}")
+            }
+            val sql = storedSql
+            val normalised = sql?.replace(Regex("\\s+"), " ")
+            if (normalised != null && "(world_id, x, z" in normalised) {
+                try {
+                    exec("DROP INDEX IF EXISTS actions_by_location")
+                    exec("CREATE INDEX actions_by_location ON actions(x, y, z, world_id)")
+                    logInfo("Rebuilt actions_by_location in the upstream (x, y, z, world) shape")
+                } catch (e: java.sql.SQLException) {
+                    logWarn("Could not rebuild actions_by_location: ${e.message}")
+                }
+            }
+        }
+    }
+
     suspend fun setupCache() {
         execute {
             Tables.ActionIdentifier.all().forEach {
@@ -146,6 +478,9 @@ object DatabaseManager {
             Tables.Source.all().forEach {
                 cache.sourceKeys[it.name] = it.id.value
             }
+            Tables.BlockState.all().forEach {
+                cache.blockStateKeys[it.state] = it.id.value
+            }
             Tables.Player.all().forEach {
                 cache.playerKeys[it.playerId] = it.id.value
                 cache.playernameKeys.forcePut(it.playerName, it.id.value)
@@ -211,6 +546,156 @@ object DatabaseManager {
         return@execute getActionsFromQuery(query)
     }
 
+    // ------------------------------------------------------------------
+    // Reoptimization: streaming rollback/restore support
+    // ------------------------------------------------------------------
+
+    /**
+     * Reoptimization: current maximum actions.id. Used as an upper bound for
+     * streaming rollbacks so that blocks logged *by* the rollback itself can
+     * never be picked up by subsequent batches of the same rollback.
+     */
+    suspend fun currentMaxActionId(): Int = execute {
+        var maxId = 0
+        exec("SELECT COALESCE(MAX(id), 0) FROM ${Tables.Actions.tableName}") { rs ->
+            if (rs.next()) maxId = rs.getInt(1)
+        }
+        maxId
+    }
+
+    /**
+     * Reoptimization: count of actions matching the params (optionally filtered by
+     * rolled-back state). Used for progress reporting without loading rows.
+     */
+    suspend fun countActionsFor(params: ActionSearchParams, rolledBack: Boolean? = null): Long = execute {
+        var op = buildQueryParams(params)
+        if (rolledBack != null) {
+            op = op and (Tables.Actions.rolledBack eq rolledBack)
+        }
+        return@execute Tables.Actions.selectAll().where(op).count()
+    }
+
+    /**
+     * Reoptimization: one batch of rollback candidates ordered newest-first,
+     * using a keyset cursor (id < [exclusiveUpperId]) instead of loading the
+     * entire result set into memory.
+     */
+    suspend fun selectRollbackBatch(params: ActionSearchParams, exclusiveUpperId: Int, limit: Int): List<ActionType> =
+        execute {
+            val query = Tables.Actions
+                .selectAll()
+                .where(
+                    buildQueryParams(params) and (Tables.Actions.rolledBack eq false) and
+                        LessOp(Tables.Actions.id, intLiteral(exclusiveUpperId)),
+                )
+                .orderBy(Tables.Actions.id, SortOrder.DESC)
+                .limit(limit)
+            return@execute getActionsFromQuery(query)
+        }
+
+    /**
+     * Reoptimization: one batch of restore candidates ordered oldest-first (keyset cursor).
+     */
+    suspend fun selectRestoreBatch(params: ActionSearchParams, exclusiveLowerId: Int, limit: Int): List<ActionType> =
+        execute {
+            val query = Tables.Actions
+                .selectAll()
+                .where(
+                    buildQueryParams(params) and (Tables.Actions.rolledBack eq true) and
+                        GreaterOp(Tables.Actions.id, intLiteral(exclusiveLowerId)),
+                )
+                .orderBy(Tables.Actions.id, SortOrder.ASC)
+                .limit(limit)
+            return@execute getActionsFromQuery(query)
+        }
+
+    /**
+     * Reoptimization: rewrites legacy text block states into dictionary references
+     * in id-ordered batches, so existing databases shrink once the rows are
+     * rewritten and vacuumed. Idempotent: rows already migrated are skipped.
+     *
+     * @return number of rows migrated
+     */
+    suspend fun compactBlockStates(
+        batchSize: Int = 5000,
+        onProgress: suspend (done: Long, total: Long) -> Unit = { _, _ -> },
+    ): Long {
+        // Reoptimization: native SQL keeps this maintenance path independent of
+        // expression-builder API churn. Cursor values are server-generated ints.
+        val total = execute {
+            var count = 0L
+            exec(
+                "SELECT COUNT(*) FROM ${Tables.Actions.tableName} " +
+                    "WHERE (block_state IS NOT NULL OR old_block_state IS NOT NULL)",
+            ) { rs ->
+                if (rs.next()) count = rs.getLong(1)
+            }
+            count
+        }
+        if (total == 0L) return 0L
+
+        var migrated = 0L
+        var cursor = 0
+        while (true) {
+            // Read batch (materialised before the write transaction starts)
+            val rows = execute {
+                val batch = mutableListOf<Triple<Int, String?, String?>>()
+                exec(
+                    "SELECT id, block_state, old_block_state FROM ${Tables.Actions.tableName} " +
+                        "WHERE (block_state IS NOT NULL OR old_block_state IS NOT NULL) AND id > $cursor " +
+                        "ORDER BY id ASC LIMIT $batchSize",
+                ) { rs ->
+                    while (rs.next()) {
+                        batch.add(
+                            Triple(rs.getInt("id"), rs.getString("block_state"), rs.getString("old_block_state")),
+                        )
+                    }
+                }
+                batch
+            }
+            if (rows.isEmpty()) break
+
+            execute {
+                for ((id, state, oldState) in rows) {
+                    val newRef = state?.let { getOrCreateBlockStateId(it) }
+                    val newOldRef = oldState?.let { getOrCreateBlockStateId(it) }
+                    Tables.Actions.update({ EqOp(Tables.Actions.id, intLiteral(id)) }) {
+                        it[Tables.Actions.blockStateRef] = newRef
+                        it[Tables.Actions.oldBlockStateRef] = newOldRef
+                        it[Tables.Actions.blockState] = null
+                        it[Tables.Actions.oldBlockState] = null
+                    }
+                }
+            }
+
+            migrated += rows.size
+            cursor = rows.last().first
+            onProgress(migrated, total)
+        }
+        return migrated
+    }
+
+    /**
+     * Reoptimization: reclaims disk space after compaction (VACUUM on SQLite,
+     * OPTIMIZE TABLE on MySQL). Must run outside a transaction.
+     */
+    suspend fun vacuumDatabase() {
+        val dataSource = compactDataSource ?: return
+        val isSQLite = databaseType.equals("SQLite", ignoreCase = true)
+        withContext(databaseContext) {
+            dataSource.connection.use { connection ->
+                connection.autoCommit = true
+                connection.createStatement().use { statement ->
+                    if (isSQLite) {
+                        statement.execute("VACUUM")
+                    } else {
+                        statement.execute("OPTIMIZE TABLE ${Tables.Actions.tableName}")
+                    }
+                }
+            }
+        }
+    }
+
     suspend fun previewActions(params: ActionSearchParams, type: Preview.Type): List<ActionType> = execute {
         when (type) {
             Preview.Type.ROLLBACK -> return@execute selectRollback(params)
@@ -218,8 +703,10 @@ object DatabaseManager {
         }
     }
 
-    private fun getActionsFromQuery(query: Query): List<ActionType> {
+    private fun Transaction.getActionsFromQuery(query: Query): List<ActionType> {
         val actions = mutableListOf<ActionType>()
+        val stateRefs = HashMap<ActionType, Int>()
+        val oldStateRefs = HashMap<ActionType, Int>()
 
         val actionIdentifierCache = DatabaseCacheService.actionIdentifierKeys.inverse()
         val worldCache = DatabaseCacheService.worldIdentifierKeys.inverse()
@@ -244,8 +731,12 @@ object DatabaseManager {
             type.world = worldCache[action[Tables.Actions.world].value]
             type.objectIdentifier = objectIdentifierCache[action[Tables.Actions.objectId].value]!!
             type.oldObjectIdentifier = objectIdentifierCache[action[Tables.Actions.oldObjectId].value]!!
+            // Reoptimization: prefer the dictionary ref, fall back to the legacy text
+            // column for rows written before the encoding was introduced.
             type.objectState = action[Tables.Actions.blockState]
             type.oldObjectState = action[Tables.Actions.oldBlockState]
+            action.getOrNull(Tables.Actions.blockStateRef)?.let { stateRefs[type] = it }
+            action.getOrNull(Tables.Actions.oldBlockStateRef)?.let { oldStateRefs[type] = it }
             type.sourceName = sourceCache[action[Tables.Actions.sourceName].value]!!
             type.sourceProfile = action.getOrNull(Tables.Actions.sourcePlayer)?.let {
                 NameAndId(playerCache[it.value]!!, playerNameCache[it.value]!!)
@@ -256,6 +747,20 @@ object DatabaseManager {
             actions.add(type)
         }
 
+        // Resolve dictionary refs in one batch (avoids per-row lookups).
+        val allRefs = (stateRefs.values + oldStateRefs.values).toSet()
+        if (allRefs.isNotEmpty()) {
+            val resolved = resolveBlockStates(allRefs)
+            for (action in actions) {
+                if (action.objectState == null) {
+                    action.objectState = stateRefs[action]?.let { resolved[it] }
+                }
+                if (action.oldObjectState == null) {
+                    action.oldObjectState = oldStateRefs[action]?.let { resolved[it] }
+                }
+            }
+        }
+
         return actions
     }
 
@@ -268,7 +773,25 @@ object DatabaseManager {
             op = op.and { Tables.Actions.z.between(params.bounds.minZ(), params.bounds.maxZ()) }
         }
 
-        if (params.before != null && params.after != null) {
+        // Reoptimization: filter on the INTEGER `time_ms` column whenever the backfill
+        // has completed - its index is ~20 bytes per row cheaper than the TEXT one and
+        // integer comparison beats string comparison. On a database whose backfill is
+        // still running we fall back to the TEXT column so results stay correct.
+        if (timeMsReady) {
+            if (params.before != null && params.after != null) {
+                val afterMs = params.after.toEpochMilli()
+                val beforeMs = params.before.toEpochMilli()
+                op = op.and {
+                    Tables.Actions.timeMs.greaterEq(afterMs) and Tables.Actions.timeMs.lessEq(beforeMs)
+                }
+            } else if (params.before != null) {
+                val beforeMs = params.before.toEpochMilli()
+                op = op.and { Tables.Actions.timeMs.lessEq(beforeMs) }
+            } else if (params.after != null) {
+                val afterMs = params.after.toEpochMilli()
+                op = op.and { Tables.Actions.timeMs.greaterEq(afterMs) }
+            }
+        } else if (params.before != null && params.after != null) {
             op = op.and {
                 Tables.Actions.timestamp.greaterEq(params.after) and Tables.Actions.timestamp.lessEq(params.before)
             }
@@ -418,7 +941,7 @@ object DatabaseManager {
         insertRegKeys(identifiers)
     }
 
-    private suspend fun <T : Any?> execute(body: suspend Transaction.() -> T): T {
+    private suspend fun <T : Any?> execute(body: suspend JdbcTransaction.() -> T): T {
         while (Ledger.server.overworld()?.noSave != false) {
             delay(timeMillis = 1000)
         }
@@ -463,7 +986,7 @@ object DatabaseManager {
         }
     }
 
-    private fun Transaction.insertActions(actions: List<ActionType>) {
+    private fun JdbcTransaction.insertActions(actions: List<ActionType>) {
         val (safe, oversized) = actions.partition {
             it.extraData == null || it.extraData!!.length <= MAX_EXTRA_DATA_BYTES
         }
@@ -475,24 +998,91 @@ object DatabaseManager {
                     "by ${action.sourceProfile?.name ?: action.sourceName}",
             )
         }
-        Tables.Actions.batchInsert(safe, shouldReturnGeneratedValues = false) { action ->
-            this[Tables.Actions.actionIdentifier] = getOrCreateActionId(action.identifier)
-            this[Tables.Actions.timestamp] = action.timestamp
-            this[Tables.Actions.x] = action.pos.x
-            this[Tables.Actions.y] = action.pos.y
-            this[Tables.Actions.z] = action.pos.z
-            this[Tables.Actions.objectId] = getOrCreateRegistryKeyId(action.objectIdentifier)
-            this[Tables.Actions.oldObjectId] = getOrCreateRegistryKeyId(action.oldObjectIdentifier)
-            this[Tables.Actions.world] = getOrCreateWorldId(
-                action.world ?: Ledger.server.overworld().dimension()
-                    .identifier(),
-            )
-            this[Tables.Actions.blockState] = action.objectState
-            this[Tables.Actions.oldBlockState] = action.oldObjectState
-            this[Tables.Actions.sourceName] = getOrCreateSourceId(action.sourceName)
-            this[Tables.Actions.sourcePlayer] = action.sourceProfile?.let { getOrCreatePlayerId(it.id) }
-            this[Tables.Actions.extraData] = action.extraData
+        if (safe.isEmpty()) return
+
+        // Reoptimization: insert through a real JDBC batch instead of Exposed's
+        // per-row statement path.
+        //
+        // Measured: Exposed's batchInsert issued one statement per row - a 13,500-row
+        // workload produced 13,500 "INSERT INTO actions" statements - and cost roughly
+        // 310-500 us per row. A plain JDBC batch on the same schema and the same rows
+        // costs 23 us per row (44 us when executed individually), so most of the insert
+        // time was statement machinery rather than SQLite.
+        //
+        // Values are still produced by Exposed's own column types via valueToDB, so the
+        // on-disk representation (notably the "yyyy-MM-dd HH:mm:ss.SSS" UTC timestamp
+        // text) is byte-for-byte what the ORM path wrote. Only the statement execution
+        // is bypassed, keeping the stored format stable across upgrades.
+        val insertSql = INSERT_ACTIONS_SQL
+
+        // JdbcTransaction wraps a plain JDBC connection; the generic parameter cannot be
+        // inferred from Kotlin here, so make the (always true for JDBC) cast explicit.
+        @Suppress("UNCHECKED_CAST")
+        val connection = this.connection.connection as java.sql.Connection
+        connection.prepareStatement(insertSql).use { ps ->
+            for (action in safe) {
+                var i = 0
+                ps.setObject(++i, getOrCreateActionId(action.identifier))
+                ps.setObject(++i, Tables.Actions.timestamp.columnType.valueToDB(action.timestamp))
+                ps.setObject(++i, action.timestamp.toEpochMilli())
+                ps.setObject(++i, action.pos.x)
+                ps.setObject(++i, action.pos.y)
+                ps.setObject(++i, action.pos.z)
+                ps.setObject(++i, getOrCreateRegistryKeyId(action.objectIdentifier))
+                ps.setObject(++i, getOrCreateRegistryKeyId(action.oldObjectIdentifier))
+                ps.setObject(
+                    ++i,
+                    getOrCreateWorldId(
+                        action.world ?: Ledger.server.overworld().dimension().identifier(),
+                    ),
+                )
+                // Dictionary-encoded block states: rows carry a nullable int reference
+                // into block_states instead of the full state string.
+                ps.setObject(++i, action.objectState?.let { getOrCreateBlockStateId(it) })
+                ps.setObject(++i, action.oldObjectState?.let { getOrCreateBlockStateId(it) })
+                ps.setObject(++i, getOrCreateSourceId(action.sourceName))
+                ps.setObject(++i, action.sourceProfile?.let { getOrCreatePlayerId(it.id) })
+                ps.setObject(++i, action.extraData)
+                ps.setObject(++i, Tables.Actions.rolledBack.columnType.valueToDB(false))
+                ps.addBatch()
+            }
+            ps.executeBatch()
+        }
+    }
+
+    /**
+     * Reoptimization: resolves (or creates) the dictionary id for a block state string.
+     */
+    private fun getOrCreateBlockStateId(state: String): Int = getOrCreateObjectId(
+        state,
+        cache.blockStateKeys,
+        Tables.BlockState,
+        Tables.BlockStates,
+        Tables.BlockStates.state,
+    )
+
+    /**
+     * Reoptimization: batch-resolves block state ids to their strings using the cache,
+     * falling back to a single IN (...) query for cache misses.
+     */
+    private fun Transaction.resolveBlockStates(ids: Set<Int>): Map<Int, String> {
+        val result = mutableMapOf<Int, String>()
+        val misses = mutableSetOf<Int>()
+        val inverse = cache.blockStateKeys.inverse()
+        for (id in ids) {
+            inverse[id]?.let { result[id] = it } ?: misses.add(id)
         }
+        if (misses.isNotEmpty()) {
+            Tables.BlockStates.selectAll()
+                .where { Tables.BlockStates.id inList misses }
+                .forEach { row ->
+                    val id = row[Tables.BlockStates.id].value
+                    val state = row[Tables.BlockStates.state]
+                    result[id] = state
+                    cache.blockStateKeys[state] = id
+                }
+        }
+        return result
     }
 
     private fun Transaction.insertOrUpdatePlayer(uuid: UUID, name: String) {
@@ -512,6 +1102,21 @@ object DatabaseManager {
         }
     }
 
+    // Reoptimization: short-lived cache for search result counts, so paging
+    // through results does not re-run the full count query on every page.
+    private val countCache = java.util.concurrent.ConcurrentHashMap<ActionSearchParams, Pair<Long, Long>>()
+
+    private fun cachedCount(params: ActionSearchParams, recompute: () -> Long): Long {
+        val now = System.currentTimeMillis()
+        countCache[params]?.let { (count, expiresAt) ->
+            if (now < expiresAt) return count
+        }
+        val count = recompute()
+        if (countCache.size > COUNT_CACHE_MAX_ENTRIES) countCache.clear()
+        countCache[params] = count to (now + COUNT_CACHE_TTL_MS)
+        return count
+    }
+
     private fun Transaction.selectActionsSearch(params: ActionSearchParams, page: Int): SearchResults {
         val actions = mutableListOf<ActionType>()
 
@@ -519,7 +1124,13 @@ object DatabaseManager {
             .selectAll()
             .andWhere { buildQueryParams(params) }
 
-        val totalActions: Long = countActions(params)
+        // Reoptimization: served from the count cache while paging (TTL 30s)
+        val totalActions: Long = cachedCount(params) {
+            Tables.Actions
+                .selectAll()
+                .andWhere { buildQueryParams(params) }
+                .count()
+        }
         if (totalActions == 0L) return SearchResults(actions, params, page, 0)
 
         query = query.orderBy(Tables.Actions.id, SortOrder.DESC)
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/database/Tables.kt b/src/main/kotlin/com/github/quiltservertools/ledger/database/Tables.kt
index ed6f634a..045509d7 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/database/Tables.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/database/Tables.kt
@@ -13,6 +13,7 @@ private const val MAX_PLAYER_NAME_LENGTH = 16
 private const val MAX_ACTION_NAME_LENGTH = 16
 private const val MAX_IDENTIFIER_LENGTH = 191
 private const val MAX_SOURCE_NAME_LENGTH = 30
+private const val MAX_BLOCK_STATE_LENGTH = 500
 
 object Tables {
     object Players : IntIdTable("players") {
@@ -53,23 +54,65 @@ object Tables {
         companion object : IntEntityClass<ObjectIdentifier>(ObjectIdentifiers)
     }
 
+    /**
+     * Reoptimization: dictionary table for block state strings.
+     * The actions table stores an int reference instead of repeating the full state
+     * string on every row (CoreProtect-style dictionary encoding).
+     */
+    object BlockStates : IntIdTable("block_states") {
+        val state = varchar("state", MAX_BLOCK_STATE_LENGTH).uniqueIndex()
+    }
+
+    class BlockState(id: EntityID<Int>) : IntEntity(id) {
+        var state by BlockStates.state
+
+        companion object : IntEntityClass<BlockState>(BlockStates)
+    }
+
     object Actions : IntIdTable("actions") {
         val actionIdentifier = reference("action_id", ActionIdentifiers.id).index()
-        val timestamp = timestamp("time").index("actions_time")
+        val timestamp = timestamp("time")
         val x = integer("x")
         val y = integer("y")
         val z = integer("z")
         val world = reference("world_id", Worlds.id)
         val objectId = reference("object_id", ObjectIdentifiers.id).index()
         val oldObjectId = reference("old_object_id", ObjectIdentifiers.id).index()
+
+        /**
+         * Reoptimization: the same instant as [timestamp], stored as epoch milliseconds.
+         *
+         * This exists purely for indexing. The legacy `time` column holds TEXT
+         * ("2026-10-05 13:22:21.702", ~23 bytes), and an index keyed on it costs
+         * ~35 bytes per row - 20% of the whole database on a 13,500-row workload.
+         * An INTEGER key costs ~15 bytes per row, so all time filtering is done on
+         * this column while `time` is kept in sync for display and compatibility.
+         */
+        val timeMs = long("time_ms").index("actions_time_ms")
+
+        // Reoptimization: nullable legacy text columns kept for backwards compatibility.
+        // New writes store the dictionary id in the *_ref columns and null here.
         val blockState = text("block_state").nullable()
         val oldBlockState = text("old_block_state").nullable()
+        val blockStateRef = integer("block_state_ref").nullable()
+        val oldBlockStateRef = integer("old_block_state_ref").nullable()
+
         val sourceName = reference("source", Sources.id).index()
         val sourcePlayer = optReference("player_id", Players.id).index()
         val extraData = text("extra_data").nullable()
         val rolledBack = bool("rolled_back").clientDefault { false }
 
         init {
+            // Reoptimization note: the composite (dimension, time) indexes that an
+            // earlier revision of this branch introduced were reverted. Ledger stored
+            // the timestamp as TEXT, so every additional index column carried ~23
+            // bytes per row; measured on a 13,500-row workload they inflated the
+            // index footprint from ~1.45 MB to ~3.29 MB (+78% total file size) with
+            // no measurable lookup win.
+            //
+            // The time column is the one place where that cost was worth removing
+            // rather than avoiding: `time_ms` (INTEGER) now carries the index, which
+            // is ~20 bytes per row cheaper than indexing the TEXT form.
             index("actions_by_location", false, x, y, z, world)
         }
     }
@@ -77,6 +120,7 @@ object Tables {
     class Action(id: EntityID<Int>) : IntEntity(id) {
         var actionIdentifier by ActionIdentifier referencedOn Actions.actionIdentifier
         var timestamp by Actions.timestamp
+        var timeMs by Actions.timeMs
         var x by Actions.x
         var y by Actions.y
         var z by Actions.z
diff --git a/src/main/kotlin/com/github/quiltservertools/ledger/utility/MessageUtils.kt b/src/main/kotlin/com/github/quiltservertools/ledger/utility/MessageUtils.kt
index 8188c68f..a0921738 100644
--- a/src/main/kotlin/com/github/quiltservertools/ledger/utility/MessageUtils.kt
+++ b/src/main/kotlin/com/github/quiltservertools/ledger/utility/MessageUtils.kt
@@ -26,7 +26,7 @@ import kotlin.time.toKotlinDuration
 object MessageUtils {
     val pageChangeAction: Identifier = Ledger.identifier("page-change")
     val teleportAction: Identifier = Ledger.identifier("teleport")
-    
+
     @OptIn(ExperimentalTime::class)
     suspend fun sendSearchResults(source: CommandSourceStack, results: SearchResults, header: Component) {
         // If the player has a Ledger compatible client, we send results as action packets rather than as chat messages
@@ -50,17 +50,17 @@ object MessageUtils {
             Component.translatable(
                 "text.ledger.footer.search",
                 Component.translatable(
-                    "text.ledger.footer.page_backward"
+                    "text.ledger.footer.page_backward",
                 ).setStyle(TextColorPallet.primaryVariant).withStyle {
-                    val tag: CompoundTag = CompoundTag().apply { this.putInt("page", results.page-1) }
-                    
+                    val tag: CompoundTag = CompoundTag().apply { this.putInt("page", results.page - 1) }
+
                     if (results.page > 1) {
                         it.withHoverEvent(
                             HoverEvent.ShowText(
-                                Component.translatable("text.ledger.footer.page_backward.hover")
-                            )
+                                Component.translatable("text.ledger.footer.page_backward.hover"),
+                            ),
                         ).withClickEvent(
-                            ClickEvent.Custom(pageChangeAction, Optional.of(tag))
+                            ClickEvent.Custom(pageChangeAction, Optional.of(tag)),
                         )
                     } else {
                         Style.EMPTY
@@ -69,23 +69,23 @@ object MessageUtils {
                 results.page.toString().literal().setStyle(TextColorPallet.primaryVariant),
                 results.pages.toString().literal().setStyle(TextColorPallet.primaryVariant),
                 Component.translatable(
-                    "text.ledger.footer.page_forward"
+                    "text.ledger.footer.page_forward",
                 ).setStyle(TextColorPallet.primaryVariant).withStyle {
-                    val tag: CompoundTag = CompoundTag().apply { this.putInt("page", results.page+1) }
+                    val tag: CompoundTag = CompoundTag().apply { this.putInt("page", results.page + 1) }
 
                     if (results.page < results.pages) {
                         it.withHoverEvent(
                             HoverEvent.ShowText(
-                                Component.translatable("text.ledger.footer.page_forward.hover")
-                            )
+                                Component.translatable("text.ledger.footer.page_forward.hover"),
+                            ),
                         ).withClickEvent(
-                            ClickEvent.Custom(pageChangeAction, Optional.of(tag))
+                            ClickEvent.Custom(pageChangeAction, Optional.of(tag)),
                         )
                     } else {
                         Style.EMPTY
                     }
-                }
-            ).setStyle(TextColorPallet.primary)
+                },
+            ).setStyle(TextColorPallet.primary),
         )
     }
 
@@ -135,8 +135,8 @@ object MessageUtils {
         message.withStyle {
             it.withHoverEvent(
                 HoverEvent.ShowText(
-                    timeMessage
-                )
+                    timeMessage,
+                ),
             )
         }
         return message
diff --git a/src/main/resources/data/ledger/lang/en_gb.json b/src/main/resources/data/ledger/lang/en_gb.json
index dd7149ee..90804f15 100644
--- a/src/main/resources/data/ledger/lang/en_gb.json
+++ b/src/main/resources/data/ledger/lang/en_gb.json
@@ -27,10 +27,12 @@
     "text.ledger.action_message.with": "with",
 
     "text.ledger.rollback.start": "Rolling back %s actions",
+    "text.ledger.rollback.progress": "Rollback in progress: %s / %s actions",
     "text.ledger.rollback.finish": "Rollback complete",
     "text.ledger.rollback.fail": "Failed to rollback %s x%s",
 
     "text.ledger.restore.start": "Restoring %s actions",
+    "text.ledger.restore.progress": "Restore in progress: %s / %s actions",
     "text.ledger.restore.finish": "Restore complete",
     "text.ledger.restore.fail": "Failed to restore %s x%s",
 
@@ -61,5 +63,9 @@
     "text.ledger.network.no_mod_info": "Unable to determine mod information for your client mod",
 
     "text.ledger.purge.starting": "------ Starting purge ------",
-    "text.ledger.purge.complete": "------ Completed purge ------"
+    "text.ledger.purge.complete": "------ Completed purge ------",
+
+    "text.ledger.compact.starting": "------ Compacting database (this may take a while) ------",
+    "text.ledger.compact.progress": "Compact in progress: %s / %s rows",
+    "text.ledger.compact.complete": "------ Compacted %s actions and vacuumed the database ------"
 }
\ No newline at end of file
diff --git a/src/main/resources/data/ledger/lang/en_us.json b/src/main/resources/data/ledger/lang/en_us.json
index b40f7d0c..a76c8cf3 100644
--- a/src/main/resources/data/ledger/lang/en_us.json
+++ b/src/main/resources/data/ledger/lang/en_us.json
@@ -27,10 +27,12 @@
     "text.ledger.action_message.with": "with",
 
     "text.ledger.rollback.start": "Rolling back %s actions",
+    "text.ledger.rollback.progress": "Rollback in progress: %s / %s actions",
     "text.ledger.rollback.finish": "Rollback complete",
     "text.ledger.rollback.fail": "Failed to rollback %s x%s",
 
     "text.ledger.restore.start": "Restoring %s actions",
+    "text.ledger.restore.progress": "Restore in progress: %s / %s actions",
     "text.ledger.restore.finish": "Restore complete",
     "text.ledger.restore.fail": "Failed to restore %s x%s",
 
@@ -76,5 +78,9 @@
     "text.ledger.purge.starting": "------ Starting purge ------",
     "text.ledger.purge.complete": "------ Completed purge ------",
 
+    "text.ledger.compact.starting": "------ Compacting database (this may take a while) ------",
+    "text.ledger.compact.progress": "Compact in progress: %s / %s rows",
+    "text.ledger.compact.complete": "------ Compacted %s actions and vacuumed the database ------",
+
     "text.ledger.player.result": "%1$s: First joined %2$s. Last joined: %3$s"
 }
\ No newline at end of file
diff --git a/src/main/resources/data/ledger/lang/zh_cn.json b/src/main/resources/data/ledger/lang/zh_cn.json
index d71f5d78..d1e4f351 100644
--- a/src/main/resources/data/ledger/lang/zh_cn.json
+++ b/src/main/resources/data/ledger/lang/zh_cn.json
@@ -25,10 +25,12 @@
   "text.ledger.action_message.location.hover": "单击传送至此",
 
   "text.ledger.rollback.start": "正在回滚，共 %s 次更改",
+  "text.ledger.rollback.progress": "回滚进行中: 已完成 %s / %s 次更改",
   "text.ledger.rollback.finish": "完成了指定规则内的回滚",
   "text.ledger.rollback.fail": "回滚失败 %s x%s",
 
   "text.ledger.restore.start": "正在还原 %s 次回档更改",
+  "text.ledger.restore.progress": "还原进行中: 已完成 %s / %s 次更改",
   "text.ledger.restore.finish": "完成了指定规则内的还原",
   "text.ledger.restore.fail": "还原失败 %s x%s",
 
@@ -50,5 +52,9 @@
   "text.ledger.action.item-remove": "拿取",
   "text.ledger.action.entity-kill": "杀死",
 
+  "text.ledger.compact.starting": "------ 正在压缩数据库（可能需要一些时间）------",
+  "text.ledger.compact.progress": "压缩进行中: 已处理 %s / %s 行",
+  "text.ledger.compact.complete": "------ 已压缩 %s 条记录并完成数据库瘦身 ------",
+
   "text.ledger.network.protocols_mismatched": "协议版本不匹配: 客户端协议版本 %d 服务端版本为 %d"
 }
diff --git a/src/main/resources/data/ledger/lang/zh_tw.json b/src/main/resources/data/ledger/lang/zh_tw.json
index 99f71e51..8d898c91 100644
--- a/src/main/resources/data/ledger/lang/zh_tw.json
+++ b/src/main/resources/data/ledger/lang/zh_tw.json
@@ -27,10 +27,12 @@
   "text.ledger.action_message.with": "和",
 
   "text.ledger.rollback.start": "正在復原 %s 項動作",
+  "text.ledger.rollback.progress": "復原進行中: 已完成 %s / %s 項動作",
   "text.ledger.rollback.finish": "復原完成",
   "text.ledger.rollback.fail": "復原失敗 %s ×%s",
 
   "text.ledger.restore.start": "正在還原 %s 項動作",
+  "text.ledger.restore.progress": "還原進行中: 已完成 %s / %s 項動作",
   "text.ledger.restore.finish": "還原完成",
   "text.ledger.restore.fail": "還原失敗 %s ×%s",
 
@@ -62,5 +64,9 @@
   "text.ledger.purge.starting": "------ 開始清除 ------",
   "text.ledger.purge.complete": "------ 完成清除 ------",
 
+  "text.ledger.compact.starting": "------ 正在壓縮資料庫（可能需要一些時間）------",
+  "text.ledger.compact.progress": "壓縮進行中: 已處理 %s / %s 行",
+  "text.ledger.compact.complete": "------ 已壓縮 %s 項記錄並完成資料庫瘦身 ------",
+
   "text.ledger.player.result": "%1$s：第一次加入 %2$s。最後加入：%3$s"
 }