fix(yaml_loader): refresh stale table cache on miss during config apply

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 ##
   ####################

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 =