Add app that tests txid matching

Author: magnetised

.formatter.exs

@@ -9,5 +9,5 @@ locals_without_parens = [
   locals_without_parens: locals_without_parens,
   export: [locals_without_parens: locals_without_parens],
   import_deps: [:plug, :phoenix, :ecto],
-  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}", "apps/**/*.{ex,exs}"]
 ]

apps/txid_match/.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

apps/txid_match/.gitignore

@@ -0,0 +1,23 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+txid_match-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

apps/txid_match/README.md

@@ -0,0 +1,23 @@
+# TxidMatch
+
+Forces postgres's txid to wrap around in order to validate transaction id
+matching.
+
+On my machine I can force a wrap around in about 8 hours. Tweak the
+`bumper_count` in `TxidMatch.Writer.Supervisor` to work with your machine.
+
+Only tested/works on Linux.
+
+    # create  a local postgres database with an in-memory data dir
+    mix txid.postgres up
+
+    # test with txid_current() as the txid function
+    # returns 64-bit ids and fails
+    mix txid "txid_current()"
+
+    # test with txid_current() as the txid function
+    # 32-bit ids and works
+    mix txid "pg_current_xact_id()::xid"
+
+    # stop the pg server and release the ramdisk
+    mix txid.postgres down

apps/txid_match/config/config.exs

@@ -0,0 +1,24 @@
+import Config
+
+# connection_opts = [
+#   username: "postgres",
+#   password: "password",
+#   hostname: "localhost",
+#   database: "txid_sync",
+#   port: 55555
+# ]
+
+connection_opts = [
+  username: "garry",
+  password: "password",
+  hostname: "localhost",
+  database: "txid_sync",
+  port: 5432
+]
+
+config :txid_match, Postgrex, connection_opts
+
+config :phoenix_sync,
+  env: config_env(),
+  mode: :embedded,
+  connection_opts: [{:sslmode, :disable} | connection_opts]

apps/txid_match/lib/mix/tasks/postgres.ex

@@ -0,0 +1,65 @@
+defmodule Mix.Tasks.Txid.Postgres do
+  use Mix.Task
+
+  @shortdoc "Manage the in-memory PostgreSQL database for txid_match"
+
+  def run(args) do
+    {_, [action], _} = OptionParser.parse(args, strict: [])
+
+    case action do
+      "up" -> postgres_up()
+      "down" -> postgres_down()
+      _ -> Mix.raise("Unknown action: #{action}. Use 'up' or 'down'.")
+    end
+  end
+
+  defp postgres_up do
+    Mix.shell().info("Starting PostgreSQL in-memory database...")
+    db_config = Application.get_env(:txid_match, TxidMatch.Postgrex, [])
+
+    {:ok, database} = Keyword.fetch(db_config, :database)
+    {:ok, host} = Keyword.fetch(db_config, :hostname)
+    {:ok, port} = Keyword.fetch(db_config, :port)
+    {:ok, username} = Keyword.fetch(db_config, :username)
+    {:ok, password} = Keyword.fetch(db_config, :password)
+
+    System.put_env("PGPASSWORD", password)
+    data_dir = data_dir()
+    # 2 GB
+    disk_size = 2 * 1024 * 1024 * 1024
+
+    File.mkdir_p!(data_dir)
+    user = System.get_env("USER") || raise "$USER not present"
+
+    {_, 0} = System.cmd("sudo", ~w(modprobe brd rd_nr=1 rd_size=#{disk_size}))
+    {_, 0} = System.cmd("sudo", ~w(mkfs.xfs /dev/ram0))
+    {_, 0} = System.cmd("sudo", ~w(mount /dev/ram0 #{data_dir}))
+    {_, 0} = System.cmd("sudo", ~w(chown #{user} #{data_dir}))
+
+    File.write!("#{data_dir}/postgresql.conf", """
+    listen_addresses = '*'
+    wal_level = logical
+    max_replication_slots = 100
+    max_connections = 1000
+    fsync = off
+    """)
+
+    {_, 0} = System.cmd("pg_ctrl", ~w[init -D #{data_dir}])
+    {_, 0} = System.cmd("pg_ctrl", ~w[start -D #{data_dir}])
+  end
+
+  defp postgres_down do
+    Mix.shell().info("Stopping PostgreSQL in-memory database...")
+    data_dir = data_dir() |> dbg
+
+    {_, _} = System.cmd("pg_ctl", ~w[stop -D #{data_dir}])
+    {_, 0} = System.cmd("sudo", ~w(umount #{data_dir}))
+    {_, 0} = System.cmd("sudo", ~w(rmmod brd))
+
+    File.rm_rf!(data_dir)
+  end
+
+  defp data_dir do
+    System.get_env("PGDATA", Path.expand("tmp/pgdata", File.cwd!()))
+  end
+end

apps/txid_match/lib/mix/tasks/txid_match.ex

@@ -0,0 +1,88 @@
+defmodule Mix.Tasks.Txid do
+  use Mix.Task
+
+  @shortdoc "Run the TXID Match Task"
+
+  def run(args) do
+    {_, [function], _} = OptionParser.parse(args, strict: [])
+
+    Mix.shell().info("Running TXID Match Task with function: #{function}")
+
+    Application.put_env(:txid_match, :function, function, persistent: true)
+
+    db_config = Application.get_env(:txid_match, Postgrex, [])
+
+    {:ok, database} = Keyword.fetch(db_config, :database)
+    {:ok, host} = Keyword.fetch(db_config, :hostname)
+    {:ok, port} = Keyword.fetch(db_config, :port)
+    {:ok, username} = Keyword.fetch(db_config, :username)
+    {:ok, password} = Keyword.fetch(db_config, :password)
+
+    System.put_env("PGPASSWORD", password)
+
+    {out, 0} =
+      System.cmd("psql", [
+        "--host",
+        host,
+        "--port",
+        "#{port}",
+        "--username",
+        username,
+        "--tuples-only",
+        "--csv",
+        "--list"
+      ])
+
+    dbs =
+      out
+      |> String.split("\n")
+      |> Enum.map(&String.split(&1, ","))
+      |> Enum.map(&Enum.at(&1, 0))
+
+    if database in dbs do
+      {_, 0} =
+        System.cmd("dropdb", [
+          "--host",
+          host,
+          "--port",
+          "#{port}",
+          "--username",
+          username,
+          "--force",
+          database
+        ])
+
+      Mix.shell().info("Dropped database #{database}")
+    end
+
+    {_, 0} =
+      System.cmd("createdb", [
+        "-T",
+        "template0",
+        "-E",
+        "UTF-8",
+        "--host",
+        host,
+        "--port",
+        "#{port}",
+        "--username",
+        username,
+        database
+      ])
+
+    Mix.shell().info("Created database #{database}")
+
+    Mix.Task.run("app.start", args)
+
+    # # Ensure the application is started
+    # Application.ensure_all_started(:txid_match)
+
+    # Run the main logic of the task
+    IO.puts("Running TXID Match Task with arguments: #{inspect(args)}")
+
+    # Here you can add the logic for your task
+    # For example, you might want to call a function from your application module
+    # TXIDMatch.SomeModule.some_function(args)
+    Process.sleep(:infinity)
+  end
+end

apps/txid_match/lib/txid_match.ex

@@ -0,0 +1,14 @@
+defmodule TxidMatch do
+  require Logger
+
+  def query!(sql, params \\ []) do
+    Logger.debug("Executing SQL: #{sql} with params: #{inspect(params)}")
+    Postgrex.query!(TxidMatch.Postgrex, sql, params)
+  end
+
+  def txid do
+    # mix txid "txid_current()"
+    # mix txid "pg_current_xact_id()::xid"
+    Application.fetch_env!(:txid_match, :function)
+  end
+end

apps/txid_match/lib/txid_match/application.ex

@@ -0,0 +1,29 @@
+defmodule TxidMatch.Application do
+  # See https://hexdocs.pm/elixir/Application.html
+  # for more information on OTP Applications
+  @moduledoc false
+
+  use Application
+
+  @impl true
+  def start(_type, _args) do
+    Logger.put_application_level(:electric, :warning)
+    Logger.put_application_level(:electric_client, :warning)
+
+    children = [
+      TxidMatch.Expectations,
+      {Postgrex,
+       Application.get_env(:txid_match, Postgrex, [])
+       |> Keyword.merge(name: TxidMatch.Postgrex, ssl: false, pool_size: 500)},
+      TxidMatch.Migrator,
+      TxidMatch.Reader,
+      {TxidMatch.Writer.Supervisor,
+       Application.get_env(:txid_match, TxidMatch.Writer.Supervisor, [])}
+    ]
+
+    # See https://hexdocs.pm/elixir/Supervisor.html
+    # for other strategies and supported options
+    opts = [strategy: :one_for_one, name: TxidMatch.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
+end

apps/txid_match/lib/txid_match/bumper.ex

@@ -0,0 +1,33 @@
+defmodule TxidMatch.Bumper do
+  def child_spec(id) when is_integer(id) do
+    %{
+      id: {__MODULE__, id},
+      start: {Task, :start_link, [__MODULE__, :run, [id]]},
+      type: :worker,
+      restart: :permanent
+    }
+  end
+
+  def run(id) do
+    DBConnection.run(
+      TxidMatch.Postgrex,
+      fn conn ->
+        query =
+          Postgrex.prepare!(
+            conn,
+            "bump",
+            "SELECT pg_current_xact_id()"
+          )
+
+        loop(conn, query, id)
+      end,
+      timeout: :infinity
+    )
+  end
+
+  defp loop(conn, query, id) do
+    %{rows: [[_txid]]} = DBConnection.execute!(conn, query, [])
+
+    loop(conn, query, id)
+  end
+end