diff --git a/CHANGELOG.md b/CHANGELOG.md index 4cd01346b..335a51741 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -87,6 +87,13 @@ and this project adheres to `503` instead of denying access like a permission failure, so clients retry instead of giving up. The built-in endpoints can also answer JSON on `Accept: application/json` +- ✨(backend) add a `migrate_documents` command replaying the legacy content of + the documents into the collaboration server, one call to its migrate endpoint + per document. Resumable and safe to re-run: what became of every document is + recorded (`impress_document_migration`), a server that is unwell is retried + with a backoff and a document it refuses is left for a later run + (`--retry-failed`). Bounded by `--concurrency`, `--rate` and `--limit`, most + recently edited documents first - ✨(collaboration) notify the backend when the worker persists new content for a document, so the lists ordered by `updated_at` follow the edits made on the collaboration server. The backend serves it on diff --git a/src/backend/core/management/commands/migrate_documents.py b/src/backend/core/management/commands/migrate_documents.py new file mode 100644 index 000000000..0e7d965bf --- /dev/null +++ b/src/backend/core/management/commands/migrate_documents.py @@ -0,0 +1,287 @@ +""" +Replay the legacy content of the documents into the collaboration server. + +The content of a document used to be a file in our object storage, one version +per save; it now lives in the collaboration server, which is able to read those +versions back and rebuild the history from them, document by document. This +command is what hands it the corpus. + +It is meant to be run again: every document it finishes is recorded, and the +collaboration server answers "already" for anything it has already migrated, so +a run that is interrupted, rate limited or killed simply picks up where it +stopped. Nothing is destroyed, on either side. +""" + +import logging +import time +from concurrent.futures import ThreadPoolExecutor + +from django.core.management.base import BaseCommand +from django.utils import timezone + +from core import models +from core.services.yhub_services import APIError, YHubError, YHubService + +logger = logging.getLogger("impress.commands.migrate_documents") + +# Reported often enough to see a run is alive, rarely enough to keep the logs +# of a corpus of hundreds of thousands of documents readable. +PROGRESS_EVERY = 500 + +# Results are written in batches: a smaller one costs a query per document, a +# larger one loses more work when the command is killed — and losing it only +# means handing those documents over again, which answers "already". +WRITE_BATCH = 200 + +# The collaboration server says whether it is worth insisting: a 5xx or a 429 +# is a server that is unwell or busy, anything else in the 4xx range is about +# this document and will fail the same way forever. +RETRY_STATUSES = frozenset({429}) + + +class Command(BaseCommand): + """Migrate the legacy content of the documents into the collaboration server.""" + + help = __doc__ + + def add_arguments(self, parser): + """Define the arguments of the command.""" + parser.add_argument( + "--concurrency", + type=int, + default=2, + help=( + "Documents migrated at the same time. The replay runs on the " + "main thread of the collaboration server, so this is bounded " + "by its cpu: raise it against a pool that serves nothing else." + ), + ) + parser.add_argument( + "--rate", + type=float, + default=0, + help="Documents per second not to exceed (0: as fast as possible)", + ) + parser.add_argument( + "--limit", + type=int, + default=0, + help="Stop after this many documents (0: all of them)", + ) + parser.add_argument( + "--created-before", + type=str, + default=None, + help=( + "Only migrate the documents created before this date " + "(ISO 8601). Documents created after the collaboration server " + "became the source of truth have no legacy content." + ), + ) + parser.add_argument( + "--retry-failed", + action="store_true", + default=False, + help="Hand over the documents a previous run could not migrate", + ) + parser.add_argument( + "--retries", + type=int, + default=3, + help="Attempts per document when the collaboration server is unwell", + ) + parser.add_argument( + "--dry-run", + action="store_true", + default=False, + help="Count what would be migrated, call nothing", + ) + + def handle(self, *args, **options): + """Hand the documents to the collaboration server, and record what it says.""" + queryset = self.get_queryset(options) + total = queryset.count() + + if options["dry_run"]: + self.stdout.write(f"{total:d} documents to migrate") + return + + self.stdout.write( + f"Migrating {total:d} documents, {options['concurrency']:d} at a time" + ) + started = time.monotonic() + counts = self.migrate(queryset, options) + elapsed = time.monotonic() - started + + done = sum(counts.values()) + rate = done / elapsed if elapsed else 0 + self.stdout.write( + f"Migrated {done:d} documents in {elapsed:.0f}s ({rate:.1f}/s): " + + ", ".join( + f"{status}={count:d}" for status, count in sorted(counts.items()) + ) + ) + if counts.get(models.DocumentMigrationStatus.FAILED): + self.stdout.write( + self.style.WARNING( + "Some documents could not be migrated, they are recorded as " + "failed: run again with --retry-failed once the cause is fixed." + ) + ) + + def get_queryset(self, options): + """ + Return the documents left to migrate, the ones that matter most first. + + A document is opened before it is missed: the ones edited recently are + the ones users are about to read, and until a document is migrated the + collaboration server only seeds its latest state, without its history. + """ + queryset = models.Document.objects.all() + + if options["created_before"]: + queryset = queryset.filter(created_at__lt=options["created_before"]) + + done = set(models.DocumentMigrationStatus.values) + if options["retry_failed"]: + done.discard(models.DocumentMigrationStatus.FAILED) + + queryset = queryset.exclude(migration__status__in=done) + + if options["limit"]: + queryset = queryset.order_by("-updated_at")[: options["limit"]] + # a sliced queryset cannot be iterated with a server-side cursor + return models.Document.objects.filter( + pk__in=queryset.values("pk") + ).order_by("-updated_at") + + return queryset.order_by("-updated_at") + + def migrate(self, queryset, options): + """Run the migration, writing what happened as the answers come in.""" + counts = {} + results = [] + done = 0 + started = time.monotonic() + + with ThreadPoolExecutor(max_workers=options["concurrency"]) as pool: + # imap-like: the documents are read from the database as the pool + # frees up, so a corpus of any size is never held in memory + documents = queryset.only("pk").iterator(chunk_size=WRITE_BATCH) + migrations = pool.map( + lambda document: self.migrate_document(document, options["retries"]), + self.paced(documents, options["rate"]), + ) + + for migration in migrations: + results.append(migration) + counts[migration.status] = counts.get(migration.status, 0) + 1 + done += 1 + + if len(results) >= WRITE_BATCH: + self.save(results) + results = [] + if done % PROGRESS_EVERY == 0: + rate = done / (time.monotonic() - started) + self.stdout.write(f" {done:d} documents ({rate:.1f}/s)") + + self.save(results) + + return counts + + @staticmethod + def paced(documents, rate): + """Yield the documents no faster than `rate` per second.""" + if not rate: + yield from documents + return + + interval = 1 / rate + next_at = time.monotonic() + for document in documents: + now = time.monotonic() + if next_at > now: + time.sleep(next_at - now) + next_at = max(next_at + interval, now) + yield document + + def migrate_document(self, document, retries): + """ + Hand one document over, and return what the collaboration server said. + + Runs in a worker thread and touches no database: the results are + written by the main thread, so the pool needs no connection of its own. + """ + service = YHubService() + + for attempt in range(1, retries + 1): + try: + result = service.migrate(document) + except YHubError as err: + if attempt < retries and self.is_retryable(err): + # the server is unwell or busy, not this document + time.sleep(2**attempt) + continue + + logger.warning("document %s was not migrated: %s", document.pk, err) + return models.DocumentMigration( + document_id=document.pk, + status=models.DocumentMigrationStatus.FAILED, + error=str(err)[:500], + updated_at=timezone.now(), + ) + + return models.DocumentMigration( + document_id=document.pk, + status=result.get("status", models.DocumentMigrationStatus.MIGRATED), + versions=result.get("versions", 0), + applied=result.get("applied", 0), + skipped=result.get("skipped", 0), + dropped=result.get("dropped", 0), + duration_ms=result.get("durationMs", 0), + updated_at=timezone.now(), + ) + + raise AssertionError( + "unreachable: the loop returns or raises" + ) # pragma: no cover + + @staticmethod + def is_retryable(err): + """ + Say whether handing the same document over again could go better. + + The collaboration server answers a 5xx or a 429 when it is unwell or + busy, and any other 4xx about the document itself — which will not fix + itself. Not reaching it at all is worth another try. + """ + if not isinstance(err, APIError): + return True + + return ( + err.status_code is None + or err.status_code >= 500 + or err.status_code in RETRY_STATUSES + ) + + @staticmethod + def save(migrations): + """Record what became of these documents, replacing what a previous run said.""" + if not migrations: + return + + models.DocumentMigration.objects.bulk_create( + migrations, + update_conflicts=True, + update_fields=[ + "status", + "versions", + "applied", + "skipped", + "dropped", + "duration_ms", + "error", + "updated_at", + ], + unique_fields=["document"], + ) diff --git a/src/backend/core/migrations/0033_documentmigration.py b/src/backend/core/migrations/0033_documentmigration.py new file mode 100644 index 000000000..9e7adcce9 --- /dev/null +++ b/src/backend/core/migrations/0033_documentmigration.py @@ -0,0 +1,92 @@ +# Generated by Django 5.2.14 on 2026-08-10 13:36 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0032_remove_linktrace_is_masked"), + ] + + operations = [ + migrations.CreateModel( + name="DocumentMigration", + fields=[ + ( + "document", + models.OneToOneField( + on_delete=django.db.models.deletion.CASCADE, + primary_key=True, + related_name="migration", + serialize=False, + to="core.document", + ), + ), + ( + "status", + models.CharField( + choices=[ + ("ok", "Migrated"), + ("already", "Already migrated"), + ("empty", "Nothing in the object storage"), + ("nothing", "No readable version"), + ("failed", "Failed"), + ], + max_length=10, + verbose_name="status", + ), + ), + ( + "versions", + models.PositiveIntegerField(default=0, verbose_name="versions"), + ), + ( + "applied", + models.PositiveIntegerField( + default=0, + help_text="versions that added content, one activity entry each", + verbose_name="applied", + ), + ), + ( + "skipped", + models.PositiveIntegerField( + default=0, + help_text="versions that could not be read", + verbose_name="skipped", + ), + ), + ( + "dropped", + models.PositiveIntegerField( + default=0, + help_text="versions older than the ones the server replays", + verbose_name="dropped", + ), + ), + ( + "duration_ms", + models.PositiveIntegerField(default=0, verbose_name="duration"), + ), + ( + "error", + models.TextField(blank=True, default="", verbose_name="error"), + ), + ( + "updated_at", + models.DateTimeField(auto_now=True, verbose_name="updated on"), + ), + ], + options={ + "verbose_name": "Document migration", + "verbose_name_plural": "Document migrations", + "db_table": "impress_document_migration", + "indexes": [ + models.Index( + fields=["status"], name="impress_doc_status_7d8208_idx" + ) + ], + }, + ), + ] diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 1b890f619..34e54c9cc 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -2133,3 +2133,66 @@ class Invitation(BaseModel): "partial_update": is_admin_or_owner, "retrieve": is_admin_or_owner, } + + +class DocumentMigrationStatus(models.TextChoices): + """What became of a document handed to the collaboration server to migrate.""" + + MIGRATED = "ok", _("Migrated") + ALREADY = "already", _("Already migrated") + EMPTY = "empty", _("Nothing in the object storage") + NOTHING = "nothing", _("No readable version") + FAILED = "failed", _("Failed") + + +class DocumentMigration(models.Model): + """ + What the collaboration server did with the legacy content of a document. + + The ledger of the backfill: the collaboration server keeps its own set of + the documents it migrated, but only of those it actually wrote history for. + A document it found nothing for is not in it and would be handed over again + on every run, and a valkey configured to evict would lose the set entirely. + This table is what the command reads to know what is left to do, and what + an operator reads to know how it went. + """ + + document = models.OneToOneField( + Document, + on_delete=models.CASCADE, + related_name="migration", + primary_key=True, + ) + status = models.CharField( + max_length=10, + choices=DocumentMigrationStatus.choices, + verbose_name=_("status"), + ) + versions = models.PositiveIntegerField(default=0, verbose_name=_("versions")) + applied = models.PositiveIntegerField( + default=0, + verbose_name=_("applied"), + help_text=_("versions that added content, one activity entry each"), + ) + skipped = models.PositiveIntegerField( + default=0, + verbose_name=_("skipped"), + help_text=_("versions that could not be read"), + ) + dropped = models.PositiveIntegerField( + default=0, + verbose_name=_("dropped"), + help_text=_("versions older than the ones the server replays"), + ) + duration_ms = models.PositiveIntegerField(default=0, verbose_name=_("duration")) + error = models.TextField(blank=True, default="", verbose_name=_("error")) + updated_at = models.DateTimeField(auto_now=True, verbose_name=_("updated on")) + + class Meta: + db_table = "impress_document_migration" + verbose_name = _("Document migration") + verbose_name_plural = _("Document migrations") + indexes = [models.Index(fields=["status"])] + + def __str__(self): + return f"{self.document_id!s}: {self.status:s}" diff --git a/src/backend/core/services/yhub_services.py b/src/backend/core/services/yhub_services.py index c67758f10..8ce04fa27 100644 --- a/src/backend/core/services/yhub_services.py +++ b/src/backend/core/services/yhub_services.py @@ -183,12 +183,14 @@ class YHubService: """ return {"X-User-Id": str(user_id)} if user_id else {} - def request(self, method, url, data=None, headers=None): + # pylint: disable-next=too-many-arguments + def request(self, method, url, *, data=None, headers=None, timeout=None): """ Send an authenticated request to the yhub API, asking it for JSON. Return the raw response, it is up to the caller to decode its body: the - endpoints do not all answer with the same payload. + endpoints do not all answer with the same payload. An endpoint doing + more than answering a document passes its own timeout. """ try: response = requests.request( @@ -201,7 +203,7 @@ class YHubService: "Accept": "application/json", **(headers or {}), }, - timeout=self.timeout, + timeout=timeout or self.timeout, ) except requests.RequestException as err: logger.exception("yhub service error: url=%s", url) @@ -285,6 +287,34 @@ class YHubService: }, ) + def migrate(self, document, force=False): + """ + Replay the legacy version history of a document into the collaboration server. + + The content of the documents used to live in our object storage, one + version of `{id}/file` per save. yhub reads them all back and rebuilds + the history with the timestamps of those versions, which is what makes + its activity line up with the versions we report. + + Answers what became of the document: `ok` when this call wrote its + history, `already` when a previous one did, `empty` when there is + nothing in the object storage (a document born in yhub) and `nothing` + when none of its versions could be read. All four are terminal, only a + failure to reach yhub raises. + + Forcing a document that is already migrated attributes its content a + second time: it is for a document whose yhub state was wiped, never for + a retry. + """ + url = self.build_url("migrate", document) + response = self.request( + "post", + f"{url}?force=true" if force else url, + timeout=settings.YHUB_MIGRATION_TIMEOUT, + ) + + return self.json_body(response) + def reset_connections(self, document, user_id=None): """ Re-check the access of the clients connected to a document. diff --git a/src/backend/core/tests/commands/test_migrate_documents.py b/src/backend/core/tests/commands/test_migrate_documents.py new file mode 100644 index 000000000..de580689e --- /dev/null +++ b/src/backend/core/tests/commands/test_migrate_documents.py @@ -0,0 +1,169 @@ +"""Unit tests for the `migrate_documents` command.""" + +from io import StringIO +from unittest import mock + +from django.core.management import call_command + +import pytest + +from core import factories, models +from core.services.yhub_services import APIError, ServiceUnavailableError, YHubService + +pytestmark = pytest.mark.django_db + + +def migrated(status="ok", **stats): + """What the collaboration server answers for a document it migrated.""" + return {"status": status, "migrated": status == "ok", **stats} + + +@pytest.fixture(name="collaboration_server", autouse=True) +def collaboration_server_fixture(): + """Answer every document as migrated, unless a test says otherwise.""" + with mock.patch.object( + YHubService, "migrate", return_value=migrated() + ) as mock_migrate: + yield mock_migrate + + +def run_command(**options): + """Run the command and return what it wrote.""" + stdout = StringIO() + call_command("migrate_documents", stdout=stdout, **options) + + return stdout.getvalue() + + +def test_commands_migrate_documents(collaboration_server): + """Every document should be handed to the collaboration server, once.""" + documents = factories.DocumentFactory.create_batch(3) + collaboration_server.return_value = migrated( + versions=4, applied=3, skipped=1, dropped=2, durationMs=42 + ) + + output = run_command() + + assert collaboration_server.call_count == 3 + handed = {call.args[0].pk for call in collaboration_server.call_args_list} + assert handed == {document.pk for document in documents} + + assert models.DocumentMigration.objects.count() == 3 + migration = models.DocumentMigration.objects.first() + assert migration.status == models.DocumentMigrationStatus.MIGRATED + assert (migration.versions, migration.applied) == (4, 3) + assert (migration.skipped, migration.dropped) == (1, 2) + assert migration.duration_ms == 42 + assert "ok=3" in output + + +@pytest.mark.parametrize("status", ["ok", "already", "empty", "nothing"]) +def test_commands_migrate_documents_records_every_outcome(collaboration_server, status): + """The four answers of the collaboration server are all terminal.""" + document = factories.DocumentFactory() + collaboration_server.return_value = migrated(status) + + run_command() + + assert models.DocumentMigration.objects.get(document=document).status == status + + # none of them is handed over again + run_command() + + assert collaboration_server.call_count == 1 + + +def test_commands_migrate_documents_failure_is_recorded_and_retried( + collaboration_server, +): + """A document that could not be migrated should be left for another run.""" + document = factories.DocumentFactory() + collaboration_server.side_effect = APIError("yhub is confused", status_code=400) + + output = run_command(retries=1) + + migration = models.DocumentMigration.objects.get(document=document) + assert migration.status == models.DocumentMigrationStatus.FAILED + assert "yhub is confused" in migration.error + assert "failed=1" in output + + # left alone by a plain run, handed over again when asked for + run_command() + assert collaboration_server.call_count == 1 + + collaboration_server.side_effect = None + collaboration_server.return_value = migrated() + run_command(retry_failed=True) + + assert collaboration_server.call_count == 2 + assert ( + models.DocumentMigration.objects.get(document=document).status + == models.DocumentMigrationStatus.MIGRATED + ) + + +def test_commands_migrate_documents_retries_a_server_that_is_unwell( + collaboration_server, +): + """A 5xx is about the server, the same document is worth handing over again.""" + document = factories.DocumentFactory() + collaboration_server.side_effect = [ + ServiceUnavailableError("connection reset"), + migrated(), + ] + + with mock.patch("time.sleep"): # no backoff wait in tests + run_command(retries=2) + + assert collaboration_server.call_count == 2 + assert ( + models.DocumentMigration.objects.get(document=document).status + == models.DocumentMigrationStatus.MIGRATED + ) + + +def test_commands_migrate_documents_does_not_retry_a_refused_document( + collaboration_server, +): + """A 4xx is about the document, insisting would only waste the server.""" + factories.DocumentFactory() + collaboration_server.side_effect = APIError("Room name is invalid", status_code=400) + + with mock.patch("time.sleep"): + run_command(retries=3) + + assert collaboration_server.call_count == 1 + + +def test_commands_migrate_documents_limit(collaboration_server): + """The most recently edited documents should be migrated first.""" + factories.DocumentFactory.create_batch(3) + recent = factories.DocumentFactory() + + run_command(limit=1) + + assert collaboration_server.call_count == 1 + assert collaboration_server.call_args[0][0].pk == recent.pk + + +def test_commands_migrate_documents_created_before(collaboration_server): + """A document created after the cutover has no legacy content to migrate.""" + old = factories.DocumentFactory() + models.Document.objects.filter(pk=old.pk).update(created_at="2020-01-01T00:00:00Z") + factories.DocumentFactory() + + run_command(created_before="2021-01-01T00:00:00Z") + + assert collaboration_server.call_count == 1 + assert collaboration_server.call_args[0][0].pk == old.pk + + +def test_commands_migrate_documents_dry_run(collaboration_server): + """A dry run should count the documents and call nothing.""" + factories.DocumentFactory.create_batch(2) + + output = run_command(dry_run=True) + + assert "2 documents to migrate" in output + collaboration_server.assert_not_called() + assert not models.DocumentMigration.objects.exists() diff --git a/src/backend/core/tests/test_integration_yhub_migration.py b/src/backend/core/tests/test_integration_yhub_migration.py index 36f06c053..89e0f8341 100644 --- a/src/backend/core/tests/test_integration_yhub_migration.py +++ b/src/backend/core/tests/test_integration_yhub_migration.py @@ -31,10 +31,8 @@ from django.core.files.storage import default_storage import pytest import requests -from core.services.jwt_services import JWTService +from core.services.jwt_services import Audiences, JWTService -# yhub verifies that its own name is the audience of the token (server.js) -YHUB_AUDIENCE = "yhub" # the org yhub is configured with; documents live under /docs/{docid} YHUB_ORG = "docs" TIMEOUT = 10 @@ -80,7 +78,8 @@ def admin_headers_fixture(settings): # pylint: disable=redefined-outer-name """ if not settings.JWT_PRIVATE_KEY: pytest.skip("JWT_PRIVATE_KEY is not configured") - token = JWTService().get_admin_token({"aud": YHUB_AUDIENCE}) + # yhub verifies that its own name is the audience of the token (server.js) + token = JWTService().get_admin_token(Audiences.YHUB) return {"Authorization": f"Bearer {token}"} @@ -330,7 +329,11 @@ def test_integration_yhub_full_migration_is_idempotent(admin_headers): again = _migrate(docid, admin_headers) assert again.status_code == 200 - assert again.json() == {"message": "Already migrated", "migrated": False} + assert again.json() == { + "status": "already", + "message": "Already migrated", + "migrated": False, + } forced = _migrate(docid, admin_headers, force="true") assert forced.json()["migrated"] is True diff --git a/src/backend/core/tests/test_services_yhub_services.py b/src/backend/core/tests/test_services_yhub_services.py index ffb90cc41..9379188c3 100644 --- a/src/backend/core/tests/test_services_yhub_services.py +++ b/src/backend/core/tests/test_services_yhub_services.py @@ -204,6 +204,44 @@ def test_create_ydoc_already_exists(mock_request): assert excinfo.value.status_code == 409 +@patch("requests.request") +def test_migrate(mock_request): + """Should ask yhub to replay the legacy history, and answer what it did.""" + mock_request.return_value.ok = True + mock_request.return_value.json.return_value = { + "status": "ok", + "message": "Migration completed", + "migrated": True, + "versions": 12, + "applied": 9, + "durationMs": 1234, + } + + result = YHubService().migrate(DOCUMENT) + + assert result["status"] == "ok" + assert result["applied"] == 9 + args, kwargs = mock_request.call_args + assert args == ( + "post", + f"http://yhub:3002/collaboration/migrate/v1/docs/{DOCUMENT.id!s}", + ) + # reading every version of a document takes longer than any other call + assert kwargs["timeout"] == 600 + + +@patch("requests.request") +def test_migrate_forced(mock_request): + """Forcing a document that is already migrated should be asked for explicitly.""" + mock_request.return_value.ok = True + mock_request.return_value.json.return_value = {"status": "ok"} + + YHubService().migrate(DOCUMENT, force=True) + + args, _kwargs = mock_request.call_args + assert args[1].endswith("?force=true") + + @patch("requests.request") def test_reset_connections(mock_request): """Should ask yhub to re-check every connection of the document.""" diff --git a/src/backend/impress/settings.py b/src/backend/impress/settings.py index 724331036..f1b3fb180 100755 --- a/src/backend/impress/settings.py +++ b/src/backend/impress/settings.py @@ -540,6 +540,14 @@ class Base(Configuration): environ_name="YHUB_API_TIMEOUT", environ_prefix=None, ) + # Replaying the legacy history of a document reads every one of its S3 + # versions, so it is the one call that can take minutes. Timing it out does + # not stop the collaboration server, it only loses the answer. + YHUB_MIGRATION_TIMEOUT = values.IntegerValue( + default=600, + environ_name="YHUB_MIGRATION_TIMEOUT", + environ_prefix=None, + ) # JWT # RSA private key (PEM) used to sign the tokens issued by diff --git a/src/yhub-server/README.md b/src/yhub-server/README.md index a0653e42e..e9b15a4ef 100644 --- a/src/yhub-server/README.md +++ b/src/yhub-server/README.md @@ -292,8 +292,12 @@ Guarantees: - **More than 500 versions**: only the newest 500 are replayed and the rest fold into the first replayed version, reported as `dropped`. -Response (200): `{ migrated, versions, applied, skipped, dropped, bytes, -durationMs }`. +Response (200): `{ status, message, migrated, versions, applied, skipped, +dropped, bytes, durationMs }`. `status` is the machine-readable outcome a +backfill driver records — `ok`, `already`, `empty` (no legacy object, a +brand-new document) or `nothing` (versions exist, none readable) — all of them +done, which is why they share one 2xx. `migrated` says whether this very call +wrote the history. Caveats: diff --git a/src/yhub-server/server.js b/src/yhub-server/server.js index 3c952b387..14d65f218 100644 --- a/src/yhub-server/server.js +++ b/src/yhub-server/server.js @@ -429,31 +429,22 @@ const api = [ const { status, ...stats } = await fullMigrate(req.yhub, req.room, { force: req.query.force === 'true', }); - if (status === 'already') { - return jsonResponse(200, { - message: 'Already migrated', - migrated: false, - }); - } - if (status === 'empty') { - // brand-new documents never had a legacy object; nothing to replay - // and nothing wrong — a backfill driver treats this as done - return jsonResponse(200, { - message: 'No legacy document in s3', - migrated: false, - ...stats, - }); - } - if (status === 'nothing') { - return jsonResponse(200, { - message: 'No usable content in the legacy versions', - migrated: false, - ...stats, - }); - } + // `status` is what a backfill driver records per document: 'ok', + // 'already', 'empty' (no legacy object — a brand-new document) or + // 'nothing' (versions exist, none readable). All four are done, hence + // one 2xx; `message` says the same thing to a human, `migrated` + // whether this call is the one that wrote the history. + const messages = { + already: 'Already migrated', + empty: 'No legacy document in s3', + nothing: 'No usable content in the legacy versions', + ok: 'Migration completed', + }; + return jsonResponse(200, { - message: 'Migration completed', - migrated: true, + status, + message: messages[status], + migrated: status === 'ok', ...stats, }); },