diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index a4b9de4..a6148b5 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -37,5 +37,8 @@ jobs:
       - name: Run the demo
         run: crystal run examples/demo.cr
 
+      - name: Run the server demo
+        run: crystal run examples/server_demo.cr
+
       - name: Build the binary
         run: shards build --production
diff --git a/README.md b/README.md
index d9f2522..090b396 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,9 @@ restart. It serves `get`, `put`, and `delete` over a local socket.
 Snapshots give you a consistent view of the store at one instant. A snapshot
 never blocks new writes. Release it when you are done.
 
-This is release 0.3.0. It adds point-in-time snapshots and consistent reads.
+This is release 0.4.0. It adds incremental compaction by level. Older data
+settles into deeper levels, so each compaction rewrites a small part of the
+store instead of the whole store.
 
 ## Features
 
@@ -22,6 +24,7 @@ This is release 0.3.0. It adds point-in-time snapshots and consistent reads.
 - Sorted tables with a block index and a bloom filter
 - Block cache for fast repeated reads
 - Background flush and compaction
+- Incremental compaction by level
 - Point-in-time snapshots with consistent reads
 - Snapshot iterators that stay valid during writes
 - Range scans with an iterator API
@@ -56,10 +59,11 @@ bin/cinderstore demo
 
 The demo loads a product catalog from `fixtures/catalog.csv`. It writes,
 scans, flushes, compacts, deletes, snapshots, and reopens a database. The
-output is deterministic.
+final step shows how leveled compaction pushes old data into deeper levels.
+The output is deterministic.
 
 ```text
-== Cinderstore 0.3.0 demo ==
+== Cinderstore 0.4.0 demo ==
 
 Loaded 24 products from ...\fixtures\catalog.csv
 Database directory: ...\cinderstore-demo
@@ -106,12 +110,24 @@ Database directory: ...\cinderstore-demo
 8. Reopen the database and verify recovery
    rows after restart: 21
 
+9. Leveled compaction pushes older data into deeper levels
+   round 1: tables: 1 (l0: 0, l1: 1), entries: 24
+   round 2: tables: 2 (l0: 0, l1: 1, l2: 1), entries: 48
+   round 3: tables: 3 (l0: 0, l1: 1, l2: 2), entries: 72
+   round 4: tables: 4 (l0: 0, l1: 1, l2: 3), entries: 96
+   after a final compact: tables: 4 (l0: 0, l1: 1, l2: 3), entries: 107
+   rows: 107, checkpoint: 42
+
 Demo complete.
 ```
 
 Step 7 shows the value of a snapshot. The live store drops SKU-0001 and adds
 two products. The snapshot still sees the state before those writes.
 
+Step 9 uses a small `base_level_bytes` target. Round 1 fills level 1. Later
+rounds overflow it, so the store moves the oldest tables into level 2. The
+checkpoint write flushes into level 0 and compacts into the lower levels.
+
 ## Use the library
 
 Require the library.
@@ -181,7 +197,7 @@ Flush and compact explicitly.
 
 ```crystal
 db.flush    # Move the memtable into a table.
-db.compact  # Merge tables and drop deleted keys.
+db.compact  # Merge tables across the levels.
 ```
 
 ## Command line tool
@@ -283,10 +299,22 @@ file. The old log is deleted only after the file is durable.
 
 ### Compaction
 
-Level-0 tables may overlap. Compaction merges every table into a fresh,
-non-overlapping level-1 set. The merge keeps the newest entry for each key.
-It drops tombstones, because it includes all data. New writes continue into
-the memory table during the merge.
+Tables live in levels. Level 0 holds the newest tables. Those tables may
+overlap, because each flush appends a fresh table. Levels 1 and deeper hold
+sorted tables with disjoint key ranges.
+
+Compaction works on one level at a time. It never rewrites the whole store.
+
+- When level 0 grows past its threshold, a merge moves its tables into
+  level 1. Overlapping level-1 tables join the merge.
+- When a deeper level exceeds its byte target, one table merges into the
+  next level. Tables that overlap its key range join the merge.
+- Each level target grows by `level_multiplier`. Old data settles into the
+  deeper levels over time.
+
+A merge keeps the newest entry for each key. A tombstone is dropped only
+when the merge reaches the bottom level. In a higher level a tombstone is
+kept, because older data may still live below it.
 
 Compaction keeps a table file on disk while a snapshot references it. The
 file is deleted only after the last snapshot releases it.
@@ -341,9 +369,15 @@ config.cache_blocks = 512
 config.sync_writes = true
 config.l0_compact_threshold = 4
 config.compact_on_flush = true
+config.base_level_bytes = 4_i64 * 1024 * 1024
+config.level_multiplier = 4
 db = Cinderstore::DB.new("data", config)
 ```
 
+`base_level_bytes` is the target size of level 1. `level_multiplier` is the
+size ratio between two neighboring levels. Set both smaller to compact more
+often. Set them larger to keep more tables on disk.
+
 ## Project layout
 
 ```text
@@ -359,14 +393,20 @@ spec/                    Test suite
 
 ## Test status
 
-The suite runs with `crystal spec`. It has 97 examples. All pass on Windows
+The suite runs with `crystal spec`. It has 107 examples. All pass on Windows
 and Linux. It covers the skip list, the memory table, the write ahead log,
 the bloom filter, and the block cache. It covers the tables, the iterators,
-and the database. It covers compaction, durability, snapshots, and the
-server protocol.
+and the database. It covers compaction, leveled compaction, durability,
+snapshots, and the server protocol.
+
+The leveled compaction tests build multiple levels and verify them. They
+check that levels stay non-overlapping. They check that tombstones survive
+in high levels and vanish at the bottom. A randomized test compares the
+store against a reference model across writes, deletes, flushes, and
+compactions.
 
 The CI workflow runs on GitHub Actions for Windows and Ubuntu. It checks
-formatting, runs the suite, runs the demo, and builds the binary.
+formatting, runs the suite, runs the demos, and builds the binary.
 
 ## Limitations
 
@@ -374,7 +414,8 @@ formatting, runs the suite, runs the demo, and builds the binary.
 - Values are limited to 4 MB.
 - Keys are limited to 4 KB.
 - The server protocol is unencrypted. Use it on localhost only.
-- Compaction is a full merge. It is correct and simple, not incremental.
+- Compaction is level based, not size based. Tables split at a fixed entry
+  count, so a level can exceed its byte target briefly.
 - No multi-threaded runtime is required. The server uses fibers.
 - Release snapshots before you close the database.
 
@@ -382,13 +423,15 @@ formatting, runs the suite, runs the demo, and builds the binary.
 
 Planned:
 
-- Release 0.2: incremental compaction by level
-- Release 0.4: optional checksum-free fast mode
-- Release 0.5: batch writes and group commit
-- Release 0.6: secondary indexes
+- Release 0.5: optional checksum-free fast mode
+- Release 0.6: batch writes and group commit
+- Release 0.7: secondary indexes
 
 Delivered:
 
+- Release 0.4: incremental compaction by level. Tables settle into levels
+  with disjoint key ranges. Compaction moves one level at a time and keeps
+  tombstones until the bottom level.
 - Release 0.3: snapshot iterators and consistent reads. Snapshots give a
   stable view of the store. Compaction keeps referenced files alive.
 
diff --git a/shard.lock b/shard.lock
index c1dbcd2..8a8a3ae 100644
--- a/shard.lock
+++ b/shard.lock
@@ -1 +1 @@
-version: 0.3.0
+version: 0.4.0
diff --git a/shard.yml b/shard.yml
index e335eb5..e532dec 100644
--- a/shard.yml
+++ b/shard.yml
@@ -1,8 +1,9 @@
 name: cinderstore
-version: 0.3.0
+version: 0.4.0
 description: An embeddable key/value store built on a log structured merge tree.
 crystal: ">= 1.10.0"
 license: Apache-2.0
+repository: github:DanielCuevas1208/cinderstore
 authors:
   - Cinderstore Contributors
 
diff --git a/spec/leveled_compaction_spec.cr b/spec/leveled_compaction_spec.cr
new file mode 100644
index 0000000..2dcfcae
--- /dev/null
+++ b/spec/leveled_compaction_spec.cr
@@ -0,0 +1,294 @@
+require "./spec_helper"
+
+# Grants a test direct control over the table layout. The database opens
+# against an empty directory first; the test then writes table files and
+# injects references to them.
+class LeveledHarness < Cinderstore::DB
+  # Replaces the level layout. The caller must create every referenced
+  # table file first.
+  def inject_levels!(levels : Array(Array(Cinderstore::DB::TableRef)))
+    @lock.synchronize do
+      @levels = levels
+      @seq = 10_000_i64
+      @next_id = 20_000_i64
+    end
+  end
+end
+
+# Writes one sorted table into `path` and returns a live table reference.
+private def make_table_ref(path : String, id : Int64, entries : Array(Cinderstore::Entry))
+  file = File.join(path, "#{Cinderstore::Util.file_stem(id)}.sst")
+  meta = nil
+  File.open(file, "w") do |io|
+    writer = Cinderstore::SstableWriter.new(io, id, 4096, 0.01)
+    entries.each { |e| writer.add(e.key, e.value, e.seq, e.alive) }
+    meta = writer.finish
+  end
+  m = meta.not_nil!
+  Cinderstore::DB::TableRef.new(m.id, m.first, m.last, m.count, file)
+end
+
+# Polls until a condition holds, then fails if it does not within 5 seconds.
+private def wait_for_compact(timeout : Time::Span = 5.seconds, &block : -> Bool) : Nil
+  deadline = Time.instant + timeout
+  until block.call
+    raise "condition not met in time" if Time.instant >= deadline
+    sleep 10.milliseconds
+  end
+end
+
+# Writes `count` keys with values of about `value_size` bytes.
+private def write_batch(db : Cinderstore::DB, prefix : String, count : Int32, value_size : Int32) : Nil
+  count.times { |i| db.put("#{prefix}-%04d" % i, "v" * value_size) }
+end
+
+describe "Cinderstore leveled compaction" do
+  it "drains level-0 tables into level 1" do
+    config = Cinderstore::SpecHelpers.fast_config
+    Cinderstore::SpecHelpers.with_db("lvl-drain", config) do |db, _path|
+      db.put("a", "1")
+      db.flush
+      db.put("b", "2")
+      db.flush
+      db.stats.l0.should eq(2)
+
+      db.compact
+      db.stats.l0.should eq(0)
+      db.stats.l1.should eq(1)
+      db.stats.levels.should eq([0, 1])
+      db.scan.map(&.[0]).should eq(%w[a b])
+    end
+  end
+
+  it "builds multiple levels when data exceeds the level targets" do
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 500
+    config.level_multiplier = 4
+    Cinderstore::SpecHelpers.with_db("lvl-multilevel", config) do |db, _path|
+      5.times do |round|
+        write_batch(db, "k#{round}", 30, 50)
+        db.flush
+        write_batch(db, "k#{round}b", 30, 50)
+        db.flush
+        db.compact
+      end
+      stats = db.stats
+      stats.l0.should eq(0)
+      stats.levels.size.should be >= 3
+      stats.levels[1..].sum.should be > 0
+      db.scan.size.should eq(5 * 60)
+    end
+  end
+
+  it "keeps every level except level 0 non-overlapping" do
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 500
+    config.level_multiplier = 4
+    Cinderstore::SpecHelpers.with_db_path("lvl-nonoverlap") do |path|
+      db = Cinderstore::DB.new(path, config)
+      begin
+        4.times do |round|
+          write_batch(db, "r#{round}", 40, 50)
+          db.flush
+          write_batch(db, "r#{round}b", 40, 50)
+          db.flush
+          db.compact
+        end
+        manifest = Cinderstore::Manifest.load(File.join(path, "MANIFEST"))
+        manifest.levels.size.should be >= 3
+        manifest.levels[1..].each do |level|
+          (0...(level.size - 1)).each do |i|
+            level[i].last.should be < level[i + 1].first
+          end
+        end
+        db.scan.size.should eq(4 * 80)
+      ensure
+        db.close rescue nil
+      end
+    end
+  end
+
+  it "keeps tombstones while a deeper level still holds older data" do
+    path = Cinderstore::SpecHelpers.tmp_db_path("lvl-keep")
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 1
+    config.level_multiplier = 1_000_000
+    db = LeveledHarness.new(path, config)
+    begin
+      top = make_table_ref(path, 100_i64, [
+        Cinderstore::Entry.new("aaa", 30_i64, false, ""),
+        Cinderstore::Entry.new("bbb", 31_i64, true, "2"),
+      ])
+      middle = make_table_ref(path, 200_i64, [
+        Cinderstore::Entry.new("aaa", 20_i64, true, "v2"),
+      ])
+      bottom = make_table_ref(path, 300_i64, [
+        Cinderstore::Entry.new("aaa", 10_i64, true, "v1"),
+      ])
+      db.inject_levels!([[] of Cinderstore::DB::TableRef, [top], [middle], [bottom]])
+
+      db.compact
+      # Only level 1 moved into level 2. The tombstone had to survive,
+      # because level 3 still stores an older version of "aaa".
+      db.stats.levels.should eq([0, 0, 1, 1])
+      db.get("aaa").should be_nil
+      db.get("bbb").should eq("2")
+    ensure
+      db.close rescue nil
+      FileUtils.rm_rf(path)
+    end
+  end
+
+  it "drops tombstones when the target level is the bottom" do
+    path = Cinderstore::SpecHelpers.tmp_db_path("lvl-drop")
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 1
+    config.level_multiplier = 1_000_000
+    db = LeveledHarness.new(path, config)
+    begin
+      top = make_table_ref(path, 100_i64, [
+        Cinderstore::Entry.new("aaa", 30_i64, false, ""),
+        Cinderstore::Entry.new("bbb", 31_i64, true, "2"),
+      ])
+      bottom = make_table_ref(path, 200_i64, [
+        Cinderstore::Entry.new("aaa", 10_i64, true, "v1"),
+        Cinderstore::Entry.new("ccc", 12_i64, true, "3"),
+      ])
+      db.inject_levels!([[] of Cinderstore::DB::TableRef, [top], [bottom]])
+
+      db.compact
+      # The tombstone and the old value cancel out, so "aaa" vanishes.
+      db.stats.levels.should eq([0, 0, 1])
+      db.stats.entries.should eq(2)
+      db.get("aaa").should be_nil
+      db.get("bbb").should eq("2")
+      db.get("ccc").should eq("3")
+      db.scan.map(&.[0]).should eq(%w[bbb ccc])
+    ensure
+      db.close rescue nil
+      FileUtils.rm_rf(path)
+    end
+  end
+
+  it "honors the configured threshold for background compaction" do
+    config = Cinderstore::SpecHelpers.fast_config
+    config.compact_on_flush = true
+    config.l0_compact_threshold = 3
+    Cinderstore::SpecHelpers.with_db("lvl-background", config) do |db, _path|
+      db.put("a", "1")
+      db.flush
+      db.put("b", "2")
+      db.flush
+      db.stats.l0.should eq(2)
+
+      db.put("c", "3")
+      db.flush
+      wait_for_compact { db.stats.l0 == 0 }
+      db.stats.l1.should eq(1)
+      db.scan.map(&.[0]).should eq(%w[a b c])
+    end
+  end
+
+  it "restores the level layout after a restart" do
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 800
+    config.level_multiplier = 4
+    Cinderstore::SpecHelpers.with_db_path("lvl-recovery") do |path|
+      db = Cinderstore::DB.new(path, config)
+      3.times do |round|
+        write_batch(db, "r#{round}", 40, 50)
+        db.flush
+        write_batch(db, "r#{round}b", 40, 50)
+        db.flush
+        db.compact
+      end
+      before = db.stats.levels
+      db.close
+
+      reopened = Cinderstore::DB.new(path, config)
+      reopened.stats.levels.should eq(before)
+      reopened.scan.size.should eq(3 * 80)
+      reopened.close
+    end
+  end
+
+  it "keeps deep table files alive while a snapshot holds them" do
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 600
+    config.level_multiplier = 4
+    Cinderstore::SpecHelpers.with_db("lvl-snapdeep", config) do |db, _path|
+      2.times do |round|
+        write_batch(db, "r#{round}", 30, 50)
+        db.flush
+        write_batch(db, "r#{round}b", 30, 50)
+        db.flush
+        db.compact
+      end
+      snap = db.snapshot
+      snapshot_rows = snap.count
+
+      write_batch(db, "r2", 30, 50)
+      db.flush
+      write_batch(db, "r2b", 30, 50)
+      db.flush
+      db.compact
+
+      snap.count.should eq(snapshot_rows)
+      db.scan.size.should eq(3 * 60)
+      snap.release
+      db.get("r2-0000").should eq("v" * 50)
+    end
+  end
+
+  it "matches a reference model across random writes, deletes, flushes, and compactions" do
+    config = Cinderstore::SpecHelpers.fast_config
+    config.base_level_bytes = 800
+    config.level_multiplier = 4
+    Cinderstore::SpecHelpers.with_db("lvl-model", config) do |db, _path|
+      rng = Random.new(0xC1DE_5EED)
+      keys = (0...40).map { |i| "key-%03d" % i }
+      model = {} of String => String
+
+      300.times do
+        case rng.rand(10)
+        when 0, 1, 2, 3, 4
+          key = keys[rng.rand(keys.size)]
+          value = "v-#{rng.rand(10_000)}"
+          db.put(key, value)
+          model[key] = value
+        when 5, 6
+          key = keys[rng.rand(keys.size)]
+          db.delete(key)
+          model.delete(key)
+        when 7
+          db.flush
+        when 8
+          db.compact
+        end
+        if rng.rand(25) == 0
+          db.scan.should eq(model.to_a.sort)
+        end
+      end
+
+      db.flush
+      db.compact
+      db.scan.should eq(model.to_a.sort)
+      db.stats.l0.should eq(0)
+    end
+  end
+
+  it "reports per-level counts in stats" do
+    config = Cinderstore::SpecHelpers.fast_config
+    Cinderstore::SpecHelpers.with_db("lvl-stats", config) do |db, _path|
+      db.put("a", "1")
+      db.flush
+      db.put("b", "2")
+      db.flush
+      db.compact
+      stats = db.stats
+      stats.levels.should eq([0, 1])
+      stats.to_h["levels"].should eq([0, 1])
+      stats.to_json.should contain("\"levels\"")
+    end
+  end
+end
diff --git a/src/cinderstore/db.cr b/src/cinderstore/db.cr
index f5707f9..3e340ea 100644
--- a/src/cinderstore/db.cr
+++ b/src/cinderstore/db.cr
@@ -15,6 +15,9 @@ module Cinderstore
     MAX_TABLE_ENTRIES = 20_000
     MAX_KEY_BYTES     =   4096
     MAX_VALUE_BYTES   = 4 * 1024 * 1024
+    # Compaction never creates levels at or above this index. The deepest
+    # level absorbs tombstones and is always compacted last.
+    MAX_LEVELS = 12
 
     # Tuning options for a database instance.
     class Config
@@ -32,6 +35,10 @@ module Cinderstore
       property l0_compact_threshold : Int32 = 4
       # Start background compaction after a flush when true.
       property compact_on_flush : Bool = true
+      # Target bytes for level 1. Deeper levels grow by `level_multiplier`.
+      property base_level_bytes : Int64 = 4_i64 * 1024 * 1024
+      # Byte ratio between consecutive non-overlapping levels.
+      property level_multiplier : Int32 = 4
     end
 
     # A snapshot of database counters for reporting.
@@ -39,6 +46,7 @@ module Cinderstore
       getter tables : Int32
       getter l0 : Int32
       getter l1 : Int32
+      getter levels : Array(Int32)
       getter entries : Int64
       getter disk_bytes : Int64
       getter memtable_bytes : Int64
@@ -47,14 +55,16 @@ module Cinderstore
       getter cache_hits : Int64
       getter cache_misses : Int64
 
-      def initialize(@tables : Int32, @l0 : Int32, @l1 : Int32, @entries : Int64,
-                     @disk_bytes : Int64, @memtable_bytes : Int64, @wal_bytes : Int64,
-                     @seq : Int64, @cache_hits : Int64, @cache_misses : Int64)
+      def initialize(@tables : Int32, @l0 : Int32, @l1 : Int32, @levels : Array(Int32),
+                     @entries : Int64, @disk_bytes : Int64, @memtable_bytes : Int64,
+                     @wal_bytes : Int64, @seq : Int64, @cache_hits : Int64,
+                     @cache_misses : Int64)
       end
 
-      def to_h : Hash(String, Int32 | Int64)
+      def to_h : Hash(String, Int32 | Int64 | Array(Int32))
         {
           "tables"         => @tables,
+          "levels"         => @levels,
           "l0"             => @l0,
           "l1"             => @l1,
           "entries"        => @entries,
@@ -72,7 +82,12 @@ module Cinderstore
       end
 
       def to_s(io : IO) : Nil
-        io << "tables: #{@tables} (l0: #{@l0}, l1: #{@l1})\n"
+        io << "tables: #{@tables} ("
+        @levels.each_with_index do |count, i|
+          io << ", " if i > 0
+          io << "l#{i}: #{count}"
+        end
+        io << ")\n"
         io << "entries: #{@entries}\n"
         io << "disk bytes: #{@disk_bytes}\n"
         io << "memtable bytes: #{@memtable_bytes}\n"
@@ -160,6 +175,9 @@ module Cinderstore
     @compacting = false
     @pending_table_id : Int64 = 0_i64
     @manifest : Manifest? = nil
+    # Round-robin cursor per level. It spreads deep compaction across the
+    # tables of a level instead of always rewriting the leftmost range.
+    @compact_cursor : Hash(Int32, Int32) = Hash(Int32, Int32).new(0)
 
     def initialize(path : String, config : Config = Config.new)
       @path = path
@@ -243,10 +261,13 @@ module Cinderstore
       trigger_compact
     end
 
-    # Merges all tables into a fresh, non-overlapping level-1 set.
+    # Runs leveled compaction until every level fits its target.
+    #
+    # A manual run drains level-0 once two tables exist there. Deeper
+    # levels are compacted when they exceed their byte target.
     def compact : Nil
       @flush_lock.synchronize do
-        do_compact
+        do_compact(false)
       end
     end
 
@@ -290,6 +311,7 @@ module Cinderstore
           tables: table_refs.size,
           l0: @levels[0].size,
           l1: @levels.size > 1 ? @levels[1].size : 0,
+          levels: @levels.map(&.size),
           entries: entries,
           disk_bytes: disk_bytes,
           memtable_bytes: @mem.approximate_bytes,
@@ -339,6 +361,7 @@ module Cinderstore
         end
       end
       @levels = [[] of TableRef] if @levels.empty?
+      trim_empty_levels
     end
 
     private def clean_orphans : Nil
@@ -485,8 +508,9 @@ module Cinderstore
 
     private def trigger_compact : Nil
       return unless @config.compact_on_flush
-      return if @levels[0].size < @config.l0_compact_threshold
       return if @compacting
+      needed = @lock.synchronize { !select_background_compaction.nil? }
+      return unless needed
       @compacting = true
       spawn do
         begin
@@ -501,30 +525,185 @@ module Cinderstore
     # Compaction
     # ------------------------------------------------------------------
 
-    private def do_compact : Nil
-      tables = nil
+    # Compacts levels until no level needs it.
+    #
+    # A manual run drains level-0 once two tables exist there. The
+    # background trigger waits for `l0_compact_threshold` tables instead.
+    private def do_compact(background : Bool) : Nil
+      loop do
+        level : Int32? = nil
+        @lock.synchronize do
+          unless @closed
+            level = background ? select_background_compaction : select_manual_compaction
+          end
+        end
+        break if @closed
+        break if level.nil?
+        compact_level(level)
+      end
+    end
+
+    # Returns the next level a background run should compact, or nil.
+    private def select_background_compaction : Int32?
+      return 0 if @levels[0].size >= @config.l0_compact_threshold
+      select_overfull_level
+    end
+
+    # Returns the next level a manual run should compact, or nil.
+    private def select_manual_compaction : Int32?
+      return 0 if @levels[0].size >= 2
+      select_overfull_level
+    end
+
+    # Returns the lowest non-zero level over its byte target, or nil.
+    private def select_overfull_level : Int32?
+      (1...MAX_LEVELS).each do |i|
+        return i if level_bytes(i) >= level_target(i)
+      end
+      nil
+    end
+
+    # Compacts one level into the next one down.
+    private def compact_level(level : Int32) : Nil
+      if level == 0
+        compact_level0
+      else
+        compact_deep_level(level)
+      end
+    end
+
+    # Merges every level-0 table into level 1, together with any level-1
+    # tables that overlap their combined key span.
+    private def compact_level0 : Nil
+      inputs = nil
+      overlaps = [] of TableRef
+      bottom = false
       @lock.synchronize do
         return if @closed
-        tables = @levels.flatten
-        return if tables.size < 2
+        level0 = @levels[0]
+        return if level0.size < 2
+        inputs = level0.dup
+        span_first = inputs.not_nil!.min_of(&.first)
+        span_last = inputs.not_nil!.max_of(&.last)
+        if level1 = @levels[1]?
+          overlaps = level1.select { |ref| ref.last >= span_first && ref.first <= span_last }
+        end
+        bottom = @levels.size <= 2
       end
+      tables = inputs.not_nil! + overlaps
+      outputs = merge_tables(tables, drop_tombstones: bottom)
+      install_compact_outputs(0, inputs.not_nil!, overlaps, 1, outputs)
+    end
 
-      outputs = merge_tables(tables.not_nil!)
+    # Merges one table from `level` into `level + 1`, together with any
+    # `level + 1` tables that overlap its key span.
+    private def compact_deep_level(level : Int32) : Nil
+      input = nil
+      overlaps = [] of TableRef
+      bottom = false
+      @lock.synchronize do
+        return if @closed
+        tables = @levels[level]?
+        return unless tables && !tables.empty?
+        index = compaction_index(level, tables.size)
+        input = tables[index]
+        target = level + 1
+        if next_level = @levels[target]?
+          overlaps = next_level.select { |ref| ref.last >= input.not_nil!.first && ref.first <= input.not_nil!.last }
+        end
+        bottom = @levels.size <= target + 1 || target >= MAX_LEVELS
+      end
+      tables = [input.not_nil!] + overlaps
+      outputs = merge_tables(tables, drop_tombstones: bottom)
+      install_compact_outputs(level, [input.not_nil!], overlaps, level + 1, outputs)
+    end
 
+    # Removes `removed` and `overlaps` and installs `outputs` in the
+    # target level. Removed tables are orphaned, so their files live until
+    # every snapshot reference is released.
+    private def install_compact_outputs(source_level : Int32, removed : Array(TableRef),
+                                        overlaps : Array(TableRef), target_level : Int32,
+                                        outputs : Array(TableMeta)) : Nil
       @lock.synchronize do
-        @levels = [[] of TableRef, outputs.map { |m| TableRef.new(m.id, m.first, m.last, m.count, table_path(m.id)) }]
+        return if @closed
+        removed.each do |ref|
+          @levels[source_level].reject! { |candidate| candidate.same?(ref) }
+        end
+        if target_level < @levels.size
+          overlaps.each do |ref|
+            @levels[target_level].reject! { |candidate| candidate.same?(ref) }
+          end
+        end
+        ensure_level(target_level)
+        outputs.each do |meta|
+          @levels[target_level] << TableRef.new(meta.id, meta.first, meta.last, meta.count, table_path(meta.id))
+        end
+        sort_level(target_level) if target_level > 0
+        trim_empty_levels
         save_manifest
-        tables.not_nil!.each do |ref|
-          # Mark the table as gone, then drop the database reference. The
-          # file stays on disk until every snapshot reference is released.
+        (removed + overlaps).each do |ref|
           ref.orphan
           ref.release
         end
       end
     end
 
-    # Merges every table into fresh tables and drops tombstones.
-    private def merge_tables(tables : Array(TableRef)) : Array(TableMeta)
+    # Grows the level list until `level` exists.
+    private def ensure_level(level : Int32) : Nil
+      while @levels.size <= level
+        @levels << [] of TableRef
+      end
+    end
+
+    # Drops trailing empty levels so the manifest stays compact.
+    private def trim_empty_levels : Nil
+      while @levels.size > 1 && @levels.last.empty?
+        @levels.pop
+      end
+    end
+
+    # Sorts the tables of one level by their first key. Levels deeper than
+    # level 0 must stay non-overlapping, so an ascending order is useful
+    # for overlap searches and for the manifest.
+    private def sort_level(level : Int32) : Nil
+      @levels[level].sort_by!(&.first)
+    end
+
+    # Returns the next table index to compact in a level. A round-robin
+    # cursor spreads rewrites across the whole level.
+    private def compaction_index(level : Int32, size : Int32) : Int32
+      index = @compact_cursor[level] % size
+      @compact_cursor[level] = (index + 1) % size
+      index
+    end
+
+    # Returns the total bytes of all tables in `level`.
+    private def level_bytes(level : Int32) : Int64
+      tables = @levels[level]?
+      return 0_i64 unless tables
+      tables.sum { |ref| file_size(ref.path) }
+    end
+
+    # Returns the byte target for `level`. Targets grow geometrically, so
+    # each level stores about `level_multiplier` times the one above. The
+    # target saturates instead of overflowing.
+    private def level_target(level : Int32) : Int64
+      factor = Math.max(@config.level_multiplier, 1)
+      target = @config.base_level_bytes
+      (level - 1).times do
+        break if target > Int64::MAX // factor
+        target = target * factor
+      end
+      target
+    end
+
+    # Merges tables into fresh tables.
+    #
+    # `drop_tombstones` is safe only when the output level is the bottom of
+    # the database, because then every older version for each key is part
+    # of the merge. Higher levels keep tombstones, or a delete could be
+    # resurrected by older data still stored below.
+    private def merge_tables(tables : Array(TableRef), drop_tombstones : Bool) : Array(TableMeta)
       readers = @lock.synchronize { tables.map { |ref| ref.reader(@block_cache) } }
       sources = [] of Store::Iter
       readers.each { |reader| sources << TableIter.new(reader) }
@@ -549,19 +728,20 @@ module Cinderstore
       }
 
       while entry = iter.next?
-        next unless entry.alive
-        if writer.nil?
-          current_id = @lock.synchronize do
-            id = @next_id
-            @next_id += 1
-            id
+        if entry.alive || !drop_tombstones
+          if writer.nil?
+            current_id = @lock.synchronize do
+              id = @next_id
+              @next_id += 1
+              id
+            end
+            io = File.open(File.join(@path, "#{Util.file_stem(current_id)}#{TMP_SUFFIX}"), "w")
+            writer = SstableWriter.new(io.not_nil!, current_id, @config.block_size, @config.bloom_fpp)
           end
-          io = File.open(File.join(@path, "#{Util.file_stem(current_id)}#{TMP_SUFFIX}"), "w")
-          writer = SstableWriter.new(io.not_nil!, current_id, @config.block_size, @config.bloom_fpp)
+          writer.not_nil!.add(entry.key, entry.value, entry.seq, entry.alive)
+          written += 1
+          finish_output.call if written >= MAX_TABLE_ENTRIES
         end
-        writer.not_nil!.add(entry.key, entry.value, entry.seq, true)
-        written += 1
-        finish_output.call if written >= MAX_TABLE_ENTRIES
       end
       finish_output.call
       outputs
diff --git a/src/cinderstore/demo.cr b/src/cinderstore/demo.cr
index 206635f..91e8295 100644
--- a/src/cinderstore/demo.cr
+++ b/src/cinderstore/demo.cr
@@ -103,10 +103,51 @@ module Cinderstore
       puts "   rows after restart: #{count}"
       reopened.close
       puts ""
+
+      puts "9. Leveled compaction pushes older data into deeper levels"
+      leveled_config = DB::Config.new
+      leveled_config.sync_writes = false
+      leveled_config.compact_on_flush = false
+      leveled_config.base_level_bytes = 2_000
+      leveled_config.level_multiplier = 4
+      leveled_path = File.join(Dir.tempdir, "cinderstore-levels")
+      FileUtils.rm_rf(leveled_path) if File.exists?(leveled_path)
+      leveled = DB.new(leveled_path, leveled_config)
+      begin
+        4.times do |round|
+          %w[a b].each do |half|
+            12.times do |i|
+              leveled.put("part-#{round}-#{half}-%02d" % i,
+                %({"name":"Part #{round} #{half} #{i}","size":#{i + round * 10}}))
+            end
+            leveled.flush
+          end
+          leveled.compact
+          puts "   round #{round + 1}: #{level_summary(leveled.stats)}"
+        end
+        leveled.put("checkpoint", "42")
+        5.times { |i| leveled.put("post-%02d" % i, "v#{i}") }
+        leveled.flush
+        5.times { |i| leveled.put("post2-%02d" % i, "v#{i}") }
+        leveled.flush
+        leveled.compact
+        puts "   after a final compact: #{level_summary(leveled.stats)}"
+        puts "   rows: #{leveled.scan.size}, checkpoint: #{leveled.get("checkpoint")}"
+      ensure
+        leveled.close
+        FileUtils.rm_rf(leveled_path)
+      end
+      puts ""
+
       puts "Demo complete."
       0
     end
 
+    private def level_summary(stats : DB::Stats) : String
+      counts = stats.levels.map_with_index { |count, i| "l#{i}: #{count}" }.join(", ")
+      "tables: #{stats.tables} (#{counts}), entries: #{stats.entries}"
+    end
+
     private def print_stats(db : DB) : Nil
       stats = db.stats
       puts "   tables: #{stats.tables} (l0: #{stats.l0}, l1: #{stats.l1}), entries: #{stats.entries}"
diff --git a/src/cinderstore/version.cr b/src/cinderstore/version.cr
index 4b9d872..3b31a8c 100644
--- a/src/cinderstore/version.cr
+++ b/src/cinderstore/version.cr
@@ -1,3 +1,3 @@
 module Cinderstore
-  VERSION = "0.3.0"
+  VERSION = "0.4.0"
 end
diff --git a/src/cli.cr b/src/cli.cr
index 8a91070..0d9f869 100644
--- a/src/cli.cr
+++ b/src/cli.cr
@@ -223,7 +223,7 @@ module Cinderstore
           scan        List keys in a range
           stats       Show database counters
           flush       Flush the memtable to a table
-          compact     Merge tables and drop stale data
+          compact     Merge tables across the levels
           demo        Run a self-contained walkthrough
           help        Show this help