Refuse a bad --source-schema with PRECONDITION and add migrate --data-schema (S-007-12, S-007-13)

changelog.d/126-source-schema-checks.md

@@ -0,0 +1,3 @@
+### Added
+
+- `cloudfall migrate` takes a repeatable `--data-schema DATABASE=NAME` beside `--data DATABASE=PATH` (and the `migrate` MCP tool a `data_schema` list): the `data:<database>` step copies only the named schemas, as `data copy --source-schema` does, and the plan names them. A `--source-schema` or `--data-schema` name must now match `^[a-z_][a-z0-9_]*$` and be at most 63 characters, and may be given once per copy; a name that doesn't, a repeated name, or a `--data-schema` for a database with no `--data` exits `PRECONDITION` (code `data_copy_schema_invalid` or `migrate_data_schema_without_data`) before any playbook runs, where a malformed name was a usage error and a repeat was silently dropped (#126)

docs/neon-migration-guide.md

@@ -181,7 +181,10 @@ By default the copy takes every non-system schema of the source, and the
 row-count check and the receipt cover all of them. To take only some,
 name each with a repeatable `--source-schema`, such as
 `--source-schema public`; the plan names the schemas it will copy, and the
-receipt lists them under `spec.schemas`, each table with its `schema`.
+receipt lists them under `spec.schemas`, each table with its `schema`. A
+name must match `^[a-z_][a-z0-9_]*$`, be at most 63 characters, and be
+named once; anything else stops the command with `PRECONDITION` before
+any playbook runs.
 
 The row-count check is the verify step and the receipt is the record's
 entry for the migration. The restore is not done because `pg_restore`
@@ -193,7 +196,8 @@ refusal guards, and cannot skip either.
 One database at a time; repeat per declared database. If the application
 is also moving in the same operation, `cloudfall migrate --data
 app=tmp/neon-app-url` inserts this step into the orchestrated plan
-between services and deploys.
+between services and deploys; `--data-schema app=public` (repeatable)
+narrows it to the named schemas, as `--source-schema` does.
 
 Note the one-time egress: Neon meters the dump like any other transfer,
 so a 14 GB dump costs about a dollar — the last transfer bill this

docs/render-migration-guide.md

@@ -152,8 +152,10 @@ database that already has tables in the copied schemas, verifies per-table
 row counts, and writes a `DataMigrationReceipt`. The row-count check is the
 verify step: the migration is not done because the restore exited zero, it
 is done because the counts match. The copy takes every non-system schema of
-the source; `data copy --source-schema NAME` (repeatable) takes only the
-named ones.
+the source; `data copy --source-schema NAME` and `migrate --data-schema
+DATABASE=NAME` (both repeatable) take only the named ones. A name must
+match `^[a-z_][a-z0-9_]*$` and be named once; anything else stops the
+command with `PRECONDITION` before any playbook runs.
 
 ## Step 7: cut over DNS
 

sdk/src/cloudfall/app.py

@@ -129,7 +129,6 @@
     ConnectionAddress,
     Hostname,
     LinuxUser,
-    PostgresSchemaName,
     ReleaseId,
     ResourceId,
     ResourceKind,
@@ -3113,6 +3112,15 @@ class MigrateArgs(EngineArgs):
             "Source URL file for a database, as DATABASE=PATH (repeatable)"
         ),
     )
+    data_schema: tuple[str, ...] = Flag(
+        default=(),
+        description=(
+            "Source schema to copy into a database given --data, as "
+            "DATABASE=NAME (repeatable); NAME matches ^[a-z_][a-z0-9_]*$, at "
+            "most 63 characters, each named once per database; without it "
+            "every non-system schema of the source is copied"
+        ),
+    )
     yes: bool = Flag(
         default=False,
         confirm=True,
@@ -3154,9 +3162,15 @@ def __post_init__(self) -> None:
             ("data", self.data),
         ):
             _pairs(entries, option)
+        _grouped_pairs(self.data_schema, "data-schema")
 
     def options(self, fleet: Fleet) -> MigrateOptions:
-        """Return what this run builds, deploys and migrates."""
+        """Return what this run builds, deploys and migrates.
+
+        A malformed or repeated ``--data-schema`` name raises
+        ``LifecycleError``, which the command answers with ``PRECONDITION``
+        (S-007-12).
+        """
         return MigrateOptions(
             plan_file=fleet.path(self.plan_file),
             builds=_pairs(self.build, "build"),
@@ -3172,6 +3186,12 @@ def options(self, fleet: Fleet) -> MigrateOptions:
                 database: fleet.path(Path(value))
                 for database, value in _pairs(self.data, "data").items()
             },
+            data_schemas={
+                database: DataCopySchemas.from_boundary(names)
+                for database, names in _grouped_pairs(
+                    self.data_schema, "data-schema"
+                ).items()
+            },
             execute=self.yes,
             restart=self.restart,
         )
@@ -3205,6 +3225,18 @@ def _pairs(entries: tuple[str, ...], option: str) -> dict[str, str]:
     return pairs
 
 
+def _grouped_pairs(entries: tuple[str, ...], option: str) -> dict[str, list[str]]:
+    """Group repeatable NAME=VALUE pairs by name, keeping every value given."""
+    grouped: dict[str, list[str]] = {}
+    for entry in entries:
+        key, separator, value = entry.partition("=")
+        if not separator or not key or not value:
+            message = f"--{option} expects NAME=VALUE, got {entry!r}"
+            raise ParseError(message, context={"flag": option})
+        grouped.setdefault(key, []).append(value)
+    return grouped
+
+
 @dataclass(frozen=True, slots=True)
 class Migration:
     """The migration plan's steps and how far the run got."""
@@ -3244,6 +3276,8 @@ def migrate(args: MigrateArgs, ctx: Ctx, fleet: Fleet) -> Migration:
     """Run each pending step, saving progress after every one."""
     try:
         result = execute_migration(args.config(fleet, ctx), args.options(fleet))
+    except LifecycleError as error:
+        raise _lifecycle_failed(error) from error
     except MigrateError as error:
         raise Exit.PRECONDITION(
             error.detail, context={"code": error.code}
@@ -3843,8 +3877,9 @@ class DataMigrateArgs(EngineArgs):
     source_schema: tuple[str, ...] = Flag(
         default=(),
         description=(
-            "Source schema to copy (repeatable); without it every non-system "
-            "schema of the source is copied"
+            "Source schema to copy (repeatable), matching ^[a-z_][a-z0-9_]*$ "
+            "and at most 63 characters, each named once; without it every "
+            "non-system schema of the source is copied"
         ),
     )
     data_migrations: Path = Flag(
@@ -3861,18 +3896,18 @@ class DataMigrateArgs(EngineArgs):
     )
 
     def __post_init__(self) -> None:
-        """Keep relative paths inside the project; check the schema names."""
+        """Keep relative paths inside the project."""
         EngineArgs.__post_init__(self)
         _inside_project(self.data_migrations, "data-migrations")
-        self.source_schemas()
 
     def source_schemas(self) -> DataCopySchemas:
-        """Return the named source schemas, refusing a malformed name."""
-        try:
-            names = tuple(PostgresSchemaName(name) for name in self.source_schema)
-        except ValueError as error:
-            raise ParseError(str(error), context={"flag": "source-schema"}) from error
-        return DataCopySchemas(names)
+        """Return the named source schemas.
+
+        A malformed or repeated name raises ``LifecycleError``, which the
+        command answers with ``PRECONDITION`` before any playbook (S-007-12),
+        not a usage error.
+        """
+        return DataCopySchemas.from_boundary(self.source_schema)
 
 
 class DeployPayload(Payload):
@@ -4066,8 +4101,8 @@ def data_copy(
     """Restore the source database into the service, or show the plan."""
     context = args.context(fleet, ctx)
     source = fleet.path(args.source_url_file)
-    schemas = args.source_schemas()
     try:
+        schemas = args.source_schemas()
         if not args.yes:
             return DataMigratePayload(
                 _plan(

sdk/src/cloudfall/domain.py

@@ -27,14 +27,16 @@
 _BUILD_TOOL_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,63}")
 _BUILD_TOOL_VERSION_PATTERN = re.compile(r"[0-9]{1,9}(?:\.[0-9]{1,9}){0,7}")
 _CONTROL_CHARACTER_PATTERN = re.compile(r"[\x00-\x1f\x7f]")
+# ^[a-z_][a-z0-9_]*$ and at most 63 characters (S-007), read by fullmatch so
+# a trailing newline can't pass as it would with $.
+_POSTGRES_SCHEMA_NAME_PATTERN = re.compile(r"[a-z_][a-z0-9_]{0,62}")
 BUILD_STEP_DEFAULT_TIMEOUT_SECONDS = 900
 BUILD_STEP_MAXIMUM_TIMEOUT_SECONDS = 3600
 ARTIFACT_CLONE_TIMEOUT_SECONDS = 1800
 """Time ``artifact build`` allows for cloning and packaging, besides the steps"""
 _FILE_MODE_PATTERN = re.compile(r"^0[0-7]{3}$")
 _SHA256_PATTERN = re.compile(r"^[a-f0-9]{64}$")
 _MAX_HOSTNAME_LENGTH = 253
-_POSTGRES_NAME_MAX_BYTES = 63
 _MAX_TCP_PORT = 65535
 _MIN_HTTP_STATUS = 100
 _MAX_HTTP_STATUS = 599
@@ -342,26 +344,22 @@ def from_boundary(cls, value: object) -> LinuxUser:
 class PostgresSchemaName:
     """A PostgreSQL schema name a data copy may select.
 
-    The name travels to the engine as an Ansible extra var, which Ansible
-    templates, so braces are refused; PostgreSQL truncates a name past 63
-    bytes, and a control character breaks the row-count lines.
+    Lower-case letters, digits and underscores, not starting with a digit,
+    at most 63 characters (S-007-12). The name travels to the engine as an
+    Ansible extra var, which Ansible templates, so no brace may pass;
+    PostgreSQL truncates a name past 63 bytes, and a quote, a semicolon, or a
+    control character has no place in a name the dump, the checks, and the
+    row-count lines all carry.
     """
 
     value: str
 
     def __post_init__(self) -> None:
-        """Refuse a name PostgreSQL or Ansible would read differently."""
-        if not self.value:
-            message = "schema name must not be empty"
-            raise ValueError(message)
-        if len(self.value.encode()) > _POSTGRES_NAME_MAX_BYTES:
-            message = f"schema name exceeds 63 bytes: {self.value!r}"
-            raise ValueError(message)
-        if _CONTROL_CHARACTER_PATTERN.search(self.value) or any(
-            brace in self.value for brace in "{}"
-        ):
+        """Refuse a name outside ``^[a-z_][a-z0-9_]*$`` or past 63 characters."""
+        if not _POSTGRES_SCHEMA_NAME_PATTERN.fullmatch(self.value):
             message = (
-                f"schema name holds a control character or a brace: {self.value!r}"
+                f"schema name {self.value!r} must match ^[a-z_][a-z0-9_]*$ "
+                "and be at most 63 characters"
             )
             raise ValueError(message)
 

sdk/src/cloudfall/lifecycle.py

@@ -30,6 +30,7 @@
     DnsMode,
     Hostname,
     IpAddress,
+    PostgresSchemaName,
     Sha256Digest,
     TlsMode,
 )
@@ -44,12 +45,11 @@
 from cloudfall.validation import SchemaCatalog, validate_config
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Mapping
+    from collections.abc import Callable, Mapping, Sequence
     from pathlib import Path
 
     from cloudfall.domain import (
         CommandArgv,
-        PostgresSchemaName,
         ReleaseCommandTimeout,
         ReleaseId,
         ResourceId,
@@ -81,6 +81,7 @@
 _ERROR_SERVICE_MISSING = "lifecycle_service_missing"
 _ERROR_DATABASE_MISSING = "lifecycle_database_missing"
 _ERROR_SOURCE_URL_MISSING = "lifecycle_source_url_missing"
+ERROR_DATA_COPY_SCHEMA_INVALID = "data_copy_schema_invalid"
 
 
 class LifecycleError(RuntimeError):
@@ -269,9 +270,28 @@ class DataCopySchemas:
     names: tuple[PostgresSchemaName, ...] = ()
 
     def __post_init__(self) -> None:
-        """Keep the names sorted so plans and receipts compare verbatim."""
-        unique = sorted(set(self.names), key=lambda name: name.value)
-        object.__setattr__(self, "names", tuple(unique))
+        """Refuse a name given twice; keep the names sorted for plans and receipts."""
+        repeated = sorted(
+            {name.value for name in self.names if self.names.count(name) > 1}
+        )
+        if repeated:
+            message = f"schema named more than once: {', '.join(repeated)}"
+            raise ValueError(message)
+        ordered = sorted(self.names, key=lambda name: name.value)
+        object.__setattr__(self, "names", tuple(ordered))
+
+    @classmethod
+    def from_boundary(cls, values: Sequence[str]) -> DataCopySchemas:
+        """Check the names a flag or a tool call gave, refusing a bad one.
+
+        Raises ``LifecycleError`` with code ``data_copy_schema_invalid`` for
+        a name outside ``^[a-z_][a-z0-9_]*$``, past 63 characters, or given
+        twice (S-007-12), before any playbook runs.
+        """
+        try:
+            return cls(tuple(PostgresSchemaName(value) for value in values))
+        except ValueError as error:
+            raise LifecycleError(ERROR_DATA_COPY_SCHEMA_INVALID, str(error)) from error
 
     @property
     def every_schema(self) -> bool:

sdk/src/cloudfall/migrate.py

@@ -28,6 +28,7 @@
 from cloudfall.domain import ReleaseId, ResourceId, TcpPort
 from cloudfall.inventory import PlatformInventory
 from cloudfall.lifecycle import (
+    DataCopySchemas,
     DeployOptions,
     LifecycleError,
     build_release_artifact,
@@ -80,6 +81,7 @@
 _ERROR_AUDIT_DRIFT = "migrate_audit_drift"
 _ERROR_ROUTES_UNHEALTHY = "migrate_routes_unhealthy"
 _ERROR_DATABASE_UNKNOWN = "migrate_database_unknown"
+_ERROR_DATA_SCHEMA_WITHOUT_DATA = "migrate_data_schema_without_data"
 
 
 class MigrateError(RuntimeError):
@@ -139,6 +141,8 @@ class MigrateOptions:
     releases: Mapping[str, str]
     environment_files: Mapping[str, Path]
     data_migrations: Mapping[str, Path] = field(default_factory=dict)
+    data_schemas: Mapping[str, DataCopySchemas] = field(default_factory=dict)
+    """The schemas each ``data:<database>`` step copies; absent copies all."""
     execute: bool = False
     restart: bool = False
 
@@ -235,8 +239,9 @@ def _computed_steps(
     steps.extend(
         PlanStep(
             f"data:{database}",
-            f"migrate external data into database {database} with "
-            "row-count verification",
+            f"migrate {_data_schemas(options, database).describe()} of the "
+            f"external data into database {database} with row-count "
+            "verification",
         )
         for database in sorted(options.data_migrations)
     )
@@ -344,6 +349,19 @@ def _require_known_databases(
             f"service: {', '.join(unknown)}"
         )
         raise MigrateError(_ERROR_DATABASE_UNKNOWN, detail)
+    # A schema for a database nothing is copied into would be silently
+    # ignored; refuse it before any step runs (S-007-12).
+    unpaired = sorted(set(options.data_schemas) - set(options.data_migrations))
+    if unpaired:
+        detail = (
+            "--data-schema names databases with no --data DATABASE=PATH: "
+            f"{', '.join(unpaired)}"
+        )
+        raise MigrateError(_ERROR_DATA_SCHEMA_WITHOUT_DATA, detail)
+
+
+def _data_schemas(options: MigrateOptions, database: str) -> DataCopySchemas:
+    return options.data_schemas.get(database, DataCopySchemas())
 
 
 def _load_plan(
@@ -388,6 +406,11 @@ def _load_plan(
             "with --restart to discard it"
         )
         raise MigrateError(_ERROR_PLAN_STALE, detail)
+    # A pending step runs with this run's options (a data step's schemas
+    # among them), so it is described by them, not by the saved plan's text.
+    for saved, fresh in zip(persisted, computed, strict=True):
+        if saved.status is StepStatus.PENDING:
+            saved.description = fresh.description
     return persisted
 
 
@@ -511,7 +534,11 @@ def _run() -> dict[str, object]:
         runners[f"deploy:{component_id}"] = _deploy_runner(component_id)
     for database, source_url_file in options.data_migrations.items():
         runners[f"data:{database}"] = _data_runner(
-            config, inventory, database, source_url_file
+            config,
+            inventory,
+            database,
+            source_url_file,
+            _data_schemas(options, database),
         )
     return runners
 
@@ -521,6 +548,7 @@ def _data_runner(
     inventory: PlatformInventory,
     database: str,
     source_url_file: Path,
+    schemas: DataCopySchemas,
 ) -> Runner:
     def _run() -> dict[str, object]:
         service = next(
@@ -538,6 +566,7 @@ def _run() -> dict[str, object]:
             database,
             source_url_file,
             config.data_migrations_directory,
+            schemas,
         )
 
     return _run