diff --git a/changelog.d/sr-1rgb.md b/changelog.d/sr-1rgb.md new file mode 100644 index 0000000..777d2f6 --- /dev/null +++ b/changelog.d/sr-1rgb.md @@ -0,0 +1,5 @@ +### Fixed + +- `StatifierRouter.Addresses.reap/3` runs on SQLite: its stamp and delete + no longer use Postgres's `= ANY(...)`, which failed every reap of a + SQLite host on 0.8.0 and later, under an integer or a text key. diff --git a/docs/adr/0002-addressing.md b/docs/adr/0002-addressing.md index 5d4ee47..3681bd7 100644 --- a/docs/adr/0002-addressing.md +++ b/docs/adr/0002-addressing.md @@ -1433,3 +1433,24 @@ creates the table, and it is run outside the version walk. The 0.9.0 section of `CHANGELOG.md` names the key, the processor, the front, the two calls and the opt-in table. + +## Note (2026-09-30, sr-1rgb): the address sweep's writes run on SQLite + +A Note, not an amendment: it decides nothing and changes no decision, +amendment or Note above it. The Note of 2026-09-27 accepting the sr-w58a +Amendment says the private `stamp/3` and `delete/2` of +`StatifierRouter.Addresses` bind the ids in `fragment("? = ANY(?)", +a.id, ^ids)`. That form is Postgres's own; on SQLite the statement fails +with "no such function: ANY", so a host on SQLite could not reap its +address table from statifier_router 0.8.0 on. + +Both functions now name their rows in `fragment("? IN (?)", a.id, +splice(^batch))`: an `IN` list with one bound parameter per id, which +Postgres and SQLite both take, written in batches of at most 500 ids a +statement (the private `in_batches/2` and `@ids_per_statement`), under +SQLite's smallest limit on bound parameters. Each id is still bound as +the table handed it over, so the Amendment's rule that the package +never casts an id it binds holds as before, and a reap answers the same +counts on Postgres as it did. The tests +`StatifierRouter.SQLiteReapTest` reap on SQLite under the default +integer key, under a text key of digits, and past one batch. diff --git a/lib/statifier_router/addresses.ex b/lib/statifier_router/addresses.ex index 2f14343..cfc2043 100644 --- a/lib/statifier_router/addresses.ex +++ b/lib/statifier_router/addresses.ex @@ -106,6 +106,7 @@ defmodule StatifierRouter.Addresses do @terminal [:completed, :failed, :cancelled] @default_limit 1_000 + @ids_per_statement 500 @typedoc """ What one reap did: how many rows it stamped `terminal_seen_at` on, how @@ -266,27 +267,40 @@ defmodule StatifierRouter.Addresses do DateTime.compare(DateTime.add(seen_at, horizon_ms, :millisecond), now) != :gt end - defp stamp(_config, [], _now), do: 0 - + # Both writes name their rows in an IN list of one bound parameter per + # id, which Postgres and SQLite both take, each id bound uncast as the + # rest of this module binds them. A list longer than + # @ids_per_statement is written in batches of that many, one statement + # each, which keeps every statement under SQLite's smallest limit on + # bound parameters (999) with room for the stamp's own time. defp stamp(config, ids, now) do - {count, _} = + in_batches(ids, fn batch -> from(a in Config.queryable(config, Address), - where: fragment("? = ANY(?)", a.id, ^ids) and is_nil(a.terminal_seen_at) + where: fragment("? IN (?)", a.id, splice(^batch)) and is_nil(a.terminal_seen_at) ) |> config.repo.update_all(set: [terminal_seen_at: now]) - - count + end) end - defp delete(_config, []), do: 0 - defp delete(config, ids) do - {count, _} = + in_batches(ids, fn batch -> config.repo.delete_all( - from(a in Config.queryable(config, Address), where: fragment("? = ANY(?)", a.id, ^ids)) + from(a in Config.queryable(config, Address), + where: fragment("? IN (?)", a.id, splice(^batch)) + ) ) + end) + end - count + # The rows `write` counts over every batch of `ids`; no statement for + # no ids. + defp in_batches(ids, write) do + ids + |> Enum.chunk_every(@ids_per_statement) + |> Enum.reduce(0, fn batch, total -> + {count, _} = write.(batch) + total + count + end) end defp next(rows, limit) when length(rows) == limit, do: List.last(rows).id diff --git a/test/statifier_router/sqlite_migrations_test.exs b/test/statifier_router/sqlite_migrations_test.exs index 16c180f..82559a7 100644 --- a/test/statifier_router/sqlite_migrations_test.exs +++ b/test/statifier_router/sqlite_migrations_test.exs @@ -1,8 +1,9 @@ defmodule StatifierRouter.SQLiteMigrationsTest do # The version walk on SQLite, through ecto_sqlite3, against a database - # file of each test's own: no Postgres, no SQL sandbox, nothing shared - # with the rest of the suite. - use ExUnit.Case, async: true + # file of each test's own: no Postgres, no SQL sandbox. The one repo + # process runs under the module's name, so every module that starts it + # is in the :sqlite_repo group, whose modules never run at once. + use ExUnit.Case, async: true, group: :sqlite_repo alias Ecto.Adapters.SQL alias Ecto.Migrator diff --git a/test/statifier_router/sqlite_reap_test.exs b/test/statifier_router/sqlite_reap_test.exs new file mode 100644 index 0000000..491a9a5 --- /dev/null +++ b/test/statifier_router/sqlite_reap_test.exs @@ -0,0 +1,234 @@ +defmodule StatifierRouter.SQLiteReapTest do + # StatifierRouter.Addresses.reap/3 on SQLite, through ecto_sqlite3, + # against a database file of each test's own, under the default key and + # under a text key built with the :primary_key option. The execution + # statuses come from StatifierRouter.StatusStore: the repo holds only the + # router's tables. In the :sqlite_repo group with + # StatifierRouter.SQLiteMigrationsTest: both start the one repo process. + use ExUnit.Case, async: true, group: :sqlite_repo + + import Ecto.Query, only: [from: 2] + + alias Ecto.Migrator + alias StatifierPersistence.Storage + alias StatifierRouter.Addresses + alias StatifierRouter.Binding + alias StatifierRouter.Config + alias StatifierRouter.Schema.Address + alias StatifierRouter.SQLiteRepo + + # A host on SQLite under the repo's default key. + defmodule MigrateIntegerKey do + @moduledoc false + use Ecto.Migration + + def up, do: StatifierRouter.Migrations.up() + def down, do: StatifierRouter.Migrations.down() + end + + # A host on SQLite under a text key the database fills in. The ids are + # made of digits alone and lead with a zero, so a reap that cast one to + # an integer would lose the zero and match no row of the text column. + defmodule MigrateTextKey do + @moduledoc false + use Ecto.Migration + + def up do + StatifierRouter.Migrations.up( + primary_key: [ + type: :text, + default: fragment("(printf('0%011d', abs(random()) % 100000000000))") + ] + ) + end + + def down, do: StatifierRouter.Migrations.down() + end + + @version 20_260_930_000_501 + + @now ~U[2026-09-30 12:00:00.000000Z] + @hour 3_600_000 + + setup do + database = + Path.join( + System.tmp_dir!(), + "statifier_router_sqlite_#{System.unique_integer([:positive])}.db" + ) + + start_supervised!({SQLiteRepo, database: database, pool_size: 1}) + + on_exit(fn -> + for suffix <- ["", "-wal", "-shm"], do: File.rm(database <> suffix) + end) + + :ok + end + + # Five parcels, one address row each: two delivered, one still on the + # van, one whose execution the store no longer holds, and one delivered + # and already stamped a day ago. + @statuses %{ + "ex_pcl_6001" => :completed, + "ex_pcl_6002" => :completed, + "ex_pcl_6003" => :active, + "ex_pcl_6005" => :completed + } + + defp config do + {:ok, config} = + Config.new( + repo: SQLiteRepo, + delivery: StatifierRouter.RecordingDelivery, + store: %Storage{adapter: StatifierRouter.StatusStore, opts: @statuses} + ) + + config + end + + defp seed do + rows = + for n <- 6001..6005 do + %{ + scope: "depot_north", + document: "parcel_delivery", + key: "pcl_#{n}", + execution_id: "ex_pcl_#{n}", + inserted_at: @now, + terminal_seen_at: if(n == 6005, do: DateTime.add(@now, -24 * @hour, :millisecond)) + } + end + + {5, _} = SQLiteRepo.insert_all(Address, rows) + :ok + end + + # The reap's answer, or the message of what it raised, so a statement + # the adapter refuses fails the assertion that names it. + defp reap(config, bindings, opts) do + Addresses.reap(config, bindings, opts) + rescue + error in Exqlite.Error -> {:raised, Exception.message(error)} + end + + # key => terminal_seen_at, for every address row left. + defp addresses(config) do + config + |> Config.queryable(Address) + |> then(&from(a in &1, select: {a.key, a.terminal_seen_at})) + |> SQLiteRepo.all() + |> Map.new() + end + + defp ids(config) do + SQLiteRepo.all(from(a in Config.queryable(config, Address), select: a.id)) + end + + # The delivered-scan binding, whose horizon keeps a delivered parcel's + # row for an hour after a reap first sees it delivered. + defp bindings do + {:ok, binding} = + Binding.new(%{ + id: "delivered_scans", + source: "depot_scans", + match: "event.kind == 'delivered'", + key: "event.parcel_id", + document: "parcel_delivery", + event: "delivered", + data: ["parcel_id"], + dedupe: %{by: :message_id, horizon_ms: @hour} + }) + + [binding] + end + + # The same reap, twice: the first stamps the two parcels it is the first + # to see delivered and deletes the orphan and the row stamped a day ago; + # the second, an hour on, deletes the two it stamped. + defp sweep(config) do + assert reap(config, bindings(), now: @now) == + {:ok, %{stamped: 2, deleted: 2, next: nil}} + + assert addresses(config) == %{ + "pcl_6001" => @now, + "pcl_6002" => @now, + "pcl_6003" => nil + } + + an_hour_on = DateTime.add(@now, @hour, :millisecond) + + assert reap(config, bindings(), now: an_hour_on) == + {:ok, %{stamped: 0, deleted: 2, next: nil}} + + assert addresses(config) == %{"pcl_6003" => nil} + end + + describe "reap/3 on SQLite" do + # sabotage: put back stamp/3's and delete/2's `? = ANY(?)` fragment -> + # red, the first reap raised "no such function: ANY". + test "stamps and deletes under the default integer key" do + :ok = Migrator.up(SQLiteRepo, @version, MigrateIntegerKey, log: false) + config = config() + :ok = seed() + + assert Enum.all?(ids(config), &is_integer/1) + sweep(config) + end + + # sabotage: cast every id to an integer before it was bound in stamp/3 + # and delete/2 -> red, the first reap stamped and deleted nothing. + test "stamps and deletes under a text key of digits, bound uncast" do + :ok = Migrator.up(SQLiteRepo, @version, MigrateTextKey, log: false) + config = config() + :ok = seed() + + assert Enum.all?(ids(config), &(is_binary(&1) and &1 =~ ~r/^0\d{11}$/)) + sweep(config) + end + end + + describe "reap/3 on SQLite, past one statement's batch of ids" do + # sabotage: in_batches/2 answered the last batch's count alone -> red, + # the first reap answered stamped: 201 for 1201 rows stamped. + test "stamps and deletes every row of a reap that examines more ids than one statement binds" do + :ok = Migrator.up(SQLiteRepo, @version, MigrateIntegerKey, log: false) + numbers = 7001..8201 + statuses = Map.new(numbers, &{"ex_pcl_#{&1}", :completed}) + + {:ok, config} = + Config.new( + repo: SQLiteRepo, + delivery: StatifierRouter.RecordingDelivery, + store: %Storage{adapter: StatifierRouter.StatusStore, opts: statuses} + ) + + for chunk <- Enum.chunk_every(numbers, 100) do + rows = + for n <- chunk do + %{ + scope: "depot_north", + document: "parcel_delivery", + key: "pcl_#{n}", + execution_id: "ex_pcl_#{n}", + inserted_at: @now + } + end + + SQLiteRepo.insert_all(Address, rows) + end + + assert reap(config, bindings(), now: @now, limit: 2_000) == + {:ok, %{stamped: 1201, deleted: 0, next: nil}} + + assert addresses(config) |> Map.values() |> Enum.uniq() == [@now] + + an_hour_on = DateTime.add(@now, @hour, :millisecond) + + assert reap(config, bindings(), now: an_hour_on, limit: 2_000) == + {:ok, %{stamped: 0, deleted: 1201, next: nil}} + + assert addresses(config) == %{} + end + end +end diff --git a/test/support/status_store.ex b/test/support/status_store.ex new file mode 100644 index 0000000..028e129 --- /dev/null +++ b/test/support/status_store.ex @@ -0,0 +1,23 @@ +defmodule StatifierRouter.StatusStore do + @moduledoc """ + A storage adapter that answers one read, `fetch_execution/2`, from a map + of execution id to status handed to it when the store is built: + + %StatifierPersistence.Storage{adapter: __MODULE__, opts: %{"ex_1" => :completed}} + + An id the map does not name answers `{:error, :execution_not_found}`. + It lets a test reap address rows in a repo that holds no execution + table, the SQLite repo of `StatifierRouter.SQLiteReapTest`. Test-only + support code, not part of the package's public API. + """ + + @doc "The execution's status as the map names it, or `:execution_not_found`." + @spec fetch_execution(%{String.t() => atom()}, String.t()) :: + {:ok, %{status: atom()}} | {:error, :execution_not_found} + def fetch_execution(statuses, execution_id) do + case Map.fetch(statuses, execution_id) do + {:ok, status} -> {:ok, %{status: status}} + :error -> {:error, :execution_not_found} + end + end +end