Gate restore on the dump's data revision, not the code head

A restore could load a database dump whose schema was older than the running
code. The manifest stamped schemarev from the code image on disk, so the
compatibility check compared the code against itself and always passed. A set
captured from a database at 7d37_import_batches_and_journal restored under
code at 7d41_drop_guest_tools_secrets, leaving the data behind the code: every
query on a table the newer migrations touched failed with 1054 Unknown column,
which surfaced as empty configuration-policy dropdowns and a failure to load
data on edit PC.

The authoritative signal is the alembic_version of the database that was
dumped, so manifests now carry datarev alongside schemarev, backup.sh reads it
from the database and warns when it has drifted from the image, and the gate
asks whether the running code knows that revision. Data behind the code is
allowed because it can be rolled forward; data ahead of the code is refused
because it cannot.

Rolling it forward has to happen somewhere with an app context. The backup
sidecar is mysql plus restic plus alembic only and has no Flask, so restore.sh
loads the dump and parks the run in an upgrading phase behind an
.upgrade-requested sentinel rather than declaring a success it cannot verify.
The api container owns the Flask CLI and already runs these exact idempotent
commands on every deploy, so a finalize endpoint there runs db upgrade plus
plugin upgrade-all and stamps the terminal phase. The UI drives it on seeing
the upgrading phase, so recovery needs no operator intervention.

The live simulation runs one codebase against itself, so its dump is always at
the same revision as the code restoring it and it can never reproduce this
drift. The gate therefore gets a standalone tester that builds throwaway
migration chains and dumps offline, invoked as a simulation preflight. It
caught a real bug in the legacy dump-parse fallback, where the revision was
matched loosely enough to return the alembic_version table name instead of the
value it holds.

shopdb/core/api/shopdbbackups.py

@@ -129,6 +129,58 @@ def _write_status(kind, phase, setname, message):
         pass
 
 
+def _append_history(kind, outcome, setname, message):
+    """Append one run record to the shared history the UI reads, newest last.
+
+    Mirrors status.sh's history_append so a finalize that completes in the api
+    (not the sidecar) still lands in the same history file the restore page
+    shows. Best-effort: history is a convenience log, never a gate.
+    """
+    root = _backups_root()
+    stamp = datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + 'Z'
+    entry = {
+        'kind': kind,
+        'outcome': outcome,
+        'setname': setname,
+        'message': message,
+        'at': stamp,
+    }
+    path = os.path.join(root, 'history', kind + 's.json')
+    try:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        entries = _read_json(path)
+        if not isinstance(entries, list):
+            entries = []
+        entries.append(entry)
+        with open(path, 'w', encoding='utf-8') as handle:
+            json.dump(entries, handle)
+    except OSError:
+        pass
+
+
+def _run_schema_upgrade():
+    """Bring the just-restored database up to the running code, in-process.
+
+    Runs the same two idempotent steps every deploy runs: the core Alembic
+    upgrade to head, then each plugin chain via the plugin manager. Both are
+    no-ops when the data already matches the code head, so this is safe to call
+    even when the restored dump was already current. Uses the live app context
+    and its configured engine, so no subprocess or extra Flask CLI is needed.
+    """
+    from flask_migrate import upgrade as alembic_upgrade
+
+    alembic_upgrade()
+
+    manager = current_app.extensions.get('plugin_manager')
+    if manager is None:
+        raise RuntimeError('Plugin manager not initialized; cannot upgrade plugins')
+    results = manager.upgrade_all_plugins()
+    failed = {name: status for name, status in (results or {}).items()
+              if status not in ('ok', 'no-migrations')}
+    if failed:
+        raise RuntimeError('Plugin upgrade failed: ' + json.dumps(failed))
+
+
 
 def _read_manifest(setdir):
     """Load a set's manifest.json, or None when absent/unreadable.
@@ -175,14 +227,59 @@ def _current_schema_revisions():
         return {}
 
 
+def _known_core_revisions():
+    """Return every core Alembic revision this build knows, heads and ancestors.
+
+    Restore compatibility is now data-driven: a set restores when the running
+    code KNOWS the dump's data revision (so `flask db upgrade` can roll it
+    forward to head). Heads alone cannot answer that - a dump stamped at an
+    older-but-valid revision is perfectly restorable - so we walk the whole
+    script history. Lazy import + broad except keeps a missing/broken alembic
+    from 500ing the list endpoint (returns empty = everything reads unknown).
+    """
+    try:
+        from alembic.config import Config
+        from alembic.script import ScriptDirectory
+    except ImportError:
+        return set()
+
+    root = current_app.root_path
+    migrations = os.path.join(os.path.dirname(root), 'migrations')
+    ini = os.path.join(migrations, 'alembic.ini')
+    if not os.path.isfile(ini):
+        return set()
+    try:
+        config = Config(ini)
+        config.set_main_option('script_location', migrations)
+        script = ScriptDirectory.from_config(config)
+        return {rev.revision for rev in script.walk_revisions()}
+    except Exception:
+        return set()
+
+
 def _is_compatible(manifest, current):
-    """A set restores onto THIS build only when the build's schema is at least
-    the set's. We compare head-revision sets: if the set was taken on a schema
-    the current code has not reached, the code is too OLD and would 500 on the
-    restored rows, so restore is blocked until the code is upgraded.
+    """Decide whether a set restores onto THIS build.
+
+    Data-driven gate (the fix for the FBJXBDB4 outage): the authoritative
+    signal is the set's `datarev` - the actual `alembic_version` of the dumped
+    DATABASE, not the code head that captured it. A restore is safe when the
+    running code KNOWS that revision (it is somewhere in this build's migration
+    history), because then the post-restore finalize can run `flask db upgrade`
+    to roll the loaded data forward to head. If the code does not know the
+    revision, the dump is from a NEWER schema than this build and would 500 on
+    restored rows, so we block until the code is upgraded.
+
+    Legacy sets predate `datarev`; fall back to the old head-subset check on
+    `schemarev` so they stay listable and restorable as before.
     """
     if not manifest:
         return False
+    datarev = manifest.get('datarev')
+    if datarev:
+        known = _known_core_revisions()
+        if not known:
+            return False
+        return datarev in known
     setrevs = set(manifest.get('schemarev', {}).get('core', []))
     currevs = set(current.get('core', []))
     if not setrevs or not currevs:
@@ -338,6 +435,57 @@ def restore_backup(snapshotid):
         message='Restore queued and validated against the live schema.')
 
 
+@shopdbbackups_bp.route('/finalize-restore', methods=['POST'])
+@require_role('admin')
+def finalize_restore():
+    """Run the post-restore schema upgrade in the app context.
+
+    The restore itself runs in the Flask-less backup sidecar, which loads the
+    dump and then drops an .upgrade-requested sentinel because it cannot run
+    'flask db upgrade' itself. This endpoint is the api-side half: it brings the
+    just-restored DATA up to the running CODE with the exact idempotent commands
+    the deploy path uses, then stamps the final 'done'. Without it a restored
+    dump that sits at an older revision leaves every newer-migration table
+    throwing 1054 Unknown column.
+
+    Idempotent and safe to call when nothing is pending: if there is no
+    .upgrade-requested sentinel it reports there was nothing to finalize.
+    """
+    root = _backups_root()
+    sentinel = os.path.join(root, '.upgrade-requested')
+    try:
+        with open(sentinel, 'r', encoding='utf-8') as handle:
+            setname = handle.read().strip()
+    except OSError:
+        return success_response(
+            data={'finalized': False},
+            message='No restore is awaiting schema upgrade.')
+
+    _write_status('restore', 'upgrading', setname,
+                  'Upgrading restored schema to code')
+    try:
+        _run_schema_upgrade()
+    except Exception as error:  # noqa: BLE001 - report any upgrade failure
+        _write_status('restore', 'failed', setname,
+                      'Schema upgrade failed: ' + str(error))
+        _append_history('restore', 'failed', setname,
+                        'Schema upgrade failed: ' + str(error))
+        return error_response(ErrorCodes.INTERNAL_ERROR,
+                              'Post-restore schema upgrade failed: ' + str(error),
+                              http_code=500)
+
+    try:
+        os.remove(sentinel)
+    except OSError:
+        pass
+    _write_status('restore', 'done', setname, 'Restore complete')
+    _append_history('restore', 'success', setname, 'Restore complete')
+    AuditLog.log('restore-finalized', 'Backup', entityname=setname)
+    return success_response(
+        data={'finalized': True, 'snapshotid': setname},
+        message='Restore finalized: schema upgraded to the running code.')
+
+
 @shopdbbackups_bp.route('/status', methods=['GET'])
 @require_role('admin')
 def get_status():