fix: address 7 PR review issues for correctness and determinism

- Fix write_backup compression flag (always-on bug)
- Fix put_if (5-tuple CAS) to update secondary indexes
- Fix put_if (6-tuple compat) to update secondary indexes
- Fix delete_if to remove entries from secondary indexes
- Fix V2 ETS fallback to rebuild index tables
- Fix meta_time fallback to use 0 instead of wall-clock time
- Fix put_many error detection pattern to match {key, {:error, _}}
- Add tests for put_if/delete_if index maintenance
- Fix tests to provide system_time in meta for deterministic time

Author: GSMLG-BOT

lib/concord/backup.ex

@@ -33,6 +33,10 @@ defmodule Concord.Backup do
 
   require Logger
 
+  alias Concord.Compression
+  alias Concord.Index
+  alias Concord.Index.Extractor
+
   @backup_extension ".backup"
   @default_backup_dir "./backups"
 
@@ -337,7 +341,8 @@ defmodule Concord.Backup do
     filename = "concord_backup_#{timestamp}#{@backup_extension}"
     backup_path = Path.join(backup_dir, filename)
 
-    encoded = :erlang.term_to_binary(backup_data, [:compressed | if(compress, do: [], else: [])])
+    opts = if compress, do: [:compressed], else: []
+    encoded = :erlang.term_to_binary(backup_data, opts)
 
     case File.write(backup_path, encoded) do
       :ok -> {:ok, backup_path}
@@ -439,6 +444,30 @@ defmodule Concord.Backup do
       end)
     end
 
+    # Rebuild index ETS tables from index definitions and KV data
+    indexes = Map.get(snapshot_data, :indexes, %{})
+    kv_data = Map.get(snapshot_data, :kv_data, [])
+
+    Enum.each(indexes, fn {name, extractor} ->
+      table = Index.index_table_name(name)
+
+      if :ets.whereis(table) == :undefined do
+        :ets.new(table, [:bag, :public, :named_table])
+      else
+        :ets.delete_all_objects(table)
+      end
+
+      Enum.each(kv_data, fn {key, stored} ->
+        value =
+          case stored do
+            %{value: v} -> Compression.decompress(v)
+            _ -> stored
+          end
+
+        Extractor.index_value(table, key, value, extractor)
+      end)
+    end)
+
     :telemetry.execute(
       [:concord, :backup, :restored],
       %{entry_count: metadata.entry_count},

lib/concord/state_machine.ex

@@ -44,7 +44,7 @@ defmodule Concord.StateMachine do
   # Extract deterministic timestamp in seconds from Ra metadata.
   # Ra's system_time is milliseconds set by the leader at proposal time.
   defp meta_time(meta) do
-    ms = Map.get(meta, :system_time, System.system_time(:millisecond))
+    ms = Map.get(meta, :system_time, 0)
     div(ms, 1000)
   end
 
@@ -221,9 +221,19 @@ defmodule Concord.StateMachine do
               if expired?(current_expires_at, now) do
                 {:error, :not_found}
               else
-                if Compression.decompress(current_value) == expected do
+                old_decompressed = Compression.decompress(current_value)
+
+                if old_decompressed == expected do
                   formatted_value = format_value(value, expires_at)
                   :ets.insert(:concord_store, {key, formatted_value})
+
+                  update_indexes_on_put(
+                    data,
+                    key,
+                    old_decompressed,
+                    Compression.decompress(value)
+                  )
+
                   :ok
                 else
                   {:error, :condition_failed}
@@ -262,8 +272,10 @@ defmodule Concord.StateMachine do
 
     result =
       check_conditional_operation(key, expected, condition_fn, now, fn ->
+        old_value = get_decompressed_value(key)
         formatted_value = format_value(value, expires_at)
         :ets.insert(:concord_store, {key, formatted_value})
+        update_indexes_on_put(data, key, old_value, Compression.decompress(value))
         :ok
       end)
 
@@ -284,7 +296,13 @@ defmodule Concord.StateMachine do
 
     result =
       check_conditional_operation(key, expected, condition_fn, now, fn ->
+        old_value = get_decompressed_value(key)
         :ets.delete(:concord_store, key)
+
+        if old_value != nil do
+          remove_from_all_indexes(data, key, old_value)
+        end
+
         :ok
       end)
 
@@ -390,7 +408,7 @@ defmodule Concord.StateMachine do
       :ok ->
         results = execute_put_many_batch(operations, data)
 
-        case Enum.find(results, fn {status, _} -> status == :error end) do
+        case Enum.find(results, fn {_key, result} -> match?({:error, _}, result) end) do
           nil ->
             duration = System.monotonic_time() - start_time
 

test/concord/bulk_operations_test.exs

@@ -73,20 +73,22 @@ defmodule Concord.BulkOperationsTest do
     end
 
     test "apply_command handles get_many", %{state: state} do
-      meta = %{index: 1}
+      now_ms = System.system_time(:millisecond)
+      now_s = div(now_ms, 1000)
+      meta = %{index: 1, system_time: now_ms}
 
       # Setup some data
       :ets.insert(:concord_store, {"key1", %{value: "value1", expires_at: nil}})
 
       :ets.insert(
         :concord_store,
-        {"key2", %{value: "value2", expires_at: System.system_time(:second) + 3600}}
+        {"key2", %{value: "value2", expires_at: now_s + 3600}}
       )
 
       # Expired
       :ets.insert(
         :concord_store,
-        {"key3", %{value: "value3", expires_at: System.system_time(:second) - 1}}
+        {"key3", %{value: "value3", expires_at: now_s - 1}}
       )
 
       {new_state, result, _effects} =
@@ -126,12 +128,13 @@ defmodule Concord.BulkOperationsTest do
     end
 
     test "apply_command handles touch_many", %{state: state} do
-      meta = %{index: 1}
-      current_time = System.system_time(:second)
+      now_ms = System.system_time(:millisecond)
+      now_s = div(now_ms, 1000)
+      meta = %{index: 1, system_time: now_ms}
 
       # Setup some data with existing TTL
-      :ets.insert(:concord_store, {"key1", format_value("value1", current_time + 100)})
-      :ets.insert(:concord_store, {"key2", format_value("value2", current_time + 200)})
+      :ets.insert(:concord_store, {"key1", format_value("value1", now_s + 100)})
+      :ets.insert(:concord_store, {"key2", format_value("value2", now_s + 200)})
       # No TTL
       :ets.insert(:concord_store, {"key3", format_value("value3", nil)})
 
@@ -141,13 +144,11 @@ defmodule Concord.BulkOperationsTest do
       assert result == {:ok, [{"key1", :ok}, {"key2", :ok}]}
       assert new_state == state
 
-      # Verify TTLs were extended - check they're greater than original time
+      # Verify TTLs were extended — now_s + additional_ttl
       [{_, updated1}] = :ets.lookup(:concord_store, "key1")
       [{_, updated2}] = :ets.lookup(:concord_store, "key2")
-      # Extended beyond original
-      assert updated1.expires_at > current_time + 100
-      # Extended beyond original
-      assert updated2.expires_at > current_time + 200
+      assert updated1.expires_at == now_s + 3600
+      assert updated2.expires_at == now_s + 7200
     end
 
     test "apply_command handles touch_many with non-existent keys", %{state: state} do
@@ -275,6 +276,70 @@ defmodule Concord.BulkOperationsTest do
       assert :ets.lookup(table, "second@test.com") == [{"second@test.com", ["user:1"]}]
     end
 
+    test "put_if (5-tuple CAS) updates secondary indexes", %{state: state} do
+      meta = %{index: 1}
+
+      # Create index
+      {state_with_index, :ok, _} =
+        StateMachine.apply_command(meta, {:create_index, "by_email", {:map_get, :email}}, state)
+
+      # Insert initial record
+      {state2, :ok, _} =
+        StateMachine.apply_command(
+          %{index: 2},
+          {:put, "user:1", %{email: "old@test.com"}, nil},
+          state_with_index
+        )
+
+      table = Concord.Index.index_table_name("by_email")
+      assert :ets.lookup(table, "old@test.com") == [{"old@test.com", ["user:1"]}]
+
+      # Use put_if CAS to update — expected matches the current value
+      {_state3, :ok, _} =
+        StateMachine.apply_command(
+          %{index: 3},
+          {:put_if, "user:1", %{email: "new@test.com"}, nil, %{email: "old@test.com"}},
+          state2
+        )
+
+      # Old index entry removed, new one present
+      assert :ets.lookup(table, "old@test.com") == []
+      assert :ets.lookup(table, "new@test.com") == [{"new@test.com", ["user:1"]}]
+    end
+
+    test "delete_if removes entries from secondary indexes", %{state: state} do
+      meta = %{index: 1}
+
+      # Create index
+      {state_with_index, :ok, _} =
+        StateMachine.apply_command(meta, {:create_index, "by_email", {:map_get, :email}}, state)
+
+      # Insert a record
+      {state2, :ok, _} =
+        StateMachine.apply_command(
+          %{index: 2},
+          {:put, "user:1", %{email: "alice@test.com"}, nil},
+          state_with_index
+        )
+
+      table = Concord.Index.index_table_name("by_email")
+      assert :ets.lookup(table, "alice@test.com") == [{"alice@test.com", ["user:1"]}]
+
+      # delete_if with a condition function that matches
+      condition_fn = fn _value -> true end
+
+      {_state3, :ok, _} =
+        StateMachine.apply_command(
+          %{index: 3},
+          {:delete_if, "user:1", nil, condition_fn},
+          state2
+        )
+
+      # Key deleted and index entry removed
+      assert :ets.lookup(:concord_store, "user:1") == []
+      assert :ets.lookup(table, "alice@test.com") == []
+    end
+
     test "query handles get_many", %{state: state} do
       # Setup some data
       :ets.insert(:concord_store, {"key1", %{value: "value1", expires_at: nil}})

test/concord/ttl_test.exs

@@ -224,8 +224,9 @@ defmodule Concord.TTLTest do
     end
 
     test "apply/3 handles cleanup_expired", %{state: state} do
-      meta = %{index: 1}
-      current_time = System.system_time(:second)
+      now_ms = System.system_time(:millisecond)
+      current_time = div(now_ms, 1000)
+      meta = %{index: 1, system_time: now_ms}
 
       # Insert test data
       :ets.insert(