From f533a24b29d501b05965c1a48b88cc5d33142fb5 Mon Sep 17 00:00:00 2001
From: DanieCuevas <43822444+DanielCuevas1208@users.noreply.github.com>
Date: Mon, 3 Aug 2026 19:08:01 -0700
Subject: [PATCH] feat: extend cinderstore

---
 .github/workflows/ci.yml   |   6 ++
 README.md                  |  74 +++++++++----
 shard.lock                 |   2 +-
 shard.yml                  |   2 +-
 spec/compaction_spec.cr    |   2 +-
 spec/levels_spec.cr        | 172 ++++++++++++++++++++++++++++++
 src/cinderstore/db.cr      | 209 +++++++++++++++++++++++++++++++++----
 src/cinderstore/demo.cr    |  26 ++++-
 src/cinderstore/version.cr |   2 +-
 9 files changed, 450 insertions(+), 45 deletions(-)
 create mode 100644 spec/levels_spec.cr

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index a4b9de4..2384454 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -4,10 +4,16 @@ on:
   push:
     branches: [main]
   pull_request:
+  schedule:
+    - cron: "17 3 * * *"
 
 permissions:
   contents: read
 
+concurrency:
+  group: ci-${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   test:
     name: Test on ${{ matrix.os }}
diff --git a/README.md b/README.md
index d9f2522..7c4e058 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,8 @@ 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 leveled compaction. Compaction now moves one
+level at a time instead of rewriting every table.
 
 ## Features
 
@@ -22,6 +23,8 @@ 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
+- Leveled compaction that moves one level at a time
+- Tombstones that drop only at the deepest level
 - Point-in-time snapshots with consistent reads
 - Snapshot iterators that stay valid during writes
 - Range scans with an iterator API
@@ -59,7 +62,7 @@ scans, flushes, compacts, deletes, snapshots, and reopens a database. 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
@@ -103,8 +106,16 @@ Database directory: ...\cinderstore-demo
    snapshot iterator SKU-0010 to SKU-0016
    SKU-0010, SKU-0011, SKU-0013, SKU-0014, SKU-0015
 
-8. Reopen the database and verify recovery
-   rows after restart: 21
+8. Leveled compaction grows deeper levels
+   Write 90 products across 3 flushes
+   tables: 5 (l0: 4, l1: 1), entries: 113
+   compact moved level 0 into level 1
+   tables: 1 (l0: 0, l1: 1), entries: 111
+   level 1 passed its target, so compact moved it to level 2
+   tables: 1 (l0: 0, l1: 0, l2: 1), entries: 111
+
+9. Reopen the database and verify recovery
+   rows after restart: 111
 
 Demo complete.
 ```
@@ -112,6 +123,10 @@ 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 8 shows leveled compaction. Three flushes pile up level 0. One compact
+merges them into level 1. The level passes its target. The next compact moves
+it to level 2.
+
 ## Use the library
 
 Require the library.
@@ -181,7 +196,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 one level into the level below.
 ```
 
 ## Command line tool
@@ -283,13 +298,24 @@ 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.
+Compaction works one level at a time. Level zero holds recent flushes. Its
+tables may overlap. Every deeper level holds tables that do not overlap.
+
+A compact moves one level into the level below. It merges the source tables
+with the tables below that overlap them. The merge keeps the newest entry
+for each key. It writes the result to the level below.
 
-Compaction keeps a table file on disk while a snapshot references it. The
-file is deleted only after the last snapshot releases it.
+Level zero compacts when it holds several tables. A deeper level compacts
+when its size passes the level target. The level target grows by a fixed
+ratio at each level.
+
+A tombstone flows down through the levels. It is dropped only when it
+reaches the deepest level. Deeper tables can hold older values. An early
+drop could resurrect a deleted key.
+
+New writes continue into the memory table during the merge. Compaction keeps
+a table file on disk while a snapshot references it. The file is deleted
+only after the last snapshot releases it.
 
 ### Snapshots
 
@@ -341,14 +367,22 @@ config.cache_blocks = 512
 config.sync_writes = true
 config.l0_compact_threshold = 4
 config.compact_on_flush = true
+config.level_target_bytes = 8_i64 * 1024 * 1024
+config.level_ratio = 10
+config.max_levels = 7
 db = Cinderstore::DB.new("data", config)
 ```
 
+`level_target_bytes` sets the size that triggers a level-one compact. Every
+deeper level has a target that is `level_ratio` times larger. `max_levels`
+caps the total number of levels.
+
 ## Project layout
 
 ```text
 src/cinderstore.cr       Library entry point
 src/cinderstore/         Core components
+src/cinderstore/db.cr    Database and leveled compaction
 src/cinderstore/snapshot.cr  Point-in-time snapshot support
 src/cli.cr               Command line tool
 examples/demo.cr         Library walkthrough
@@ -359,11 +393,11 @@ 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 106 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 durability, snapshots, and the server protocol.
+It covers leveled compaction across several levels.
 
 The CI workflow runs on GitHub Actions for Windows and Ubuntu. It checks
 formatting, runs the suite, runs the demo, and builds the binary.
@@ -374,7 +408,7 @@ 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 rewrites whole levels, not single tables.
 - No multi-threaded runtime is required. The server uses fibers.
 - Release snapshots before you close the database.
 
@@ -382,15 +416,17 @@ 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.3: snapshot iterators and consistent reads. Snapshots give a
   stable view of the store. Compaction keeps referenced files alive.
+- Release 0.4: leveled compaction. Compaction moves one level at a time.
+  A level compacts when it passes its size target. Tombstones drop at the
+  deepest level.
 
 ## License
 
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..5225a79 100644
--- a/shard.yml
+++ b/shard.yml
@@ -1,5 +1,5 @@
 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
diff --git a/spec/compaction_spec.cr b/spec/compaction_spec.cr
index f0c3a09..451cf40 100644
--- a/spec/compaction_spec.cr
+++ b/spec/compaction_spec.cr
@@ -103,7 +103,7 @@ describe "Cinderstore compaction" do
     end
   end
 
-  it "handles many small flushes and a full merge" do
+  it "handles many small flushes and compacts them into one table" do
     config = Cinderstore::SpecHelpers.fast_config
     Cinderstore::SpecHelpers.with_db("compact-many", config) do |db, _path|
       6.times do |round|
diff --git a/spec/levels_spec.cr b/spec/levels_spec.cr
new file mode 100644
index 0000000..964b681
--- /dev/null
+++ b/spec/levels_spec.cr
@@ -0,0 +1,172 @@
+require "./spec_helper"
+
+# A config with a tiny level-one target so tests can grow levels fast.
+private def tiny_level_config(level_target_bytes : Int64 = 4_096)
+  config = Cinderstore::SpecHelpers.fast_config
+  config.level_target_bytes = level_target_bytes
+  config
+end
+
+# Writes `count` keys with enough bytes to exceed a 4 KB level target.
+private def put_batch(db : Cinderstore::DB, prefix : String, count : Int32) : Nil
+  count.times do |i|
+    db.put("#{prefix}-%03d" % i, "v#{i}-" + "x" * 40)
+  end
+end
+
+describe "Cinderstore leveled compaction" do
+  it "moves a lone flushed table down one level" do
+    config = Cinderstore::SpecHelpers.fast_config
+    Cinderstore::SpecHelpers.with_db("levels-move", config) do |db, _path|
+      db.put("a", "1")
+      db.put("b", "2")
+      db.flush
+      db.stats.levels.should eq([1])
+      db.compact
+      db.stats.levels.should eq([0, 1])
+      db.get("a").should eq("1")
+      db.get("b").should eq("2")
+    end
+  end
+
+  it "keeps tables in a level disjoint" do
+    config = Cinderstore::SpecHelpers.fast_config
+    Cinderstore::SpecHelpers.with_db("levels-disjoint", config) do |db, _path|
+      db.put("a", "1")
+      db.put("m", "2")
+      db.flush
+      db.compact
+      db.put("n", "3")
+      db.put("z", "4")
+      db.flush
+      db.compact
+      db.stats.levels.should eq([0, 2])
+      db.scan.map(&.[0]).should eq(%w[a m n z])
+    end
+  end
+
+  it "preserves level-one tables that do not overlap a compact" do
+    config = Cinderstore::SpecHelpers.fast_config
+    Cinderstore::SpecHelpers.with_db("levels-preserve", config) do |db, _path|
+      db.put("a", "1")
+      db.put("m", "2")
+      db.flush
+      db.compact
+      db.put("n", "3")
+      db.put("z", "4")
+      db.flush
+      db.compact
+      db.put("a", "updated")
+      db.flush
+      db.compact
+      db.stats.levels.should eq([0, 2])
+      db.get("a").should eq("updated")
+      db.scan.map(&.[1]).should eq(%w[updated 2 3 4])
+    end
+  end
+
+  it "compacts an overflowing level one into level two" do
+    config = tiny_level_config
+    Cinderstore::SpecHelpers.with_db("levels-grow", config) do |db, _path|
+      put_batch(db, "key", 120)
+      db.flush
+      db.compact
+      db.stats.levels.should eq([0, 1])
+      db.compact
+      db.stats.levels.should eq([0, 0, 1])
+      db.get("key-040").should eq("v40-" + "x" * 40)
+    end
+  end
+
+  it "keeps a tombstone while older data lives below" do
+    config = tiny_level_config
+    Cinderstore::SpecHelpers.with_db("levels-keep", config) do |db, _path|
+      put_batch(db, "k", 120)
+      db.flush
+      db.compact
+      db.compact
+      db.stats.levels.should eq([0, 0, 1])
+
+      db.delete("k-040")
+      db.flush
+      put_batch(db, "j", 120)
+      db.flush
+      db.compact
+      db.stats.levels.should eq([0, 1, 1])
+      db.get("k-040").should be_nil
+    end
+  end
+
+  it "drops a tombstone once it reaches the deepest level" do
+    config = tiny_level_config
+    Cinderstore::SpecHelpers.with_db("levels-drop", config) do |db, _path|
+      put_batch(db, "k", 120)
+      db.flush
+      db.compact
+      db.compact
+      db.delete("k-040")
+      db.flush
+      put_batch(db, "j", 120)
+      db.flush
+      db.compact
+      db.stats.levels.should eq([0, 1, 1])
+
+      db.compact
+      db.stats.levels.should eq([0, 0, 1])
+      db.get("k-040").should be_nil
+      db.scan.size.should eq(239)
+    end
+  end
+
+  it "reads the newest value across three levels" do
+    config = tiny_level_config
+    Cinderstore::SpecHelpers.with_db("levels-newest", config) do |db, _path|
+      put_batch(db, "k", 120)
+      db.flush
+      db.compact
+      db.compact
+      db.put("k-040", "newest")
+      db.flush
+      db.compact
+      db.get("k-040").should eq("newest")
+      db.scan.size.should eq(120)
+    end
+  end
+
+  it "recovers a multi-level store after restart" do
+    config = tiny_level_config
+    Cinderstore::SpecHelpers.with_db_path("levels-restart") do |path|
+      db = Cinderstore::DB.new(path, config)
+      put_batch(db, "k", 120)
+      db.flush
+      db.compact
+      db.compact
+      db.put("j-100", "later")
+      db.flush
+      db.compact
+      db.stats.levels.should eq([0, 1, 1])
+      db.close
+
+      reopened = Cinderstore::DB.new(path, config)
+      reopened.get("k-040").should eq("v40-" + "x" * 40)
+      reopened.get("j-100").should eq("later")
+      reopened.scan.size.should eq(121)
+      reopened.close
+    end
+  end
+
+  it "reports per level table counts in stats" do
+    config = tiny_level_config
+    Cinderstore::SpecHelpers.with_db("levels-stats", config) do |db, _path|
+      put_batch(db, "k", 120)
+      db.flush
+      db.compact
+      db.compact
+      stats = db.stats
+      stats.levels.should eq([0, 0, 1])
+      stats.tables.should eq(1)
+      stats.l0.should eq(0)
+      stats.to_json.should contain("levels")
+    end
+  end
+end
diff --git a/src/cinderstore/db.cr b/src/cinderstore/db.cr
index f5707f9..f16cf0b 100644
--- a/src/cinderstore/db.cr
+++ b/src/cinderstore/db.cr
@@ -32,6 +32,12 @@ module Cinderstore
       property l0_compact_threshold : Int32 = 4
       # Start background compaction after a flush when true.
       property compact_on_flush : Bool = true
+      # Size target for level one. A full level compacts into the level below.
+      property level_target_bytes : Int64 = 8_i64 * 1024 * 1024
+      # Size growth factor between consecutive levels.
+      property level_ratio : Int32 = 10
+      # Maximum number of levels, including level zero.
+      property max_levels : Int32 = 7
     end
 
     # A snapshot of database counters for reporting.
@@ -39,6 +45,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,16 +54,17 @@ 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,
           "l0"             => @l0,
           "l1"             => @l1,
+          "levels"         => @levels,
           "entries"        => @entries,
           "disk_bytes"     => @disk_bytes,
           "memtable_bytes" => @memtable_bytes,
@@ -73,6 +81,7 @@ module Cinderstore
 
       def to_s(io : IO) : Nil
         io << "tables: #{@tables} (l0: #{@l0}, l1: #{@l1})\n"
+        io << "levels: #{@levels.join(", ")}\n"
         io << "entries: #{@entries}\n"
         io << "disk bytes: #{@disk_bytes}\n"
         io << "memtable bytes: #{@memtable_bytes}\n"
@@ -243,7 +252,12 @@ module Cinderstore
       trigger_compact
     end
 
-    # Merges all tables into a fresh, non-overlapping level-1 set.
+    # Compacts the level that needs it most.
+    #
+    # A compact moves one level of tables into the level below. It merges
+    # the source tables with the tables below that overlap them. Level zero
+    # compacts first when it holds several tables. Deeper levels compact
+    # when their size passes the level target.
     def compact : Nil
       @flush_lock.synchronize do
         do_compact
@@ -290,6 +304,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,
@@ -485,7 +500,7 @@ module Cinderstore
 
     private def trigger_compact : Nil
       return unless @config.compact_on_flush
-      return if @levels[0].size < @config.l0_compact_threshold
+      return unless needs_compact?
       return if @compacting
       @compacting = true
       spawn do
@@ -501,20 +516,128 @@ module Cinderstore
     # Compaction
     # ------------------------------------------------------------------
 
+    # One planned merge: the tables to read and where to write them.
+    private record CompactionStep, level : Int32, output_level : Int32,
+      sources : Array(TableRef), drop_tombstones : Bool
+
     private def do_compact : Nil
-      tables = nil
+      step = nil
       @lock.synchronize do
         return if @closed
-        tables = @levels.flatten
-        return if tables.size < 2
+        step = pick_compaction_step
+      end
+      return unless step
+      compact_step(step.not_nil!)
+    end
+
+    # Selects the level that needs compaction next.
+    #
+    # Level zero compacts first when it holds enough tables. Otherwise the
+    # shallowest level whose size passes its target compacts. An explicit
+    # compact also moves a small level zero into level one.
+    private def pick_compaction_step : CompactionStep?
+      if @levels[0].size >= @config.l0_compact_threshold
+        return level0_step
+      end
+      if level = next_overflowing_level
+        return level_step(level)
       end
+      return level0_step unless @levels[0].empty?
+      nil
+    end
+
+    private def level0_step : CompactionStep?
+      return nil if @levels[0].empty?
+      sources = @levels[0].dup
+      range_start = sources.map(&.first).min
+      range_finish = sources.map(&.last).max
+      sources.concat(overlapping_tables(1, range_start, range_finish))
+      CompactionStep.new(0, 1, sources, deepest_level <= 1)
+    end
 
-      outputs = merge_tables(tables.not_nil!)
+    private def level_step(level : Int32) : CompactionStep?
+      output = (level + 1) < @config.max_levels ? level + 1 : level
+      sources = @levels[level].dup
+      return nil if sources.empty?
+      range_start = sources.map(&.first).min
+      range_finish = sources.map(&.last).max
+      sources.concat(overlapping_tables(output, range_start, range_finish)) unless output == level
+      CompactionStep.new(level, output, sources, deepest_level <= output)
+    end
+
+    # Returns the shallowest level above zero that passes its size target.
+    private def next_overflowing_level : Int32?
+      (1...@levels.size).each do |level|
+        return level if !@levels[level].empty? && level_bytes(level) >= target_bytes(level)
+      end
+      nil
+    end
+
+    # Returns the index of the deepest level that holds a table, or -1.
+    private def deepest_level : Int32
+      i = @levels.size - 1
+      while i >= 0
+        return i unless @levels[i].empty?
+        i -= 1
+      end
+      -1
+    end
+
+    private def needs_compact? : Bool
+      return true if @levels[0].size >= @config.l0_compact_threshold
+      !next_overflowing_level.nil?
+    end
+
+    private def target_bytes(level : Int32) : Int64
+      @config.level_target_bytes.to_i64 * (@config.level_ratio.to_i64 ** (level - 1))
+    end
+
+    private def level_bytes(level : Int32) : Int64
+      return 0_i64 if level >= @levels.size
+      @levels[level].sum { |ref| file_size(ref.path) }
+    end
+
+    # Returns the tables of a level that overlap `[start, finish]`.
+    #
+    # Tables in a level sort by first key and never overlap. The first
+    # table whose range touches the input range starts the run. Every table
+    # after it is a candidate until a table starts past the finish key.
+    private def overlapping_tables(level : Int32, start : String, finish : String) : Array(TableRef)
+      tables = level < @levels.size ? @levels[level] : [] of TableRef
+      result = [] of TableRef
+      found = false
+      tables.each do |table|
+        if found
+          break if table.first > finish
+          result << table
+        elsif table.last >= start
+          found = true
+          result << table
+        end
+      end
+      result
+    end
+
+    # Executes one planned compaction.
+    private def compact_step(step : CompactionStep) : Nil
+      sources = step.sources
+      if sources.size == 1
+        return if step.level == step.output_level
+        move_table(sources.first, step.output_level)
+        return
+      end
+
+      readers = @lock.synchronize { sources.map { |ref| ref.reader(@block_cache) } }
+      iter = [] of Store::Iter
+      readers.each { |reader| iter << TableIter.new(reader) }
+      outputs = write_merged_tables(MergeIter.new(iter), step.drop_tombstones)
 
       @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
+        sources.each { |ref| remove_ref(ref) }
+        outputs.each { |meta| append_table(step.output_level, meta) }
         save_manifest
-        tables.not_nil!.each do |ref|
+        sources.each do |ref|
           # Mark the table as gone, then drop the database reference. The
           # file stays on disk until every snapshot reference is released.
           ref.orphan
@@ -523,12 +646,47 @@ module Cinderstore
       end
     end
 
-    # Merges every table into fresh tables and drops tombstones.
-    private def merge_tables(tables : Array(TableRef)) : Array(TableMeta)
-      readers = @lock.synchronize { tables.map { |ref| ref.reader(@block_cache) } }
-      sources = [] of Store::Iter
-      readers.each { |reader| sources << TableIter.new(reader) }
-      iter = MergeIter.new(sources)
+    # Moves a lone table into the next level without rewriting it.
+    #
+    # A table moves only when nothing below overlaps it. The move keeps the
+    # level disjoint and avoids an unnecessary disk rewrite.
+    private def move_table(ref : TableRef, level : Int32) : Nil
+      @lock.synchronize do
+        return if @closed
+        remove_ref(ref)
+        grow_levels(level)
+        index = @levels[level].index { |table| table.first > ref.first } || @levels[level].size
+        @levels[level].insert(index, ref)
+        save_manifest
+      end
+    end
+
+    # Removes a table reference from every level.
+    private def remove_ref(ref : TableRef) : Nil
+      @levels.each { |level| level.delete(ref) }
+    end
+
+    # Appends a freshly written table to a level.
+    #
+    # The merge yields keys in ascending order, so the output tables are
+    # sorted and disjoint. Appending preserves that order.
+    private def append_table(level : Int32, meta : TableMeta) : Nil
+      grow_levels(level)
+      @levels[level] << TableRef.new(meta.id, meta.first, meta.last, meta.count, table_path(meta.id))
+    end
+
+    private def grow_levels(level : Int32) : Nil
+      while @levels.size <= level
+        @levels << [] of TableRef
+      end
+    end
+
+    # Merges a stream of entries into fresh tables for the next level.
+    #
+    # The merge keeps the newest entry for each key. A tombstone is written
+    # through when a deeper level still holds data. It is dropped only when
+    # the output level is the deepest, because nothing older can hide below.
+    private def write_merged_tables(iter : MergeIter, drop_tombstones : Bool) : Array(TableMeta)
       outputs = [] of TableMeta
       writer : SstableWriter? = nil
       io : File? = nil
@@ -549,7 +707,7 @@ module Cinderstore
       }
 
       while entry = iter.next?
-        next unless entry.alive
+        next if !entry.alive && drop_tombstones
         if writer.nil?
           current_id = @lock.synchronize do
             id = @next_id
@@ -559,7 +717,7 @@ module Cinderstore
           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
-        writer.not_nil!.add(entry.key, entry.value, entry.seq, true)
+        writer.not_nil!.add(entry.key, entry.value, entry.seq, entry.alive)
         written += 1
         finish_output.call if written >= MAX_TABLE_ENTRIES
       end
@@ -575,7 +733,7 @@ module Cinderstore
       manifest = @manifest.not_nil!
       manifest.seq = @seq
       manifest.next_id = @next_id
-      manifest.levels = @levels.map do |level|
+      manifest.levels = trimmed_levels.map do |level|
         level.map do |ref|
           Manifest::TableInfo.new(ref.id, ref.first, ref.last, ref.count)
         end
@@ -583,6 +741,15 @@ module Cinderstore
       manifest.save(File.join(@path, MANIFEST_NAME))
     end
 
+    # Returns the levels without trailing empty levels.
+    private def trimmed_levels : Array(Array(TableRef))
+      levels = @levels.dup
+      while levels.size > 1 && levels.last.empty?
+        levels.pop
+      end
+      levels
+    end
+
     private def wal_path(id : Int64) : String
       File.join(@path, "#{Util.file_stem(id)}#{WAL_SUFFIX}")
     end
diff --git a/src/cinderstore/demo.cr b/src/cinderstore/demo.cr
index 206635f..4651786 100644
--- a/src/cinderstore/demo.cr
+++ b/src/cinderstore/demo.cr
@@ -25,6 +25,8 @@ module Cinderstore
       config = DB::Config.new
       config.sync_writes = false
       config.compact_on_flush = false
+      config.level_target_bytes = 2_048
+      config.level_ratio = 10
       db = DB.new(path, config)
 
       puts "== Cinderstore #{VERSION} demo =="
@@ -95,9 +97,31 @@ module Cinderstore
       snap.release
       puts ""
 
+      puts "8. Leveled compaction grows deeper levels"
+      puts "   Write 90 products across 3 flushes"
+      3.times do |batch|
+        30.times do |i|
+          index = batch * 30 + i
+          sku = "SKU-01%02d" % index
+          db.put(sku, %({"name":"Part #{index}","price":#{index + 1}.00,"stock":#{index}}))
+        end
+        db.flush
+      end
+      stats = db.stats
+      puts "   tables: #{stats.tables} (l0: #{stats.l0}, l1: #{stats.l1}), entries: #{stats.entries}"
+      db.compact
+      stats = db.stats
+      puts "   compact moved level 0 into level 1"
+      puts "   tables: #{stats.tables} (l0: #{stats.l0}, l1: #{stats.l1}), entries: #{stats.entries}"
+      db.compact
+      stats = db.stats
+      puts "   level 1 passed its target, so compact moved it to level 2"
+      puts "   tables: #{stats.tables} (l0: #{stats.l0}, l1: #{stats.l1}, l2: #{stats.levels[2]}), entries: #{stats.entries}"
+      puts ""
+
       db.close
 
-      puts "8. Reopen the database and verify recovery"
+      puts "9. Reopen the database and verify recovery"
       reopened = DB.new(path, config)
       count = reopened.scan.size
       puts "   rows after restart: #{count}"
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