From 744270d36fed23501d8594aa1d8cbd7eb3ebece7 Mon Sep 17 00:00:00 2001
From: Lucas Polesello <lucas.c4d@gmail.com>
Date: Thu, 30 Jul 2026 17:51:34 -0300
Subject: [PATCH 1/7] Stop recalculating hash if caps is same size as entry

---
 src/lavinmq/clustering/server.cr | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/src/lavinmq/clustering/server.cr b/src/lavinmq/clustering/server.cr
index 87624d4148..475459d6eb 100644
--- a/src/lavinmq/clustering/server.cr
+++ b/src/lavinmq/clustering/server.cr
@@ -198,7 +198,11 @@ module LavinMQ
         sha1 = Digest::SHA1.new
         snapshot.each do |path, mfile|
           # The cache holds full-size hashes; a capped pass must recompute.
-          cached_hash = caps ? nil : @file_index.shared { |_files, checksums| checksums[path]? }
+          cached_hash = @file_index.shared do |_f, checksums|
+            entry = checksums[path]?
+            cap = caps ? (caps[path]? || 0u64) : nil
+            entry if entry && (cap.nil? || entry.size == caps)
+          end
           if cached_hash
             yield({path, cached_hash})
           else

From 9abade3544b1a155d288d654398271743d7bfc50 Mon Sep 17 00:00:00 2001
From: kickster97 <christina@84codes.com>
Date: Fri, 31 Jul 2026 09:36:46 +0200
Subject: [PATCH 2/7] Track covered size in checksums so the capped sync pass
 can reuse them

The capped full_sync pass bypassed the checksum cache entirely and
re-hashed every file from disk while holding @lock, stalling all
replicated operations for the duration. Checksums entries now record
how many bytes each hash covers, so the capped pass reuses a cached
hash exactly when it covers the requested cut, which is every file not
written to between the two passes. Comparing against the file's current
size instead would race local writes that haven't invalidated the cache
yet.

The persisted format gains an optional size field; old checksums.sha1
files still parse, their entries just never match a sized lookup.

Fixes #2162

Co-authored-by: Lucas Polesello <lucas.c4d@gmail.com>
---
 spec/clustering/checksums_spec.cr   | 39 ++++++++++++++++++++
 spec/clustering/server_spec.cr      | 47 ++++++++++++++++++++++++
 src/lavinmq/clustering/checksums.cr | 56 ++++++++++++++++++++++-------
 src/lavinmq/clustering/server.cr    | 20 +++++++----
 4 files changed, 143 insertions(+), 19 deletions(-)

diff --git a/spec/clustering/checksums_spec.cr b/spec/clustering/checksums_spec.cr
index 75c5a36163..da39342e53 100644
--- a/spec/clustering/checksums_spec.cr
+++ b/spec/clustering/checksums_spec.cr
@@ -61,4 +61,43 @@ describe LavinMQ::Clustering::Checksums do
       restored["b"]?.should eq Digest::SHA1.digest("b")
     end
   end
+
+  it "persists and restores the covered size" do
+    with_datadir do |data_dir|
+      hash = Digest::SHA1.digest("hello")
+      written = LavinMQ::Clustering::Checksums.new(data_dir)
+      written.append("q/msgs.0000000001", hash, 5i64)
+
+      restored = LavinMQ::Clustering::Checksums.new(data_dir)
+      restored.restore
+      restored["q/msgs.0000000001"]?.should eq hash
+      restored.hash_for?("q/msgs.0000000001", 5i64).should eq hash
+      restored.hash_for?("q/msgs.0000000001", 4i64).should be_nil
+    end
+  end
+
+  it "keeps covered sizes across a store rewrite" do
+    with_datadir do |data_dir|
+      hash = Digest::SHA1.digest("hello")
+      checksums = LavinMQ::Clustering::Checksums.new(data_dir)
+      checksums.set("q/msgs.0000000001", hash, 5i64)
+      checksums.store
+
+      restored = LavinMQ::Clustering::Checksums.new(data_dir)
+      restored.restore
+      restored.hash_for?("q/msgs.0000000001", 5i64).should eq hash
+    end
+  end
+
+  it "restores sizeless entries and never serves them for a sized lookup" do
+    with_datadir do |data_dir|
+      hash = Digest::SHA1.digest("hello")
+      File.write File.join(data_dir, "checksums.sha1"), "#{hash.hexstring} *q/msgs.0000000001\n"
+
+      checksums = LavinMQ::Clustering::Checksums.new(data_dir)
+      checksums.restore
+      checksums["q/msgs.0000000001"]?.should eq hash
+      checksums.hash_for?("q/msgs.0000000001", 5i64).should be_nil
+    end
+  end
 end
diff --git a/spec/clustering/server_spec.cr b/spec/clustering/server_spec.cr
index 5bdb5bd040..77aae817dc 100644
--- a/spec/clustering/server_spec.cr
+++ b/spec/clustering/server_spec.cr
@@ -74,6 +74,53 @@ describe LavinMQ::Clustering::Server, tags: "etcd" do
         FileUtils.rm_rf LavinMQ::Config.instance.data_dir
       end
     end
+
+    describe "with caps" do
+      it "reuses the cached hash when it covers exactly the cap" do
+        data_dir = LavinMQ::Config.instance.data_dir
+        Dir.mkdir_p(data_dir)
+        server = LavinMQ::Clustering::Server.new(
+          LavinMQ::Config.instance,
+          NullCoordinator.new,
+          0)
+        content = "hello world"
+        path = File.join(data_dir, "capped_cache_test")
+        File.write path, content
+        server.register_file(path)
+        server.files_with_hash { |_path_hash| }
+        # Deleted from disk: the capped pass can only produce the hash from
+        # the cache, proving it doesn't re-read the file.
+        File.delete path
+
+        caps = {"capped_cache_test" => content.bytesize.to_i64}
+        hashes = [] of Bytes
+        server.files_with_hash(caps) { |_path, hash| hashes << hash }
+        hashes.should eq [Digest::SHA1.digest(content)]
+      ensure
+        FileUtils.rm_rf LavinMQ::Config.instance.data_dir
+      end
+
+      it "recomputes when the cap does not match the cached hash's size" do
+        data_dir = LavinMQ::Config.instance.data_dir
+        Dir.mkdir_p(data_dir)
+        server = LavinMQ::Clustering::Server.new(
+          LavinMQ::Config.instance,
+          NullCoordinator.new,
+          0)
+        content = "hello world"
+        path = File.join(data_dir, "capped_recompute_test")
+        File.write path, content
+        server.register_file(path)
+        server.files_with_hash { |_path_hash| }
+
+        caps = {"capped_recompute_test" => 5i64}
+        hashes = [] of Bytes
+        server.files_with_hash(caps) { |_path, hash| hashes << hash }
+        hashes.should eq [Digest::SHA1.digest(content[0, 5])]
+      ensure
+        FileUtils.rm_rf LavinMQ::Config.instance.data_dir
+      end
+    end
   end
 
   describe "#followers" do
diff --git a/src/lavinmq/clustering/checksums.cr b/src/lavinmq/clustering/checksums.cr
index ec8fb3ada7..9ee9d48bdc 100644
--- a/src/lavinmq/clustering/checksums.cr
+++ b/src/lavinmq/clustering/checksums.cr
@@ -2,7 +2,13 @@ module LavinMQ
   module Clustering
     class Checksums
       Log = LavinMQ::Log.for "clustering.checksums"
-      @checksums = Hash(String, Bytes).new
+
+      # `size` is the number of bytes the hash covers, when known. Entries
+      # without a size (older checksums.sha1 files, follower-written entries)
+      # can never be served for a sized lookup (#hash_for?).
+      record Entry, hash : Bytes, size : Int64?
+
+      @checksums = Hash(String, Entry).new
       # Always-open handle to checksums.sha1, kept open across rewrites so
       # #append never has to check/reopen it: #append writes one line at a time
       # and #store adopts the freshly-renamed file's handle here.
@@ -20,8 +26,8 @@ module LavinMQ
         # can keep using it afterwards.
         tmp = "#{checksums_path}.tmp"
         f = File.new(tmp, "w")
-        @checksums.each do |path, hash|
-          f.puts "#{hash.hexstring} *#{path}"
+        @checksums.each do |path, entry|
+          f.puts line(path, entry)
         end
         f.flush
         File.rename(tmp, checksums_path)
@@ -34,9 +40,10 @@ module LavinMQ
       # progress survives a crash mid-sync (see Client#sync_files). No fsync:
       # the page cache survives a process crash and the cache is only an
       # optimization (a stale entry just triggers a re-fetch, never data loss).
-      def append(path : String, hash : Bytes) : Nil
-        @checksums[path] = hash
-        @checksum_file.puts "#{hash.hexstring} *#{path}"
+      def append(path : String, hash : Bytes, size : Int64? = nil) : Nil
+        entry = Entry.new(hash, size)
+        @checksums[path] = entry
+        @checksum_file.puts line(path, entry)
         @checksum_file.flush
       end
 
@@ -44,9 +51,13 @@ module LavinMQ
         File.open(checksums_path) do |f|
           loop do
             hash = f.read_string(40).hexbytes
-            f.skip(2) # " *"
-            path = f.read_line
-            @checksums[path] = hash
+            rest = f.read_line
+            if rest.starts_with?(" *") # sizeless format: "<hash> *<path>"
+              @checksums[rest[2..]] = Entry.new(hash, nil)
+            elsif idx = rest.index(" *", 1) # sized format: "<hash> <size> *<path>"
+              size = rest[1...idx].to_i64?
+              @checksums[rest[idx + 2..]] = Entry.new(hash, size)
+            end
           rescue IO::EOFError
             break
           end
@@ -60,12 +71,23 @@ module LavinMQ
         Log.info { "Checksums not found" }
       end
 
-      def []?(path)
-        @checksums[path]?
+      def []?(path) : Bytes?
+        @checksums[path]?.try &.hash
+      end
+
+      # The hash for `path` only if it covers exactly `size` bytes.
+      def hash_for?(path, size : Int64) : Bytes?
+        if entry = @checksums[path]?
+          entry.hash if entry.size == size
+        end
+      end
+
+      def []=(path, hash : Bytes)
+        @checksums[path] = Entry.new(hash, nil)
       end
 
-      def []=(path, value)
-        @checksums[path] = value
+      def set(path : String, hash : Bytes, size : Int64) : Nil
+        @checksums[path] = Entry.new(hash, size)
       end
 
       def delete(path)
@@ -80,6 +102,14 @@ module LavinMQ
         @checksums.size
       end
 
+      private def line(path : String, entry : Entry) : String
+        if size = entry.size
+          "#{entry.hash.hexstring} #{size} *#{path}"
+        else
+          "#{entry.hash.hexstring} *#{path}"
+        end
+      end
+
       private def checksums_path : String
         File.join(@data_dir, "checksums.sha1")
       end
diff --git a/src/lavinmq/clustering/server.cr b/src/lavinmq/clustering/server.cr
index 475459d6eb..eaa634db4c 100644
--- a/src/lavinmq/clustering/server.cr
+++ b/src/lavinmq/clustering/server.cr
@@ -197,25 +197,33 @@ module LavinMQ
         snapshot = @file_index.shared { |files, _checksums| files.dup }
         sha1 = Digest::SHA1.new
         snapshot.each do |path, mfile|
-          # The cache holds full-size hashes; a capped pass must recompute.
-          cached_hash = @file_index.shared do |_f, checksums|
-            entry = checksums[path]?
-            cap = caps ? (caps[path]? || 0u64) : nil
-            entry if entry && (cap.nil? || entry.size == caps)
+          # Without caps any cached hash is valid: every write invalidates the
+          # entry, so an entry that still exists covers the whole file. With
+          # caps the hash must cover exactly caps[path] bytes. Checking the
+          # cap against the file's current size instead would race local
+          # writes that haven't invalidated the cache yet.
+          cached_hash = @file_index.shared do |_files, checksums|
+            if caps
+              checksums.hash_for?(path, caps[path]? || 0i64)
+            else
+              checksums[path]?
+            end
           end
           if cached_hash
             yield({path, cached_hash})
           else
             filename = File.join(@data_dir, path)
             begin
+              hashed_size = 0i64
               File.open(filename) do |f|
                 size = mfile ? mfile.size : f.size.to_i64
                 size = Math.min(size, caps[path]? || 0i64) if caps
+                hashed_size = size.to_i64
                 sha1.update IO::Sized.new(f, size)
               end
               hash = sha1.final
               sha1.reset
-              @file_index.lock { |_files, checksums| checksums[path] = hash } unless caps
+              @file_index.lock { |_files, checksums| checksums.set(path, hash, hashed_size) } unless caps
               yield({path, hash})
             rescue File::NotFoundError
               next # File disappeared since we took the snapshot, just skip it.

From 72e009b333c7c123407a08724bb43f9057c37e89 Mon Sep 17 00:00:00 2001
From: kickster97 <christina@84codes.com>
Date: Tue, 4 Aug 2026 08:41:30 +0200
Subject: [PATCH 3/7] Serialize checksum entries via Entry#to_io and
 Entry.from_io

Keeps the on-disk line format of a serialized Entry in one place
instead of splitting it between the writer and the parser in restore.
Written bytes are unchanged, so existing checksums.sha1 files stay
readable.
---
 src/lavinmq/clustering/checksums.cr | 46 ++++++++++++++++++-----------
 1 file changed, 28 insertions(+), 18 deletions(-)

diff --git a/src/lavinmq/clustering/checksums.cr b/src/lavinmq/clustering/checksums.cr
index 9ee9d48bdc..0a42068091 100644
--- a/src/lavinmq/clustering/checksums.cr
+++ b/src/lavinmq/clustering/checksums.cr
@@ -6,7 +6,29 @@ module LavinMQ
       # `size` is the number of bytes the hash covers, when known. Entries
       # without a size (older checksums.sha1 files, follower-written entries)
       # can never be served for a sized lookup (#hash_for?).
-      record Entry, hash : Bytes, size : Int64?
+      record Entry, hash : Bytes, size : Int64? do
+        # One line per entry: "<hash> <size> *<path>", or "<hash> *<path>"
+        # when the size is unknown. The path is the caller's hash key, so it
+        # travels alongside the entry rather than being stored in it.
+        def to_io(io : IO, path : String) : Nil
+          io << hash.hexstring
+          if s = size
+            io << ' ' << s
+          end
+          io << " *" << path << '\n'
+        end
+
+        # Returns nil for malformed lines. Raises IO::EOFError at end of file.
+        def self.from_io(io : IO) : {String, Entry}?
+          hash = io.read_string(40).hexbytes
+          rest = io.read_line
+          if rest.starts_with?(" *")
+            {rest[2..], new(hash, nil)}
+          elsif idx = rest.index(" *", 1)
+            {rest[idx + 2..], new(hash, rest[1...idx].to_i64?)}
+          end
+        end
+      end
 
       @checksums = Hash(String, Entry).new
       # Always-open handle to checksums.sha1, kept open across rewrites so
@@ -27,7 +49,7 @@ module LavinMQ
         tmp = "#{checksums_path}.tmp"
         f = File.new(tmp, "w")
         @checksums.each do |path, entry|
-          f.puts line(path, entry)
+          entry.to_io(f, path)
         end
         f.flush
         File.rename(tmp, checksums_path)
@@ -43,20 +65,16 @@ module LavinMQ
       def append(path : String, hash : Bytes, size : Int64? = nil) : Nil
         entry = Entry.new(hash, size)
         @checksums[path] = entry
-        @checksum_file.puts line(path, entry)
+        entry.to_io(@checksum_file, path)
         @checksum_file.flush
       end
 
       def restore : Nil
         File.open(checksums_path) do |f|
           loop do
-            hash = f.read_string(40).hexbytes
-            rest = f.read_line
-            if rest.starts_with?(" *") # sizeless format: "<hash> *<path>"
-              @checksums[rest[2..]] = Entry.new(hash, nil)
-            elsif idx = rest.index(" *", 1) # sized format: "<hash> <size> *<path>"
-              size = rest[1...idx].to_i64?
-              @checksums[rest[idx + 2..]] = Entry.new(hash, size)
+            if parsed = Entry.from_io(f)
+              path, entry = parsed
+              @checksums[path] = entry
             end
           rescue IO::EOFError
             break
@@ -102,14 +120,6 @@ module LavinMQ
         @checksums.size
       end
 
-      private def line(path : String, entry : Entry) : String
-        if size = entry.size
-          "#{entry.hash.hexstring} #{size} *#{path}"
-        else
-          "#{entry.hash.hexstring} *#{path}"
-        end
-      end
-
       private def checksums_path : String
         File.join(@data_dir, "checksums.sha1")
       end

From c1fd25b24eeacf3f746dba60fc5ba8a731859212 Mon Sep 17 00:00:00 2001
From: kickster97 <christina@84codes.com>
Date: Tue, 4 Aug 2026 09:32:51 +0200
Subject: [PATCH 4/7] Revert "Serialize checksum entries via Entry#to_io and
 Entry.from_io"

This reverts commit c2901f087adbdfd3c697c758bc235d3ea5df61b3.
---
 src/lavinmq/clustering/checksums.cr | 46 +++++++++++------------------
 1 file changed, 18 insertions(+), 28 deletions(-)

diff --git a/src/lavinmq/clustering/checksums.cr b/src/lavinmq/clustering/checksums.cr
index 0a42068091..9ee9d48bdc 100644
--- a/src/lavinmq/clustering/checksums.cr
+++ b/src/lavinmq/clustering/checksums.cr
@@ -6,29 +6,7 @@ module LavinMQ
       # `size` is the number of bytes the hash covers, when known. Entries
       # without a size (older checksums.sha1 files, follower-written entries)
       # can never be served for a sized lookup (#hash_for?).
-      record Entry, hash : Bytes, size : Int64? do
-        # One line per entry: "<hash> <size> *<path>", or "<hash> *<path>"
-        # when the size is unknown. The path is the caller's hash key, so it
-        # travels alongside the entry rather than being stored in it.
-        def to_io(io : IO, path : String) : Nil
-          io << hash.hexstring
-          if s = size
-            io << ' ' << s
-          end
-          io << " *" << path << '\n'
-        end
-
-        # Returns nil for malformed lines. Raises IO::EOFError at end of file.
-        def self.from_io(io : IO) : {String, Entry}?
-          hash = io.read_string(40).hexbytes
-          rest = io.read_line
-          if rest.starts_with?(" *")
-            {rest[2..], new(hash, nil)}
-          elsif idx = rest.index(" *", 1)
-            {rest[idx + 2..], new(hash, rest[1...idx].to_i64?)}
-          end
-        end
-      end
+      record Entry, hash : Bytes, size : Int64?
 
       @checksums = Hash(String, Entry).new
       # Always-open handle to checksums.sha1, kept open across rewrites so
@@ -49,7 +27,7 @@ module LavinMQ
         tmp = "#{checksums_path}.tmp"
         f = File.new(tmp, "w")
         @checksums.each do |path, entry|
-          entry.to_io(f, path)
+          f.puts line(path, entry)
         end
         f.flush
         File.rename(tmp, checksums_path)
@@ -65,16 +43,20 @@ module LavinMQ
       def append(path : String, hash : Bytes, size : Int64? = nil) : Nil
         entry = Entry.new(hash, size)
         @checksums[path] = entry
-        entry.to_io(@checksum_file, path)
+        @checksum_file.puts line(path, entry)
         @checksum_file.flush
       end
 
       def restore : Nil
         File.open(checksums_path) do |f|
           loop do
-            if parsed = Entry.from_io(f)
-              path, entry = parsed
-              @checksums[path] = entry
+            hash = f.read_string(40).hexbytes
+            rest = f.read_line
+            if rest.starts_with?(" *") # sizeless format: "<hash> *<path>"
+              @checksums[rest[2..]] = Entry.new(hash, nil)
+            elsif idx = rest.index(" *", 1) # sized format: "<hash> <size> *<path>"
+              size = rest[1...idx].to_i64?
+              @checksums[rest[idx + 2..]] = Entry.new(hash, size)
             end
           rescue IO::EOFError
             break
@@ -120,6 +102,14 @@ module LavinMQ
         @checksums.size
       end
 
+      private def line(path : String, entry : Entry) : String
+        if size = entry.size
+          "#{entry.hash.hexstring} #{size} *#{path}"
+        else
+          "#{entry.hash.hexstring} *#{path}"
+        end
+      end
+
       private def checksums_path : String
         File.join(@data_dir, "checksums.sha1")
       end

From 6daca15bfa525d86548648cebb380b4cec13d9a4 Mon Sep 17 00:00:00 2001
From: kickster97 <christina@84codes.com>
Date: Wed, 5 Aug 2026 16:19:43 +0200
Subject: [PATCH 5/7] Record the covered size for every follower checksum write

The compare loop, file_from_socket and the close-time digest
finalization now store how many bytes their hashes cover, so a node
promoted to leader can reuse them in the capped sync pass instead of
re-hashing every file under the replication lock.
---
 spec/clustering/client_sync_spec.cr | 28 +++++++++++++++-------------
 spec/clustering_spec.cr             |  4 ++--
 src/lavinmq/clustering/client.cr    |  9 +++++----
 3 files changed, 22 insertions(+), 19 deletions(-)

diff --git a/spec/clustering/client_sync_spec.cr b/spec/clustering/client_sync_spec.cr
index ee5d77e780..2362e5ff00 100644
--- a/spec/clustering/client_sync_spec.cr
+++ b/spec/clustering/client_sync_spec.cr
@@ -61,26 +61,28 @@ module ClientSyncSpec
     client.close
   end
 
-  def self.persisted_checksums(data_dir : String) : Hash(String, String)
+  def self.persisted_checksums(data_dir : String) : Hash(String, {String, String})
     path = File.join(data_dir, "checksums.sha1")
-    return Hash(String, String).new unless File.exists?(path)
+    return Hash(String, {String, String}).new unless File.exists?(path)
     File.read_lines(path).to_h do |line|
-      hash, _, filename = line.partition(" *")
-      {filename, hash} # a later line wins, as in Checksums#restore
+      hash, _, rest = line.partition(" ")
+      size, _, filename = rest.partition(" *")
+      {filename, {hash, size}} # a later line wins, as in Checksums#restore
     end
   end
 
-  # Every line in checksums.sha1 must be the hash of what's on disk right now:
-  # one that isn't makes the next sync throw the file away and re-fetch it from
-  # the leader. Returns them for further assertions.
+  # Every line in checksums.sha1 must be the hash and size of what's on disk
+  # right now: one that isn't makes the next sync throw the file away and
+  # re-fetch it from the leader. Returns the hashes for further assertions.
   def self.checksums_matching_disk(data_dir : String) : Hash(String, String)
-    checksums = persisted_checksums(data_dir)
-    checksums.each do |filename, hash|
+    persisted_checksums(data_dir).to_h do |filename, (hash, size)|
       path = File.join(data_dir, filename)
       File.exists?(path).should be_true, "checksum for missing file #{filename}"
-      hash.should eq Digest::SHA1.digest(File.read(path)).hexstring
+      content = File.read(path)
+      hash.should eq Digest::SHA1.digest(content).hexstring
+      size.should eq content.bytesize.to_s
+      {filename, hash}
     end
-    checksums
   end
 
   describe LavinMQ::Clustering::Client do
@@ -629,7 +631,7 @@ module ClientSyncSpec
           checksums_file = File.join(data_dir, "checksums.sha1")
           File.exists?(checksums_file).should be_true
           expected = Digest::SHA1.digest(content).hexstring
-          File.read(checksums_file).should contain "#{expected} *queue1/messages.dat"
+          File.read(checksums_file).should contain "#{expected} #{content.bytesize} *queue1/messages.dat"
         end
       end
 
@@ -660,7 +662,7 @@ module ClientSyncSpec
           checksums_file = File.join(data_dir, "checksums.sha1")
           File.exists?(checksums_file).should be_true
           expected = Digest::SHA1.digest(content).hexstring
-          File.read(checksums_file).should contain "#{expected} *queue1/messages.dat"
+          File.read(checksums_file).should contain "#{expected} #{content.bytesize} *queue1/messages.dat"
         end
       end
 
diff --git a/spec/clustering_spec.cr b/spec/clustering_spec.cr
index 29140fb9f2..7fe63305ed 100644
--- a/spec/clustering_spec.cr
+++ b/spec/clustering_spec.cr
@@ -626,9 +626,9 @@ describe LavinMQ::Clustering::Client, tags: %w[etcd slow] do
         # Should have checksums for multiple files (queue definition + message segments)
         lines.size.should be >= 2
 
-        # Verify each line has correct checksum format: 40 hex chars, space, asterisk, path
+        # Verify each line has correct checksum format: 40 hex chars, covered size, asterisk, path
         lines.each do |line|
-          line.should match(/^[0-9a-f]{40} \*/)
+          line.should match(/^[0-9a-f]{40} \d+ \*/)
         end
 
         # Should have checksum for the queue's message segment file
diff --git a/src/lavinmq/clustering/client.cr b/src/lavinmq/clustering/client.cr
index bdeb2f29cd..c883584247 100644
--- a/src/lavinmq/clustering/client.cr
+++ b/src/lavinmq/clustering/client.cr
@@ -209,8 +209,8 @@ module LavinMQ
       # covers only part of the content) or the file is gone.
       private def adopt_digest(filename : String, sha1 : Digest::SHA1?) : Nil
         return unless sha1
-        return unless File.exists?(File.join(@data_dir, filename))
-        @checksums[filename] = sha1.final
+        return unless info = File.info?(File.join(@data_dir, filename))
+        @checksums.set(filename, sha1.final, info.size)
       end
 
       private def set_socket_opts(socket)
@@ -351,10 +351,11 @@ module LavinMQ
       # survives a crash.
       private def hash_file(filename : String, path : String) : Bytes
         Log.debug { "Calculating checksum for #{filename}" }
+        size = File.size(path)
         sha1 = Digest::SHA1.new
         sha1.file(path)
         hash = sha1.final
-        @checksums.append(filename, hash)
+        @checksums.append(filename, hash, size)
         hash
       end
 
@@ -421,7 +422,7 @@ module LavinMQ
           remaining.zero? || raise IO::EOFError.new
           # Persist immediately too: a file received here is complete and
           # stable, so a crash mid-sync won't force re-hashing it on restart.
-          @checksums.append(filename, sha1.final)
+          @checksums.append(filename, sha1.final, length)
         end
         Log.debug { "Received #{filename}, #{length.humanize_bytes}" }
       end

From f8b4ed90ce1a081f1a3f7723b2492359ae8bfed8 Mon Sep 17 00:00:00 2001
From: kickster97 <christina@84codes.com>
Date: Wed, 5 Aug 2026 16:19:53 +0200
Subject: [PATCH 6/7] Require covered sizes on checksum writes and hand out
 whole entries

append and set now always take the covered size; only entries restored
from an old-format checksums.sha1 lack one. Lookups return the whole
Entry (hash plus covered size) and each caller decides what coverage it
accepts: the capped sync pass demands an exact match, the uncapped pass
takes any recorded coverage and recomputes sizeless entries outside the
lock so they gain a size the capped pass can trust. Without that
healing, every entry on an upgraded or promoted node stays sizeless
indefinitely and the capped pass keeps re-hashing all untouched files
under the replication lock on each follower join.
---
 spec/clustering/checksums_spec.cr   | 33 +++++++++++------------
 spec/clustering/client_sync_spec.cr | 16 ++++++-----
 spec/clustering/server_spec.cr      | 41 +++++++++++++++++++++++++++++
 src/lavinmq/clustering/checksums.cr | 25 ++++++------------
 src/lavinmq/clustering/client.cr    |  2 +-
 src/lavinmq/clustering/server.cr    | 35 ++++++++++++++----------
 6 files changed, 95 insertions(+), 57 deletions(-)

diff --git a/spec/clustering/checksums_spec.cr b/spec/clustering/checksums_spec.cr
index da39342e53..a3d1a71a54 100644
--- a/spec/clustering/checksums_spec.cr
+++ b/spec/clustering/checksums_spec.cr
@@ -6,11 +6,11 @@ describe LavinMQ::Clustering::Checksums do
     with_datadir do |data_dir|
       checksums = LavinMQ::Clustering::Checksums.new(data_dir)
       hash = Digest::SHA1.digest("hello")
-      checksums.append("queue1/msgs.0000000001", hash)
+      checksums.append("queue1/msgs.0000000001", hash, 5i64)
 
       path = File.join(data_dir, "checksums.sha1")
       File.exists?(path).should be_true
-      File.read(path).should eq "#{hash.hexstring} *queue1/msgs.0000000001\n"
+      File.read(path).should eq "#{hash.hexstring} 5 *queue1/msgs.0000000001\n"
     end
   end
 
@@ -18,11 +18,11 @@ describe LavinMQ::Clustering::Checksums do
     with_datadir do |data_dir|
       hash = Digest::SHA1.digest("hello")
       written = LavinMQ::Clustering::Checksums.new(data_dir)
-      written.append("queue1/msgs.0000000001", hash)
+      written.append("queue1/msgs.0000000001", hash, 5i64)
 
       restored = LavinMQ::Clustering::Checksums.new(data_dir)
       restored.restore
-      restored["queue1/msgs.0000000001"]?.should eq hash
+      restored["queue1/msgs.0000000001"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(hash, 5i64)
 
       # one-shot: the on-disk copy is discarded after restore, so a stale hash
       # can't outlive a 2nd crash before a clean store rewrites it.
@@ -36,13 +36,13 @@ describe LavinMQ::Clustering::Checksums do
   it "rewrites a clean snapshot on store" do
     with_datadir do |data_dir|
       checksums = LavinMQ::Clustering::Checksums.new(data_dir)
-      checksums.append("a", Digest::SHA1.digest("a"))
-      checksums.append("b", Digest::SHA1.digest("b"))
+      checksums.append("a", Digest::SHA1.digest("a"), 1i64)
+      checksums.append("b", Digest::SHA1.digest("b"), 1i64)
       checksums.store
 
       lines = File.read(File.join(data_dir, "checksums.sha1")).lines
       lines.size.should eq checksums.size
-      lines.each(&.should(match(/^[0-9a-f]{40} \*/)))
+      lines.each(&.should(match(/^[0-9a-f]{40} \d+ \*/)))
       # No torn temp file left behind by the atomic rename.
       File.exists?(File.join(data_dir, "checksums.sha1.tmp")).should be_false
     end
@@ -51,14 +51,14 @@ describe LavinMQ::Clustering::Checksums do
   it "keeps persisting via append after a store rewrite" do
     with_datadir do |data_dir|
       checksums = LavinMQ::Clustering::Checksums.new(data_dir)
-      checksums.append("a", Digest::SHA1.digest("a"))
+      checksums.append("a", Digest::SHA1.digest("a"), 1i64)
       checksums.store # rewrites and adopts the new handle
-      checksums.append("b", Digest::SHA1.digest("b"))
+      checksums.append("b", Digest::SHA1.digest("b"), 1i64)
 
       restored = LavinMQ::Clustering::Checksums.new(data_dir)
       restored.restore
-      restored["a"]?.should eq Digest::SHA1.digest("a")
-      restored["b"]?.should eq Digest::SHA1.digest("b")
+      restored["a"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(Digest::SHA1.digest("a"), 1i64)
+      restored["b"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(Digest::SHA1.digest("b"), 1i64)
     end
   end
 
@@ -70,9 +70,7 @@ describe LavinMQ::Clustering::Checksums do
 
       restored = LavinMQ::Clustering::Checksums.new(data_dir)
       restored.restore
-      restored["q/msgs.0000000001"]?.should eq hash
-      restored.hash_for?("q/msgs.0000000001", 5i64).should eq hash
-      restored.hash_for?("q/msgs.0000000001", 4i64).should be_nil
+      restored["q/msgs.0000000001"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(hash, 5i64)
     end
   end
 
@@ -85,19 +83,18 @@ describe LavinMQ::Clustering::Checksums do
 
       restored = LavinMQ::Clustering::Checksums.new(data_dir)
       restored.restore
-      restored.hash_for?("q/msgs.0000000001", 5i64).should eq hash
+      restored["q/msgs.0000000001"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(hash, 5i64)
     end
   end
 
-  it "restores sizeless entries and never serves them for a sized lookup" do
+  it "restores old-format entries without a covered size" do
     with_datadir do |data_dir|
       hash = Digest::SHA1.digest("hello")
       File.write File.join(data_dir, "checksums.sha1"), "#{hash.hexstring} *q/msgs.0000000001\n"
 
       checksums = LavinMQ::Clustering::Checksums.new(data_dir)
       checksums.restore
-      checksums["q/msgs.0000000001"]?.should eq hash
-      checksums.hash_for?("q/msgs.0000000001", 5i64).should be_nil
+      checksums["q/msgs.0000000001"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(hash, nil)
     end
   end
 end
diff --git a/spec/clustering/client_sync_spec.cr b/spec/clustering/client_sync_spec.cr
index 2362e5ff00..2a7a260694 100644
--- a/spec/clustering/client_sync_spec.cr
+++ b/spec/clustering/client_sync_spec.cr
@@ -85,6 +85,8 @@ module ClientSyncSpec
     end
   end
 
+  alias Entry = LavinMQ::Clustering::Checksums::Entry
+
   describe LavinMQ::Clustering::Client do
     describe "stream_changes" do
       # Regression: a single large action must be acked incrementally as its
@@ -367,12 +369,12 @@ module ClientSyncSpec
           client = make_client(data_dir)
           client.hash_local_files_public
 
-          client.@checksums["queue1/messages.dat"]?.should eq Digest::SHA1.digest("a")
-          client.@checksums["definitions.amqp"]?.should eq Digest::SHA1.digest("b")
+          client.@checksums["queue1/messages.dat"]?.should eq Entry.new(Digest::SHA1.digest("a"), 1i64)
+          client.@checksums["definitions.amqp"]?.should eq Entry.new(Digest::SHA1.digest("b"), 1i64)
           # Persisted too, so a crash before the sync doesn't waste the work.
           checksums_file = File.read(File.join(data_dir, "checksums.sha1"))
-          checksums_file.should contain "#{Digest::SHA1.digest("a").hexstring} *queue1/messages.dat"
-          checksums_file.should contain "#{Digest::SHA1.digest("b").hexstring} *definitions.amqp"
+          checksums_file.should contain "#{Digest::SHA1.digest("a").hexstring} 1 *queue1/messages.dat"
+          checksums_file.should contain "#{Digest::SHA1.digest("b").hexstring} 1 *definitions.amqp"
         end
       end
 
@@ -394,12 +396,12 @@ module ClientSyncSpec
           File.write File.join(data_dir, "cached.dat"), "original"
           client = make_client(data_dir)
           cached = Digest::SHA1.digest("cached")
-          client.@checksums.append("cached.dat", cached)
+          client.@checksums.append("cached.dat", cached, 8i64)
 
           client.hash_local_files_public
 
           client.files_hashed.should eq 0
-          client.@checksums["cached.dat"]?.should eq cached
+          client.@checksums["cached.dat"]?.should eq Entry.new(cached, 8i64)
         end
       end
 
@@ -415,7 +417,7 @@ module ClientSyncSpec
           client.hash_local_files_public
 
           client.@checksums["unreadable.dat"]?.should be_nil
-          client.@checksums["readable.dat"]?.should eq Digest::SHA1.digest("yep")
+          client.@checksums["readable.dat"]?.should eq Entry.new(Digest::SHA1.digest("yep"), 3i64)
         end
       end
 
diff --git a/spec/clustering/server_spec.cr b/spec/clustering/server_spec.cr
index 77aae817dc..706aac768d 100644
--- a/spec/clustering/server_spec.cr
+++ b/spec/clustering/server_spec.cr
@@ -120,6 +120,47 @@ describe LavinMQ::Clustering::Server, tags: "etcd" do
       ensure
         FileUtils.rm_rf LavinMQ::Config.instance.data_dir
       end
+
+      # Regression: entries restored from an old-format checksums.sha1 have
+      # no size. The uncapped pass must treat them as misses and recompute,
+      # so the capped pass can reuse the healed entry instead of re-hashing
+      # every upgraded file under the lock.
+      it "heals restored sizeless entries in the uncapped pass" do
+        data_dir = LavinMQ::Config.instance.data_dir
+        Dir.mkdir_p(data_dir)
+        content = "hello world"
+        path = File.join(data_dir, "sizeless_heal_test")
+        File.write path, content
+        checksums_path = File.join(data_dir, "checksums.sha1")
+        stale_hash = Digest::SHA1.digest("stale")
+        File.write checksums_path, "#{stale_hash.hexstring} *sizeless_heal_test\n"
+
+        server = LavinMQ::Clustering::Server.new(
+          LavinMQ::Config.instance,
+          NullCoordinator.new,
+          0)
+        server.register_file(path)
+        tcp_server = TCPServer.new("localhost", 0)
+        spawn { server.listen(tcp_server) }
+        # restore (run by listen) truncates the file once it's loaded
+        wait_for { File.size(checksums_path) == 0 }
+
+        # The sizeless entry is a miss: recomputed, not served stale.
+        uncapped = [] of Bytes
+        server.files_with_hash { |_path, hash| uncapped << hash }
+        uncapped.should eq [Digest::SHA1.digest(content)]
+
+        # Deleted from disk: the capped pass can only reuse the healed entry.
+        File.delete path
+        caps = {"sizeless_heal_test" => content.bytesize.to_i64}
+        capped = [] of Bytes
+        server.files_with_hash(caps) { |_path, hash| capped << hash }
+        capped.should eq [Digest::SHA1.digest(content)]
+      ensure
+        server.try &.close
+        tcp_server.try &.close
+        FileUtils.rm_rf LavinMQ::Config.instance.data_dir
+      end
     end
   end
 
diff --git a/src/lavinmq/clustering/checksums.cr b/src/lavinmq/clustering/checksums.cr
index 9ee9d48bdc..d60c50c652 100644
--- a/src/lavinmq/clustering/checksums.cr
+++ b/src/lavinmq/clustering/checksums.cr
@@ -3,9 +3,9 @@ module LavinMQ
     class Checksums
       Log = LavinMQ::Log.for "clustering.checksums"
 
-      # `size` is the number of bytes the hash covers, when known. Entries
-      # without a size (older checksums.sha1 files, follower-written entries)
-      # can never be served for a sized lookup (#hash_for?).
+      # `size` is the number of bytes the hash covers. Only entries restored
+      # from an old-format checksums.sha1 lack it; callers must treat those
+      # as unusable until recomputed.
       record Entry, hash : Bytes, size : Int64?
 
       @checksums = Hash(String, Entry).new
@@ -40,7 +40,7 @@ module LavinMQ
       # progress survives a crash mid-sync (see Client#sync_files). No fsync:
       # the page cache survives a process crash and the cache is only an
       # optimization (a stale entry just triggers a re-fetch, never data loss).
-      def append(path : String, hash : Bytes, size : Int64? = nil) : Nil
+      def append(path : String, hash : Bytes, size : Int64) : Nil
         entry = Entry.new(hash, size)
         @checksums[path] = entry
         @checksum_file.puts line(path, entry)
@@ -71,19 +71,10 @@ module LavinMQ
         Log.info { "Checksums not found" }
       end
 
-      def []?(path) : Bytes?
-        @checksums[path]?.try &.hash
-      end
-
-      # The hash for `path` only if it covers exactly `size` bytes.
-      def hash_for?(path, size : Int64) : Bytes?
-        if entry = @checksums[path]?
-          entry.hash if entry.size == size
-        end
-      end
-
-      def []=(path, hash : Bytes)
-        @checksums[path] = Entry.new(hash, nil)
+      # Hash and size are handed out together; the caller decides whether
+      # the recorded coverage fits its use.
+      def []?(path) : Entry?
+        @checksums[path]?
       end
 
       def set(path : String, hash : Bytes, size : Int64) : Nil
diff --git a/src/lavinmq/clustering/client.cr b/src/lavinmq/clustering/client.cr
index c883584247..78a398d328 100644
--- a/src/lavinmq/clustering/client.cr
+++ b/src/lavinmq/clustering/client.cr
@@ -260,7 +260,7 @@ module LavinMQ
           if File.exists? path
             # Pre-computed by #hash_local_files, except for files that appeared
             # after that pass.
-            unless local_hash = @checksums[filename]?
+            unless local_hash = @checksums[filename]?.try &.hash
               local_hash = hash_file(filename, path)
               Fiber.yield # CPU bound, so allow other fibers to run
             end
diff --git a/src/lavinmq/clustering/server.cr b/src/lavinmq/clustering/server.cr
index eaa634db4c..b8cdf476ac 100644
--- a/src/lavinmq/clustering/server.cr
+++ b/src/lavinmq/clustering/server.cr
@@ -197,19 +197,8 @@ module LavinMQ
         snapshot = @file_index.shared { |files, _checksums| files.dup }
         sha1 = Digest::SHA1.new
         snapshot.each do |path, mfile|
-          # Without caps any cached hash is valid: every write invalidates the
-          # entry, so an entry that still exists covers the whole file. With
-          # caps the hash must cover exactly caps[path] bytes. Checking the
-          # cap against the file's current size instead would race local
-          # writes that haven't invalidated the cache yet.
-          cached_hash = @file_index.shared do |_files, checksums|
-            if caps
-              checksums.hash_for?(path, caps[path]? || 0i64)
-            else
-              checksums[path]?
-            end
-          end
-          if cached_hash
+          cap = caps ? caps[path]? || 0i64 : nil
+          if cached_hash = cached_hash(path, cap)
             yield({path, cached_hash})
           else
             filename = File.join(@data_dir, path)
@@ -217,7 +206,7 @@ module LavinMQ
               hashed_size = 0i64
               File.open(filename) do |f|
                 size = mfile ? mfile.size : f.size.to_i64
-                size = Math.min(size, caps[path]? || 0i64) if caps
+                size = Math.min(size, cap) if cap
                 hashed_size = size.to_i64
                 sha1.update IO::Sized.new(f, size)
               end
@@ -232,6 +221,24 @@ module LavinMQ
         end
       end
 
+      # Reuse a cached hash only when its recorded coverage fits: exactly
+      # `cap` bytes for a capped pass, any known coverage otherwise. Checking
+      # a cap against the file's current size instead would race local writes
+      # that haven't invalidated the cache yet. A sizeless entry (restored
+      # from an old-format file) is never reused, so the uncapped pass
+      # recomputes it and it gains a size a later capped pass can trust.
+      private def cached_hash(path : String, cap : Int64?) : Bytes?
+        @file_index.shared do |_files, checksums|
+          if entry = checksums[path]?
+            if cap
+              entry.hash if entry.size == cap
+            elsif entry.size
+              entry.hash
+            end
+          end
+        end
+      end
+
       # Yields the file (or nil if missing) along with its real data size in
       # bytes. For MFile-backed sparse files the size comes from mfile.size,
       # not from File.size which would be the capacity. `cap` (see

From bfc1f13f60a5d97f9ddcd32f1c19a383ac25eca9 Mon Sep 17 00:00:00 2001
From: kickster97 <christina@84codes.com>
Date: Fri, 7 Aug 2026 10:30:09 +0200
Subject: [PATCH 7/7] Make checksum entry size non-nilable, drop old-format
 entries on restore

---
 spec/clustering/checksums_spec.cr   |  4 ++--
 src/lavinmq/clustering/checksums.cr | 23 +++++++++--------------
 src/lavinmq/clustering/server.cr    | 16 +++++-----------
 3 files changed, 16 insertions(+), 27 deletions(-)

diff --git a/spec/clustering/checksums_spec.cr b/spec/clustering/checksums_spec.cr
index a3d1a71a54..ce66a6a052 100644
--- a/spec/clustering/checksums_spec.cr
+++ b/spec/clustering/checksums_spec.cr
@@ -87,14 +87,14 @@ describe LavinMQ::Clustering::Checksums do
     end
   end
 
-  it "restores old-format entries without a covered size" do
+  it "drops old-format entries without a covered size on restore" do
     with_datadir do |data_dir|
       hash = Digest::SHA1.digest("hello")
       File.write File.join(data_dir, "checksums.sha1"), "#{hash.hexstring} *q/msgs.0000000001\n"
 
       checksums = LavinMQ::Clustering::Checksums.new(data_dir)
       checksums.restore
-      checksums["q/msgs.0000000001"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(hash, nil)
+      checksums["q/msgs.0000000001"]?.should be_nil
     end
   end
 end
diff --git a/src/lavinmq/clustering/checksums.cr b/src/lavinmq/clustering/checksums.cr
index d60c50c652..115fa2cedc 100644
--- a/src/lavinmq/clustering/checksums.cr
+++ b/src/lavinmq/clustering/checksums.cr
@@ -3,10 +3,8 @@ module LavinMQ
     class Checksums
       Log = LavinMQ::Log.for "clustering.checksums"
 
-      # `size` is the number of bytes the hash covers. Only entries restored
-      # from an old-format checksums.sha1 lack it; callers must treat those
-      # as unusable until recomputed.
-      record Entry, hash : Bytes, size : Int64?
+      # `size` is the number of bytes the hash covers.
+      record Entry, hash : Bytes, size : Int64
 
       @checksums = Hash(String, Entry).new
       # Always-open handle to checksums.sha1, kept open across rewrites so
@@ -52,11 +50,12 @@ module LavinMQ
           loop do
             hash = f.read_string(40).hexbytes
             rest = f.read_line
-            if rest.starts_with?(" *") # sizeless format: "<hash> *<path>"
-              @checksums[rest[2..]] = Entry.new(hash, nil)
-            elsif idx = rest.index(" *", 1) # sized format: "<hash> <size> *<path>"
-              size = rest[1...idx].to_i64?
-              @checksums[rest[idx + 2..]] = Entry.new(hash, size)
+            # Line format: "<hash> <size> *<path>". Old-format lines without a
+            # size are dropped; a hash with unknown coverage is unusable.
+            if idx = rest.index(" *", 1)
+              if size = rest[1...idx].to_i64?
+                @checksums[rest[idx + 2..]] = Entry.new(hash, size)
+              end
             end
           rescue IO::EOFError
             break
@@ -94,11 +93,7 @@ module LavinMQ
       end
 
       private def line(path : String, entry : Entry) : String
-        if size = entry.size
-          "#{entry.hash.hexstring} #{size} *#{path}"
-        else
-          "#{entry.hash.hexstring} *#{path}"
-        end
+        "#{entry.hash.hexstring} #{entry.size} *#{path}"
       end
 
       private def checksums_path : String
diff --git a/src/lavinmq/clustering/server.cr b/src/lavinmq/clustering/server.cr
index b8cdf476ac..3f6a421508 100644
--- a/src/lavinmq/clustering/server.cr
+++ b/src/lavinmq/clustering/server.cr
@@ -198,7 +198,7 @@ module LavinMQ
         sha1 = Digest::SHA1.new
         snapshot.each do |path, mfile|
           cap = caps ? caps[path]? || 0i64 : nil
-          if cached_hash = cached_hash(path, cap)
+          if cached_hash = cached_hash?(path, cap)
             yield({path, cached_hash})
           else
             filename = File.join(@data_dir, path)
@@ -222,19 +222,13 @@ module LavinMQ
       end
 
       # Reuse a cached hash only when its recorded coverage fits: exactly
-      # `cap` bytes for a capped pass, any known coverage otherwise. Checking
+      # `cap` bytes for a capped pass, any coverage otherwise. Checking
       # a cap against the file's current size instead would race local writes
-      # that haven't invalidated the cache yet. A sizeless entry (restored
-      # from an old-format file) is never reused, so the uncapped pass
-      # recomputes it and it gains a size a later capped pass can trust.
-      private def cached_hash(path : String, cap : Int64?) : Bytes?
+      # that haven't invalidated the cache yet.
+      private def cached_hash?(path : String, cap : Int64?) : Bytes?
         @file_index.shared do |_files, checksums|
           if entry = checksums[path]?
-            if cap
-              entry.hash if entry.size == cap
-            elsif entry.size
-              entry.hash
-            end
+            entry.hash if cap.nil? || entry.size == cap
           end
         end
       end