Reuse cached checksums in the capped full_sync pass

spec/clustering/checksums_spec.cr

@@ -6,23 +6,23 @@ 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
 
   it "restores appended hashes once, then discards them (one-shot)" 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,50 @@ 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
+
+  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 LavinMQ::Clustering::Checksums::Entry.new(hash, 5i64)
+    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["q/msgs.0000000001"]?.should eq LavinMQ::Clustering::Checksums::Entry.new(hash, 5i64)
+    end
+  end
+
+  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 be_nil
     end
   end
 end

spec/clustering/client_sync_spec.cr

@@ -61,28 +61,32 @@ 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
 
+  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
@@ -365,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
 
@@ -392,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
 
@@ -413,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
 
@@ -629,7 +633,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 +664,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
 

spec/clustering/server_spec.cr

@@ -74,6 +74,94 @@ 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
+
+      # 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
 
   describe "#followers" do

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

src/lavinmq/clustering/checksums.cr

@@ -2,7 +2,11 @@ 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.
+      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 +24,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,19 +38,25 @@ 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
+        entry = Entry.new(hash, size)
+        @checksums[path] = entry
+        @checksum_file.puts line(path, entry)
         @checksum_file.flush
       end
 
       def restore : Nil
         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
+            # 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
           end
@@ -60,12 +70,14 @@ module LavinMQ
         Log.info { "Checksums not found" }
       end
 
-      def []?(path)
+      # Hash and size are handed out together; the caller decides whether
+      # the recorded coverage fits its use.
+      def []?(path) : Entry?
         @checksums[path]?
       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 +92,10 @@ module LavinMQ
         @checksums.size
       end
 
+      private def line(path : String, entry : Entry) : String
+        "#{entry.hash.hexstring} #{entry.size} *#{path}"
+      end
+
       private def checksums_path : String
         File.join(@data_dir, "checksums.sha1")
       end