fix: return 32-bit txids to ensure matching with sync messages (#71)

Fixes #68

Author: magnetised

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- Changed type of returned txid to binary to prevent JSON serialization issues ([#71](https://github.com/electric-sql/phoenix_sync/pull/71))
+
 ## [0.5.0] - 2025-08-13
 
 ### Added

apps/phoenix_sync_example/mix.exs

@@ -50,7 +50,7 @@ defmodule PhoenixSyncExample.MixProject do
       {:dns_cluster, "~> 0.1.1"},
       {:bandit, "~> 1.5"},
       {:phoenix_sync, path: "../.."},
-      {:electric, "~> 1.0.24"}
+      {:electric, "~> 1.1.0"}
     ]
   end
 

apps/phoenix_sync_example/mix.lock

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

apps/txid_match/.formatter.exs

@@ -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/.gitignore

@@ -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/README.md

@@ -0,0 +1,17 @@
+import Config
+
+
+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/config/config.exs

@@ -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/postgres.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/mix/tasks/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