diff --git a/lib/sequin/yaml_loader.ex b/lib/sequin/yaml_loader.ex index 55a105e74..b36b35fab 100644 --- a/lib/sequin/yaml_loader.ex +++ b/lib/sequin/yaml_loader.ex @@ -727,6 +727,8 @@ defmodule Sequin.YamlLoader do ) do with {:ok, source_database} <- fetch_database(databases, source_db_name, :source_database), {:ok, destination_database} <- fetch_database(databases, dest_db_name, :destination_database), + {:ok, destination_database} <- ensure_table_cached(destination_database, dest_schema, dest_table), + {:ok, source_database} <- ensure_table_cached(source_database, source_schema, source_table), {:ok, destination_table} <- fetch_table(destination_database.tables, dest_schema, dest_table, :destination_table), {:ok, source_table_struct} <- @@ -794,6 +796,22 @@ defmodule Sequin.YamlLoader do end end + # The cached `tables` can be stale during config apply (only refreshed on create + # and by a 6-hourly cron), so on a miss refresh once from the live database. + defp ensure_table_cached(%PostgresDatabase{} = database, schema, table_name) do + if Enum.any?(database.tables, &(&1.schema == schema && &1.name == table_name)) do + {:ok, database} + else + case Databases.update_tables(database) do + {:ok, refreshed} -> + {:ok, %{database | tables: refreshed.tables, tables_refreshed_at: refreshed.tables_refreshed_at}} + + {:error, error} -> + {:error, error} + end + end + end + #################### ## HTTP Endpoints ## #################### diff --git a/test/sequin/yaml_loader_test.exs b/test/sequin/yaml_loader_test.exs index a00fa908b..c5fcd0fd8 100644 --- a/test/sequin/yaml_loader_test.exs +++ b/test/sequin/yaml_loader_test.exs @@ -188,6 +188,53 @@ defmodule Sequin.YamlLoaderTest do end end + describe "change_retentions" do + test "refreshes a stale table cache on miss during config apply" do + # First apply creates the database; create_db refreshes its table cache. + assert :ok = YamlLoader.apply_from_yml!(playground_yml()) + + [%PostgresDatabase{} = db] = Repo.all(PostgresDatabase) + + # Simulate a stale cache (as if the tables were created after the last + # refresh): clear the cached table list on the existing database record. + {:ok, _} = db |> PostgresDatabase.changeset(%{tables: []}) |> Repo.update() + + # Applying a change_retention whose tables are absent from the stale cache + # must still succeed: the loader refreshes tables from the live database on a + # miss instead of failing the apply. + assert :ok = + YamlLoader.apply_from_yml!(""" + account: + name: "Playground" + + users: + - email: "admin@sequinstream.com" + password: "sequinpassword!" + + databases: + - name: "test-db" + username: "postgres" + password: "postgres" + hostname: "localhost" + database: "sequin_test" + slot_name: "#{replication_slot()}" + publication_name: "#{@publication}" + + change_retentions: + - name: "test-pipeline" + source_database: "test-db" + source_table_schema: "public" + source_table_name: "Characters" + destination_database: "test-db" + destination_table_schema: "public" + destination_table_name: "sequin_events" + """) + + assert [%WalPipeline{} = pipeline] = Repo.all(WalPipeline) + assert pipeline.name == "test-pipeline" + end + end + describe "databases" do test "creates a database" do assert :ok =