From fc1bf3551a89babbb22ed3f951b3e9e8a44a97cb Mon Sep 17 00:00:00 2001 From: lsabor Date: Wed, 19 Aug 2026 14:41:20 -0700 Subject: [PATCH 1/6] feat: archive long private bot comment texts to S3 Long, private, bot-authored comments older than a month make up the bulk of the comments table. This moves their full text into S3 and leaves a 200-character stub behind in the database, keeping the original retrievable on demand. - add `archive_bot_comment_texts` management command (--dry-run, --limit, --batch-size), scheduled monthly from the cron runner - add `Comment.is_text_archived`; the S3 key is derived from the comment id, so no pointer needs to be stored on the row - store only the original text: the per-language columns are cleared, since bot/private comments are never translated - add GET /comments//full-text/, gated by the existing comment permission check (private comments resolve to author-only) - refuse to edit an archived comment, both in `update_comment` and in the admin, where the text columns become read-only The truncating UPDATE is guarded on `edited_at` so an edit racing the upload cannot lose text, and it bypasses save() so archiving neither bumps `edited_at` nor recomputes the search vector. The migration only adds the field. Archiving is never triggered by it. Co-Authored-By: Claude Opus 5 --- comments/admin.py | 23 +- .../commands/archive_bot_comment_texts.py | 76 ++++ .../0027_comment_is_text_archived.py | 18 + comments/models.py | 9 + comments/serializers/common.py | 1 + comments/services/common.py | 7 + comments/services/text_archive.py | 252 +++++++++++++ comments/tasks.py | 27 ++ comments/urls.py | 5 + comments/views/common.py | 29 +- metaculus_web/settings.py | 6 + misc/management/commands/cron.py | 9 + tests/unit/test_comments/test_text_archive.py | 330 ++++++++++++++++++ 13 files changed, 789 insertions(+), 3 deletions(-) create mode 100644 comments/management/commands/archive_bot_comment_texts.py create mode 100644 comments/migrations/0027_comment_is_text_archived.py create mode 100644 comments/services/text_archive.py create mode 100644 tests/unit/test_comments/test_text_archive.py diff --git a/comments/admin.py b/comments/admin.py index 259d64fe20..c6056e3a51 100644 --- a/comments/admin.py +++ b/comments/admin.py @@ -1,8 +1,10 @@ from admin_auto_filters.filters import AutocompleteFilterFactory +from django.conf import settings from django.contrib import admin from django.contrib.postgres.search import SearchQuery -from utils.models import CustomTranslationAdmin +from utils.models import CustomTranslationAdmin, uniques_ordered_list +from utils.translation import build_supported_localized_fieldname from .models import Comment, KeyFactor, KeyFactorDriver @@ -30,12 +32,14 @@ class CommentAdmin(CustomTranslationAdmin): "created_at", "is_soft_deleted", "is_private", + "is_text_archived", ] list_filter = [ AutocompleteFilterFactory("Author", "author"), AutocompleteFilterFactory("Post", "on_post"), "is_soft_deleted", "is_private", + "is_text_archived", AutocompleteFilterFactory("Project", "on_project"), ] autocomplete_fields = [ @@ -43,7 +47,7 @@ class CommentAdmin(CustomTranslationAdmin): "on_post", "on_project", ] - readonly_fields = ["included_forecast"] + readonly_fields = ["included_forecast", "is_text_archived"] fields = [ "author", "text", @@ -52,6 +56,7 @@ class CommentAdmin(CustomTranslationAdmin): "is_soft_deleted", "included_forecast", "is_private", + "is_text_archived", ] # `search_fields` must be non-empty for Django admin to render the search box # and dispatch to `get_search_results`, but its contents are unused because we @@ -62,6 +67,20 @@ class CommentAdmin(CustomTranslationAdmin): def should_update_translations(self, obj): return not obj.on_post.is_private() + def get_readonly_fields(self, request, obj=None): + readonly_fields = list(super().get_readonly_fields(request, obj)) + + if obj and obj.is_text_archived: + # Only a stub of the text is left in the db, so editing it here + # would bypass the `update_comment` guard and leave the row out of + # sync with the archived original + readonly_fields += ["text"] + [ + build_supported_localized_fieldname("text", lang) + for lang, _label in settings.LANGUAGES + ] + + return uniques_ordered_list(readonly_fields) + def get_search_results(self, request, queryset, search_term): search_term = search_term.strip() if not search_term: diff --git a/comments/management/commands/archive_bot_comment_texts.py b/comments/management/commands/archive_bot_comment_texts.py new file mode 100644 index 0000000000..2b08c872bb --- /dev/null +++ b/comments/management/commands/archive_bot_comment_texts.py @@ -0,0 +1,76 @@ +from django.core.management.base import BaseCommand, CommandError + +from comments.services.text_archive import ( + ARCHIVE_AGE_DAYS, + ARCHIVE_MIN_TEXT_LENGTH, + ARCHIVE_STUB_LENGTH, + DEFAULT_BATCH_SIZE, + archive_bot_comment_texts, + check_is_enabled, +) + + +class Command(BaseCommand): + help = ( + "Moves the full text of private bot comments older than " + f"{ARCHIVE_AGE_DAYS} days and longer than {ARCHIVE_MIN_TEXT_LENGTH} " + f"characters to S3, leaving a {ARCHIVE_STUB_LENGTH}-character stub in " + "the database. Runs monthly as a cron job; the full text stays " + "readable through the comment-full-text endpoint." + ) + + def add_arguments(self, parser): + parser.add_argument( + "--dry-run", + action="store_true", + help="Report what would be archived without writing to S3 or the database", + ) + parser.add_argument( + "--limit", + type=int, + default=None, + help="Maximum number of comments to archive (useful for the first backfill)", + ) + parser.add_argument( + "--batch-size", + type=int, + default=DEFAULT_BATCH_SIZE, + help=f"Comments per database update (default: {DEFAULT_BATCH_SIZE})", + ) + + def handle(self, *args, **options): + dry_run = options["dry_run"] + + if not check_is_enabled(): + raise CommandError( + "AWS_STORAGE_BUCKET_COMMENTS_TEXT is not configured, " + "comment text archiving is disabled." + ) + + stats = archive_bot_comment_texts( + dry_run=dry_run, + limit=options["limit"], + batch_size=options["batch_size"], + ) + + verb = "Would archive" if dry_run else "Archived" + self.stdout.write( + f"{verb} {stats.archived} comment(s), " + f"reclaiming {stats.chars_reclaimed:,} characters" + ) + + if stats.sample_ids: + sample = ", ".join(str(pk) for pk in stats.sample_ids) + self.stdout.write(f"Sample comment ids: {sample}") + + if stats.skipped: + self.stdout.write( + self.style.WARNING( + f"Skipped {stats.skipped} comment(s) edited during the run" + ) + ) + + if stats.failed: + self.stdout.write( + self.style.ERROR(f"Failed to upload {stats.failed} comment(s)") + ) diff --git a/comments/migrations/0027_comment_is_text_archived.py b/comments/migrations/0027_comment_is_text_archived.py new file mode 100644 index 0000000000..8963d98a6e --- /dev/null +++ b/comments/migrations/0027_comment_is_text_archived.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.15 on 2026-08-19 17:19 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('comments', '0026_comment_key_factor_votes_score'), + ] + + operations = [ + migrations.AddField( + model_name='comment', + name='is_text_archived', + field=models.BooleanField(default=False, editable=False, help_text='True if the full text has been moved to S3 and only a truncated stub remains in the text columns. Archived comments cannot be edited; use the comment-full-text endpoint to read them.'), + ), + ] diff --git a/comments/models.py b/comments/models.py index c35836561b..b23585ba5a 100644 --- a/comments/models.py +++ b/comments/models.py @@ -99,6 +99,15 @@ class Comment(TimeStampedModel, TranslatedModel): is_soft_deleted = models.BooleanField(default=False, db_index=True) # Some comments with KeyFactors can have empty text text = models.TextField(max_length=150_000, blank=True) + # Set by the `archive_bot_comment_texts` command. The full text lives in S3 + # under a key derived from the comment id, so no pointer is stored here. + is_text_archived = models.BooleanField( + default=False, + editable=False, + help_text="True if the full text has been moved to S3 and only a " + "truncated stub remains in the text columns. Archived comments " + "cannot be edited; use the comment-full-text endpoint to read them.", + ) on_post = models.ForeignKey( Post, models.CASCADE, null=True, related_name="comments" ) diff --git a/comments/serializers/common.py b/comments/serializers/common.py index 4d435c4f42..c3f334de8f 100644 --- a/comments/serializers/common.py +++ b/comments/serializers/common.py @@ -76,6 +76,7 @@ class Meta: "text_edited_at", "is_soft_deleted", "text", + "is_text_archived", "on_post", "on_post_data", "included_forecast", diff --git a/comments/services/common.py b/comments/services/common.py index 418f0bf6e6..62e0baba3f 100644 --- a/comments/services/common.py +++ b/comments/services/common.py @@ -187,6 +187,13 @@ def perform_create_comment( def update_comment( comment: Comment, text: str = None, included_forecast: Forecast = None ): + if comment.is_text_archived: + # Only a stub of the text remains in the db, so we can neither diff + # against it nor let it be overwritten + raise ValidationError( + "This comment's text has been archived and can no longer be edited." + ) + differ = difflib.Differ() diff = list(differ.compare(comment.text.splitlines(), text.splitlines())) diff --git a/comments/services/text_archive.py b/comments/services/text_archive.py new file mode 100644 index 0000000000..760835f10a --- /dev/null +++ b/comments/services/text_archive.py @@ -0,0 +1,252 @@ +import json +import logging +from dataclasses import dataclass, field +from datetime import timedelta + +from django.conf import settings +from django.core.serializers.json import DjangoJSONEncoder +from django.db.models import Count, Q, QuerySet, Sum, Value +from django.db.models.functions import Coalesce, Length, NullIf, Substr +from django.utils import timezone + +from comments.models import Comment +from utils.aws import get_boto_client +from utils.translation import build_supported_localized_fieldname + +logger = logging.getLogger(__name__) + +# Comments older than this are eligible for archiving +ARCHIVE_AGE_DAYS = 30 +# Only archive comments whose text is longer than this. Below this, it's not important +# to move. +ARCHIVE_MIN_TEXT_LENGTH = 2000 +# Length of the stub left behind in the text columns +ARCHIVE_STUB_LENGTH = 200 + +S3_KEY_PREFIX = "comments_text" + +DEFAULT_BATCH_SIZE = 500 + +# `text` is the base column shadowed by modeltranslation: it holds a duplicate +# of the original content that is written on save but never read back (reads of +# `comment.text` resolve to `text_original` through the translation +# descriptor). `text_original` may be NULL or empty on rows that were never +# saved through the descriptor, so fall back to the base column. +ORIGINAL_TEXT = Coalesce(NullIf("text_original", Value("")), "text") + + +def check_is_enabled() -> bool: + return bool(settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT) + + +def build_key(comment_id: int) -> str: + """ + The archive key is derived from the comment id, so it never needs to be + stored on the comment itself. This function is the only place that knows + the key layout. + """ + + return f"{S3_KEY_PREFIX}/{comment_id}.json" + + +def upload_text(comment_id: int, text: str) -> str: + """ + Uploads the full original text of a comment to S3 and returns the key. + + Only the original text is stored: bot/private comments are never + translated (see `trigger_update_comment_translations`), and storing + machine translations of an archived text would be pointless anyway. + """ + + s3 = get_boto_client("s3") + key = build_key(comment_id) + + s3.put_object( + Bucket=settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT, + Key=key, + Body=json.dumps( + { + "comment_id": comment_id, + "archived_at": timezone.now(), + "text": text, + }, + cls=DjangoJSONEncoder, + ), + ContentType="application/json", + ) + + return key + + +def fetch_text(comment_id: int) -> str | None: + """ + Reads the archived full text of a comment back from S3. + Returns None if the object is missing. + """ + + s3 = get_boto_client("s3") + + try: + obj = s3.get_object( + Bucket=settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT, + Key=build_key(comment_id), + ) + except s3.exceptions.NoSuchKey: + logger.error("Archived text is missing for comment %s", comment_id) + + return None + + return json.loads(obj["Body"].read().decode("utf-8"))["text"] + + +def get_full_text(comment: Comment) -> str | None: + """ + Full text of a comment, transparently reading from the archive when the + stored text has been truncated. + """ + + if not comment.is_text_archived: + return comment.text + + return fetch_text(comment.pk) + + +def get_archivable_comments() -> QuerySet[Comment]: + """ + Long private bot comments old enough to be archived. + + Soft-deleted comments are included: their text is not rendered anywhere, + but it still occupies the row, and archiving keeps it recoverable. + """ + + cutoff = timezone.now() - timedelta(days=ARCHIVE_AGE_DAYS) + + return ( + Comment.objects.filter( + author__is_bot=True, + is_private=True, + is_text_archived=False, + created_at__lt=cutoff, + ) + .annotate(text_length=Length(ORIGINAL_TEXT)) + .filter(text_length__gt=ARCHIVE_MIN_TEXT_LENGTH) + ) + + +@dataclass +class ArchiveStats: + archived: int = 0 + failed: int = 0 + skipped: int = 0 + chars_reclaimed: int = 0 + sample_ids: list[int] = field(default_factory=list) + + +def _build_truncate_kwargs() -> dict: + """ + Update kwargs that leave a stub in both copies of the original text and + drop every machine translation. + """ + + stub = Substr(ORIGINAL_TEXT, 1, ARCHIVE_STUB_LENGTH) + kwargs = {"text": stub, "text_original": stub, "is_text_archived": True} + + for lang, _label in settings.LANGUAGES: + if lang == settings.ORIGINAL_LANGUAGE_CODE: + continue + + kwargs[build_supported_localized_fieldname("text", lang)] = None + + return kwargs + + +def archive_bot_comment_texts( + dry_run: bool = False, + limit: int | None = None, + batch_size: int = DEFAULT_BATCH_SIZE, +) -> ArchiveStats: + """ + Moves the full text of long, private, old bot comments to S3, leaving a + truncated stub in the database. + """ + + stats = ArchiveStats() + queryset = get_archivable_comments() + + if dry_run: + # Aggregate without transferring any text + totals = queryset.aggregate(count=Count("id"), chars=Sum("text_length")) + count = totals["count"] or 0 + + if limit is not None: + count = min(count, limit) + + stats.archived = count + stats.chars_reclaimed = max( + (totals["chars"] or 0) - count * ARCHIVE_STUB_LENGTH, 0 + ) + stats.sample_ids = list( + queryset.order_by("id").values_list("id", flat=True)[:5] + ) + + return stats + + started_at = timezone.now() + truncate_kwargs = _build_truncate_kwargs() + cursor = 0 + + while limit is None or stats.archived + stats.failed < limit: + page_size = batch_size + if limit is not None: + page_size = min(batch_size, limit - stats.archived - stats.failed) + + rows = list( + queryset.filter(id__gt=cursor) + .order_by("id") + .values("id", "text", "text_original", "text_length")[:page_size] + ) + + if not rows: + break + + # Advance past the whole page, including rows that failed to upload, so + # a persistent failure can never stall the run. Skipped rows stay + # eligible for the next one. + cursor = rows[-1]["id"] + uploaded_ids = [] + + for row in rows: + text = row["text_original"] or row["text"] + + try: + upload_text(row["id"], text) + except Exception: + logger.exception("Failed to archive text of comment %s", row["id"]) + stats.failed += 1 + + continue + + uploaded_ids.append(row["id"]) + + if not uploaded_ids: + continue + + # Only truncate rows that have not been touched since the run began, so + # an edit racing the upload can never lose text. `edited_at` is + # nullable on rows that predate TimeStampedModel.save. + untouched = Comment.objects.filter(pk__in=uploaded_ids).filter( + Q(edited_at__lt=started_at) | Q(edited_at__isnull=True) + ) + archived_ids = set(untouched.values_list("id", flat=True)) + updated = untouched.update(**truncate_kwargs) + + stats.archived += updated + stats.skipped += len(uploaded_ids) - updated + stats.chars_reclaimed += sum( + max(row["text_length"] - ARCHIVE_STUB_LENGTH, 0) + for row in rows + if row["id"] in archived_ids + ) + stats.sample_ids = (stats.sample_ids + sorted(archived_ids))[:5] + + return stats diff --git a/comments/tasks.py b/comments/tasks.py index 3da964b152..d4ccc1a8b9 100644 --- a/comments/tasks.py +++ b/comments/tasks.py @@ -101,3 +101,30 @@ def update_current_top_comments_of_week(): # Update the week before week_start_date = week_start_date - timedelta(days=7) update_top_comments_of_week(week_start_date) + + +@dramatiq.actor +def job_archive_bot_comment_texts(): + # Import here to avoid circular imports + from comments.services.text_archive import ( + archive_bot_comment_texts, + check_is_enabled, + ) + + if not check_is_enabled(): + # Logged as an error rather than skipped silently: once this job is + # scheduled, a missing bucket means the monthly cleanup never runs + logger.error( + "AWS_STORAGE_BUCKET_COMMENTS_TEXT is not configured, " + "comment text archiving cannot run" + ) + + return + + stats = archive_bot_comment_texts() + + logger.info( + f"Archived the text of {stats.archived} bot comment(s), " + f"reclaiming {stats.chars_reclaimed} characters " + f"({stats.failed} failed, {stats.skipped} skipped)" + ) diff --git a/comments/urls.py b/comments/urls.py index 6c3a590a19..f98ab9e933 100644 --- a/comments/urls.py +++ b/comments/urls.py @@ -10,6 +10,11 @@ name="comment-delete", ), path("comments//edit/", common.comment_edit_api_view, name="comment-edit"), + path( + "comments//full-text/", + common.comment_full_text_api_view, + name="comment-full-text", + ), path("comments//vote/", common.comment_vote_api_view, name="comment-vote"), path( "comments//toggle_cmm/", diff --git a/comments/views/common.py b/comments/views/common.py index 2c5a3c1371..2f5b30b18d 100644 --- a/comments/views/common.py +++ b/comments/views/common.py @@ -4,7 +4,7 @@ from django.utils import timezone from rest_framework import serializers, status from rest_framework.decorators import api_view, permission_classes -from rest_framework.exceptions import PermissionDenied, ValidationError +from rest_framework.exceptions import NotFound, PermissionDenied, ValidationError from rest_framework.permissions import AllowAny, IsAuthenticated, IsAdminUser from rest_framework.request import Request from rest_framework.response import Response @@ -24,6 +24,7 @@ serialize_comments_of_the_week_many, ) from comments.services.common import ( + get_comment_permission_for_user, set_comment_excluded_from_week_top, create_comment, perform_create_comment, @@ -35,6 +36,7 @@ toggle_cmm, ) from comments.services.feed import get_comments_feed +from comments.services.text_archive import get_full_text from notifications.services import send_comment_report_notification_to_staff from posts.services.common import get_post_permission_for_user from projects.permissions import ObjectPermission @@ -232,6 +234,31 @@ def comment_report_api_view(request, pk=int): return Response(status=status.HTTP_204_NO_CONTENT) +@api_view(["GET"]) +def comment_full_text_api_view(request: Request, pk: int): + """ + Returns the untruncated text of a single comment, reading it back from the + archive if it has been moved out of the database. + """ + + comment = get_object_or_404(Comment, pk=pk) + + # Private comments resolve to no permission for anyone but their author + permission = get_comment_permission_for_user(comment, user=request.user) + ObjectPermission.can_view(permission, raise_exception=True) + + if comment.is_soft_deleted: + # Mirrors the comment serializer, which never exposes deleted text + raise PermissionDenied("This comment has been deleted.") + + text = get_full_text(comment) + + if text is None: + raise NotFound("The archived text of this comment could not be retrieved.") + + return Response({"id": comment.pk, "text": text}) + + @api_view(["POST"]) def comment_create_oldapi_view(request: Request): """ diff --git a/metaculus_web/settings.py b/metaculus_web/settings.py index 27eed75342..54d8c71e59 100644 --- a/metaculus_web/settings.py +++ b/metaculus_web/settings.py @@ -429,6 +429,12 @@ def get_jwt_encryption_config(): AWS_STORAGE_BUCKET_POST_VERSION_HISTORY = os.environ.get( "AWS_STORAGE_BUCKET_POST_VERSION_HISTORY" ) +# S3 bucket holding the `comments_text/` prefix of archived comment texts. +# Comment text archiving will be disabled if this isn’t set. There is +# deliberately no fallback to another bucket: the archive is the only copy of +# the text, so a missing setting must disable the feature rather than silently +# write somewhere unintended. +AWS_STORAGE_BUCKET_COMMENTS_TEXT = os.environ.get("AWS_STORAGE_BUCKET_COMMENTS_TEXT") # Cloudflare captcha # https://developers.cloudflare.com/turnstile/get-started/server-side-validation/ diff --git a/misc/management/commands/cron.py b/misc/management/commands/cron.py index 272c1bd62f..7833783a6b 100644 --- a/misc/management/commands/cron.py +++ b/misc/management/commands/cron.py @@ -10,6 +10,7 @@ from comments.tasks import ( update_current_top_comments_of_week, + job_archive_bot_comment_texts, job_finalize_and_send_weekly_top_comments, ) from misc.jobs import sync_itn_articles @@ -241,6 +242,14 @@ def handle(self, *args, **options): max_instances=1, replace_existing=True, ) + scheduler.add_job( + close_old_connections(job_archive_bot_comment_texts.send), + # First day of every month at 04:00 + trigger=CronTrigger.from_crontab("0 4 1 * *"), + id="comments_archive_bot_comment_texts", + max_instances=1, + replace_existing=True, + ) # # Cache warm-up jobs diff --git a/tests/unit/test_comments/test_text_archive.py b/tests/unit/test_comments/test_text_archive.py new file mode 100644 index 0000000000..7e8a214f51 --- /dev/null +++ b/tests/unit/test_comments/test_text_archive.py @@ -0,0 +1,330 @@ +import json +from datetime import timedelta +from io import StringIO + +import pytest # noqa +from django.core.management import call_command +from django.core.management.base import CommandError +from django.urls import reverse +from django.utils import timezone +from rest_framework.exceptions import ValidationError + +from comments.models import Comment +from comments.services.common import update_comment +from comments.services.text_archive import ( + ARCHIVE_MIN_TEXT_LENGTH, + ARCHIVE_STUB_LENGTH, + archive_bot_comment_texts, + build_key, + get_archivable_comments, +) +from posts.models import Post +from projects.permissions import ObjectPermission +from tests.unit.test_comments.factories import factory_comment +from tests.unit.test_posts.factories import factory_post +from tests.unit.test_projects.factories import factory_project +from tests.unit.test_questions.conftest import * # noqa +from tests.unit.test_users.factories import factory_user + +LONG_TEXT = "b" * (ARCHIVE_MIN_TEXT_LENGTH + 500) + + +@pytest.fixture() +def bot(user1): + return factory_user(username="bot1", email="bot1@metaculus.com", is_bot=True) + + +@pytest.fixture() +def post(user1): + return factory_post( + author=user1, + default_project=factory_project( + default_permission=ObjectPermission.FORECASTER, + ), + curation_status=Post.CurationStatus.APPROVED, + ) + + +def factory_archivable_comment(author, post, text=LONG_TEXT, **kwargs): + kwargs.setdefault("created_at", timezone.now() - timedelta(days=60)) + kwargs.setdefault("is_private", True) + + return factory_comment( + author=author, + on_post=post, + text=text, + text_original=text, + **kwargs, + ) + + +@pytest.fixture() +def s3_stub(mocker): + """ + Minimal in-memory stand-in for the S3 client used by the archive service. + """ + + objects = {} + + class Client: + class exceptions: + class NoSuchKey(Exception): + pass + + def put_object(self, Bucket, Key, Body, **kwargs): + objects[Key] = Body + + def get_object(self, Bucket, Key): + if Key not in objects: + raise Client.exceptions.NoSuchKey() + + return {"Body": mocker.Mock(read=lambda: objects[Key].encode("utf-8"))} + + mocker.patch( + "comments.services.text_archive.get_boto_client", return_value=Client() + ) + mocker.patch("django.conf.settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT", "test-bucket") + + return objects + + +class TestArchivableCommentsQueryset: + def test_includes_long_old_private_bot_comments(self, bot, post): + comment = factory_archivable_comment(bot, post) + + assert list(get_archivable_comments()) == [comment] + + def test_excludes_recent_comments(self, bot, post): + factory_archivable_comment(bot, post, created_at=timezone.now()) + + assert not get_archivable_comments().exists() + + def test_excludes_short_comments(self, bot, post): + factory_archivable_comment(bot, post, text="a" * ARCHIVE_MIN_TEXT_LENGTH) + + assert not get_archivable_comments().exists() + + def test_excludes_public_comments(self, bot, post): + factory_archivable_comment(bot, post, is_private=False) + + assert not get_archivable_comments().exists() + + def test_excludes_human_comments(self, user1, post): + factory_archivable_comment(user1, post) + + assert not get_archivable_comments().exists() + + def test_excludes_already_archived_comments(self, bot, post): + factory_archivable_comment(bot, post, is_text_archived=True) + + assert not get_archivable_comments().exists() + + def test_includes_soft_deleted_comments(self, bot, post): + comment = factory_archivable_comment(bot, post, is_soft_deleted=True) + + assert list(get_archivable_comments()) == [comment] + + +class TestArchiveBotCommentTexts: + def test_archives_text_to_s3_and_truncates_row(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + + stats = archive_bot_comment_texts() + + assert stats.archived == 1 + assert stats.failed == 0 + assert stats.skipped == 0 + assert stats.chars_reclaimed == len(LONG_TEXT) - ARCHIVE_STUB_LENGTH + + # Full text is in S3 + payload = json.loads(s3_stub[build_key(comment.pk)]) + assert payload["comment_id"] == comment.pk + assert payload["text"] == LONG_TEXT + + # Only a stub is left in both copies of the original text + comment.refresh_from_db() + assert comment.is_text_archived is True + assert comment.text_original == LONG_TEXT[:ARCHIVE_STUB_LENGTH] + assert ( + Comment.objects.filter(pk=comment.pk).values_list("text", flat=True)[0] + == LONG_TEXT[:ARCHIVE_STUB_LENGTH] + ) + + def test_drops_translations(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post, text_en="translated") + + archive_bot_comment_texts() + + comment.refresh_from_db() + assert comment.text_en is None + + def test_does_not_bump_edited_at(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + edited_at = comment.edited_at + + archive_bot_comment_texts() + + comment.refresh_from_db() + assert comment.edited_at == edited_at + + def test_is_idempotent(self, bot, post, s3_stub): + factory_archivable_comment(bot, post) + + assert archive_bot_comment_texts().archived == 1 + assert archive_bot_comment_texts().archived == 0 + + def test_dry_run_writes_nothing(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + + stats = archive_bot_comment_texts(dry_run=True) + + assert stats.archived == 1 + assert stats.chars_reclaimed == len(LONG_TEXT) - ARCHIVE_STUB_LENGTH + assert stats.sample_ids == [comment.pk] + + assert s3_stub == {} + comment.refresh_from_db() + assert comment.is_text_archived is False + assert comment.text_original == LONG_TEXT + + def test_upload_failure_leaves_comment_intact(self, bot, post, s3_stub, mocker): + comment = factory_archivable_comment(bot, post) + mocker.patch( + "comments.services.text_archive.upload_text", + side_effect=RuntimeError("s3 is down"), + ) + + stats = archive_bot_comment_texts() + + assert stats.archived == 0 + assert stats.failed == 1 + + comment.refresh_from_db() + assert comment.is_text_archived is False + assert comment.text_original == LONG_TEXT + + def test_respects_limit(self, bot, post, s3_stub): + factory_archivable_comment(bot, post) + factory_archivable_comment(bot, post) + + assert archive_bot_comment_texts(limit=1).archived == 1 + assert get_archivable_comments().count() == 1 + + def test_processes_multiple_batches(self, bot, post, s3_stub): + for _ in range(5): + factory_archivable_comment(bot, post) + + assert archive_bot_comment_texts(batch_size=2).archived == 5 + assert not get_archivable_comments().exists() + + +class TestArchiveCommand: + def test_dry_run_reports_without_writing(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + out = StringIO() + + call_command("archive_bot_comment_texts", "--dry-run", stdout=out) + + assert "Would archive 1 comment(s)" in out.getvalue() + assert s3_stub == {} + comment.refresh_from_db() + assert comment.is_text_archived is False + + def test_archives(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + out = StringIO() + + call_command("archive_bot_comment_texts", stdout=out) + + assert "Archived 1 comment(s)" in out.getvalue() + comment.refresh_from_db() + assert comment.is_text_archived is True + + def test_errors_when_bucket_is_not_configured(self, bot, post, mocker): + mocker.patch("django.conf.settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT", None) + + with pytest.raises(CommandError): + call_command("archive_bot_comment_texts", "--dry-run") + + +class TestArchivedCommentEditing: + def test_archived_comment_cannot_be_edited(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + comment.refresh_from_db() + + with pytest.raises(ValidationError): + update_comment(comment, text="new text") + + def test_unarchived_comment_can_still_be_edited(self, bot, post): + comment = factory_archivable_comment(bot, post, created_at=timezone.now()) + + update_comment(comment, text="new text") + + comment.refresh_from_db() + assert comment.text == "new text" + + +class TestCommentFullTextApiView: + def test_author_reads_archived_text( + self, bot, post, s3_stub, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + + response = create_client_for_user(bot).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 200 + assert response.data["text"] == LONG_TEXT + + def test_returns_db_text_when_not_archived( + self, bot, post, s3_stub, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + + response = create_client_for_user(bot).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 200 + assert response.data["text"] == LONG_TEXT + + def test_other_user_cannot_read_private_comment( + self, bot, post, s3_stub, user2, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + + response = create_client_for_user(user2).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 403 + + def test_soft_deleted_comment_is_not_readable( + self, bot, post, s3_stub, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + Comment.objects.filter(pk=comment.pk).update(is_soft_deleted=True) + + response = create_client_for_user(bot).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 403 + + def test_missing_archive_object_returns_404( + self, bot, post, s3_stub, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + s3_stub.clear() + + response = create_client_for_user(bot).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 404 From 0c7fedbde3ecbbe819f4c9bcc895f188f8a0f08e Mon Sep 17 00:00:00 2001 From: lsabor Date: Thu, 20 Aug 2026 10:33:21 -0700 Subject: [PATCH 2/6] fix: truncate the base text column when archiving bot comments The archiving run left the full text in the base `text` column on every row it processed, forfeiting roughly half the space the feature exists to reclaim. On a full local run that was 14 GB across 328,988 rows. `Comment` is registered with modeltranslation, whose `MultilingualQuerySet` rewrites every mention of `text` into the current language's column. The `text=stub` kwarg was rewritten to `text_original=stub`, colliding with the `text_original` kwarg already present, so the base column was never written. The same rewriting affected the read path: `values("text")` returned `text_original`, and `Length(ORIGINAL_TEXT)` measured `text_original` twice instead of falling back to `text`, hiding rows whose text only lives in the base column from the eligibility filter. Both the eligibility queryset and the truncating update now chain `rewrite(False)`. No data was at risk: the S3 objects were written from the correct source and verified byte-identical against the surviving column. The existing coverage asserted on a `values_list("text")` that was itself rewritten, so it passed against the bug. It now reads through `rewrite(False)`, and a new case covers rows with an empty `text_original`. Also in this change: - Scope dry-run totals to `--limit`. The count was capped but the character sum was not, so a limited run reported more reclaimable characters than an unlimited one. - Upload comments concurrently behind a shared boto client. S3 has no multi-object PUT and the uploads are round-trip bound; each comment remains its own independently retrievable object. Measured 10.2s -> 2.3s for 32 comments at a concurrency of 8. - Use botocore's `standard` retry mode, which handles S3 throttling responses explicitly rather than relying on the looser `legacy` default. - Print per-batch progress with rate and ETA, so a multi-hour backfill is observable. The count this needs is skipped when no progress callback is supplied, keeping the cron path unchanged. Co-Authored-By: Claude Opus 5 --- .../commands/archive_bot_comment_texts.py | 76 ++++++++- comments/services/text_archive.py | 155 +++++++++++++----- tests/unit/test_comments/test_text_archive.py | 49 +++++- 3 files changed, 237 insertions(+), 43 deletions(-) diff --git a/comments/management/commands/archive_bot_comment_texts.py b/comments/management/commands/archive_bot_comment_texts.py index 2b08c872bb..757bff77fe 100644 --- a/comments/management/commands/archive_bot_comment_texts.py +++ b/comments/management/commands/archive_bot_comment_texts.py @@ -1,3 +1,7 @@ +import time +from collections.abc import Callable + +from django.conf import settings from django.core.management.base import BaseCommand, CommandError from comments.services.text_archive import ( @@ -5,11 +9,27 @@ ARCHIVE_MIN_TEXT_LENGTH, ARCHIVE_STUB_LENGTH, DEFAULT_BATCH_SIZE, + DEFAULT_CONCURRENCY, + S3_KEY_PREFIX, + ArchiveStats, archive_bot_comment_texts, check_is_enabled, ) +def format_duration(seconds: float) -> str: + seconds = int(seconds) + hours, remainder = divmod(seconds, 3600) + minutes, seconds = divmod(remainder, 60) + + if hours: + return f"{hours}h{minutes:02d}m" + if minutes: + return f"{minutes}m{seconds:02d}s" + + return f"{seconds}s" + + class Command(BaseCommand): help = ( "Moves the full text of private bot comments older than " @@ -37,6 +57,36 @@ def add_arguments(self, parser): default=DEFAULT_BATCH_SIZE, help=f"Comments per database update (default: {DEFAULT_BATCH_SIZE})", ) + parser.add_argument( + "--concurrency", + type=int, + default=DEFAULT_CONCURRENCY, + help=( + "Uploads to keep in flight at once. S3 has no multi-object PUT, " + "so this is what makes a large backfill finish in minutes " + f"rather than hours (default: {DEFAULT_CONCURRENCY})" + ), + ) + + def _write_progress(self, stats, started: float): + elapsed = time.monotonic() - started + done = stats.archived + stats.failed + stats.skipped + rate = done / elapsed if elapsed else 0 + percent = (done / stats.total * 100) if stats.total else 0 + remaining = max(stats.total - done, 0) + eta = format_duration(remaining / rate) if rate else "?" + + line = ( + f" {done:,}/{stats.total:,} ({percent:.1f}%) " + f"{stats.chars_reclaimed:,} chars reclaimed " + f"{rate:.1f}/s elapsed {format_duration(elapsed)} eta {eta}" + ) + + if stats.failed or stats.skipped: + line += f" [{stats.failed} failed, {stats.skipped} skipped]" + + self.stdout.write(line) + self.stdout.flush() def handle(self, *args, **options): dry_run = options["dry_run"] @@ -47,16 +97,38 @@ def handle(self, *args, **options): "comment text archiving is disabled." ) + started = time.monotonic() + on_progress: Callable[[ArchiveStats], None] | None = None + + if not dry_run: + self.stdout.write( + f"Archiving to {settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT}/" + f"{S3_KEY_PREFIX}/ with concurrency {options['concurrency']}, " + f"batches of {options['batch_size']}" + ) + self.stdout.write("Counting eligible comments...") + self.stdout.flush() + + def write_progress(stats: ArchiveStats) -> None: + self._write_progress(stats, started) + + on_progress = write_progress + stats = archive_bot_comment_texts( dry_run=dry_run, limit=options["limit"], batch_size=options["batch_size"], + concurrency=options["concurrency"], + on_progress=on_progress, ) verb = "Would archive" if dry_run else "Archived" + elapsed = ( + "" if dry_run else f" in {format_duration(time.monotonic() - started)}" + ) self.stdout.write( - f"{verb} {stats.archived} comment(s), " - f"reclaiming {stats.chars_reclaimed:,} characters" + f"{verb} {stats.archived:,} comment(s), " + f"reclaiming {stats.chars_reclaimed:,} characters{elapsed}" ) if stats.sample_ids: diff --git a/comments/services/text_archive.py b/comments/services/text_archive.py index 760835f10a..acfedc9006 100644 --- a/comments/services/text_archive.py +++ b/comments/services/text_archive.py @@ -1,8 +1,11 @@ import json import logging +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from datetime import timedelta +from botocore.config import Config from django.conf import settings from django.core.serializers.json import DjangoJSONEncoder from django.db.models import Count, Q, QuerySet, Sum, Value @@ -26,6 +29,10 @@ S3_KEY_PREFIX = "comments_text" DEFAULT_BATCH_SIZE = 500 +# S3 has no multi-object PUT, so the only way to cut the wall-clock cost of the +# uploads is to keep several of them in flight at once. They are latency bound, +# not bandwidth bound, so this scales close to linearly. +DEFAULT_CONCURRENCY = 8 # `text` is the base column shadowed by modeltranslation: it holds a duplicate # of the original content that is written on save but never read back (reads of @@ -49,7 +56,28 @@ def build_key(comment_id: int) -> str: return f"{S3_KEY_PREFIX}/{comment_id}.json" -def upload_text(comment_id: int, text: str) -> str: +def get_archive_s3_client(concurrency: int = 1): + """ + S3 client for the archive. Building a client is expensive, so callers that + upload many objects should build one and pass it around. The connection + pool has to be at least as large as the number of concurrent uploads, or + botocore serialises them behind the default pool of 10. + """ + + return get_boto_client( + "s3", + config=Config( + max_pool_connections=max(concurrency, 10), + # S3 answers a request rate it cannot sustain with 503 SlowDown. + # We run far below the limit, but `standard` mode covers the + # throttling error codes explicitly and backs off with jitter, + # rather than relying on the looser `legacy` default. + retries={"mode": "standard", "max_attempts": 5}, + ), + ) + + +def upload_text(comment_id: int, text: str, s3=None) -> str: """ Uploads the full original text of a comment to S3 and returns the key. @@ -58,7 +86,7 @@ def upload_text(comment_id: int, text: str) -> str: machine translations of an archived text would be pointless anyway. """ - s3 = get_boto_client("s3") + s3 = s3 or get_archive_s3_client() key = build_key(comment_id) s3.put_object( @@ -122,7 +150,13 @@ def get_archivable_comments() -> QuerySet[Comment]: cutoff = timezone.now() - timedelta(days=ARCHIVE_AGE_DAYS) return ( - Comment.objects.filter( + # `rewrite(False)` is essential, not an optimisation. Comment is + # registered with modeltranslation, whose queryset rewrites every + # mention of `text` into the current language's column. Without it, + # `Length(ORIGINAL_TEXT)` degrades to measuring `text_original` twice + # and rows whose text only lives in the base column are never seen. + Comment.objects.rewrite(False) + .filter( author__is_bot=True, is_private=True, is_text_archived=False, @@ -135,6 +169,10 @@ def get_archivable_comments() -> QuerySet[Comment]: @dataclass class ArchiveStats: + # Number of comments the run expects to process. Only populated when a + # progress callback asks for it, since counting means measuring the length + # of every candidate text. + total: int = 0 archived: int = 0 failed: int = 0 skipped: int = 0 @@ -164,22 +202,30 @@ def archive_bot_comment_texts( dry_run: bool = False, limit: int | None = None, batch_size: int = DEFAULT_BATCH_SIZE, + concurrency: int = DEFAULT_CONCURRENCY, + on_progress: Callable[[ArchiveStats], None] | None = None, ) -> ArchiveStats: """ Moves the full text of long, private, old bot comments to S3, leaving a truncated stub in the database. + + `on_progress` is called with the running stats after every batch. """ stats = ArchiveStats() queryset = get_archivable_comments() if dry_run: - # Aggregate without transferring any text - totals = queryset.aggregate(count=Count("id"), chars=Sum("text_length")) - count = totals["count"] or 0 + # Aggregate without transferring any text. The limit has to be applied + # before aggregating, so that the reported totals describe the rows the + # real run would actually touch. + scoped = queryset.order_by("id") if limit is not None: - count = min(count, limit) + scoped = scoped[:limit] + + totals = scoped.aggregate(count=Count("id"), chars=Sum("text_length")) + count = totals["count"] or 0 stats.archived = count stats.chars_reclaimed = max( @@ -191,9 +237,20 @@ def archive_bot_comment_texts( return stats + if on_progress is not None: + # Counting is not free: the eligibility filter measures the length of + # every candidate text, so this reads the whole candidate set + total = queryset.count() + stats.total = min(total, limit) if limit is not None else total + started_at = timezone.now() truncate_kwargs = _build_truncate_kwargs() cursor = 0 + concurrency = max(concurrency, 1) + # One client, shared by every worker: botocore clients are safe to call + # from multiple threads once built, and building one per upload is pure + # overhead + s3 = get_archive_s3_client(concurrency) while limit is None or stats.archived + stats.failed < limit: page_size = batch_size @@ -215,38 +272,56 @@ def archive_bot_comment_texts( cursor = rows[-1]["id"] uploaded_ids = [] - for row in rows: - text = row["text_original"] or row["text"] - - try: - upload_text(row["id"], text) - except Exception: - logger.exception("Failed to archive text of comment %s", row["id"]) - stats.failed += 1 - - continue - - uploaded_ids.append(row["id"]) - - if not uploaded_ids: - continue - - # Only truncate rows that have not been touched since the run began, so - # an edit racing the upload can never lose text. `edited_at` is - # nullable on rows that predate TimeStampedModel.save. - untouched = Comment.objects.filter(pk__in=uploaded_ids).filter( - Q(edited_at__lt=started_at) | Q(edited_at__isnull=True) - ) - archived_ids = set(untouched.values_list("id", flat=True)) - updated = untouched.update(**truncate_kwargs) - - stats.archived += updated - stats.skipped += len(uploaded_ids) - updated - stats.chars_reclaimed += sum( - max(row["text_length"] - ARCHIVE_STUB_LENGTH, 0) - for row in rows - if row["id"] in archived_ids - ) - stats.sample_ids = (stats.sample_ids + sorted(archived_ids))[:5] + # Each comment is still its own independently retrievable object; the + # requests are simply issued in parallel, since they are round-trip + # bound. The database update below waits for the whole page, so an + # upload can never be outrun by its own truncation. + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = { + pool.submit( + upload_text, row["id"], row["text_original"] or row["text"], s3 + ): row["id"] + for row in rows + } + + for future, comment_id in futures.items(): + try: + future.result() + except Exception: + logger.exception("Failed to archive text of comment %s", comment_id) + stats.failed += 1 + + continue + + uploaded_ids.append(comment_id) + + if uploaded_ids: + # Only truncate rows that have not been touched since the run + # began, so an edit racing the upload can never lose text. + # `edited_at` is nullable on rows that predate + # TimeStampedModel.save. + # `rewrite(False)` again: modeltranslation's `update()` rewrites + # the `text` kwarg to `text_original`, which collides with the + # `text_original` kwarg and leaves the base column holding the + # full text — silently forfeiting half the space this reclaims. + untouched = ( + Comment.objects.rewrite(False) + .filter(pk__in=uploaded_ids) + .filter(Q(edited_at__lt=started_at) | Q(edited_at__isnull=True)) + ) + archived_ids = set(untouched.values_list("id", flat=True)) + updated = untouched.update(**truncate_kwargs) + + stats.archived += updated + stats.skipped += len(uploaded_ids) - updated + stats.chars_reclaimed += sum( + max(row["text_length"] - ARCHIVE_STUB_LENGTH, 0) + for row in rows + if row["id"] in archived_ids + ) + stats.sample_ids = (stats.sample_ids + sorted(archived_ids))[:5] + + if on_progress is not None: + on_progress(stats) return stats diff --git a/tests/unit/test_comments/test_text_archive.py b/tests/unit/test_comments/test_text_archive.py index 7e8a214f51..cdc876bf3d 100644 --- a/tests/unit/test_comments/test_text_archive.py +++ b/tests/unit/test_comments/test_text_archive.py @@ -145,8 +145,14 @@ def test_archives_text_to_s3_and_truncates_row(self, bot, post, s3_stub): comment.refresh_from_db() assert comment.is_text_archived is True assert comment.text_original == LONG_TEXT[:ARCHIVE_STUB_LENGTH] + # `rewrite(False)` is what makes this assertion meaningful: a plain + # `values_list("text")` is rewritten by modeltranslation to read + # `text_original`, so it would pass even if the base column still + # held the full text. assert ( - Comment.objects.filter(pk=comment.pk).values_list("text", flat=True)[0] + Comment.objects.rewrite(False) + .filter(pk=comment.pk) + .values_list("text", flat=True)[0] == LONG_TEXT[:ARCHIVE_STUB_LENGTH] ) @@ -187,6 +193,22 @@ def test_dry_run_writes_nothing(self, bot, post, s3_stub): assert comment.is_text_archived is False assert comment.text_original == LONG_TEXT + def test_dry_run_totals_are_scoped_to_the_limit(self, bot, post, s3_stub): + for _ in range(3): + factory_archivable_comment(bot, post) + + per_comment = len(LONG_TEXT) - ARCHIVE_STUB_LENGTH + unlimited = archive_bot_comment_texts(dry_run=True) + limited = archive_bot_comment_texts(dry_run=True, limit=1) + + assert unlimited.archived == 3 + assert unlimited.chars_reclaimed == 3 * per_comment + + # The limited estimate must describe only the rows a real run would + # touch, not the whole queryset + assert limited.archived == 1 + assert limited.chars_reclaimed == per_comment + def test_upload_failure_leaves_comment_intact(self, bot, post, s3_stub, mocker): comment = factory_archivable_comment(bot, post) mocker.patch( @@ -210,6 +232,31 @@ def test_respects_limit(self, bot, post, s3_stub): assert archive_bot_comment_texts(limit=1).archived == 1 assert get_archivable_comments().count() == 1 + def test_archives_comments_whose_text_is_only_in_the_base_column( + self, bot, post, s3_stub + ): + """ + Rows written before modeltranslation was introduced have an empty + `text_original`. They are the largest rows in the table, so they must + not fall through the eligibility filter. + """ + + comment = factory_archivable_comment(bot, post) + Comment.objects.rewrite(False).filter(pk=comment.pk).update( + text=LONG_TEXT, text_original="" + ) + + assert archive_bot_comment_texts().archived == 1 + + payload = json.loads(s3_stub[build_key(comment.pk)]) + assert payload["text"] == LONG_TEXT + assert ( + Comment.objects.rewrite(False) + .filter(pk=comment.pk) + .values_list("text", flat=True)[0] + == LONG_TEXT[:ARCHIVE_STUB_LENGTH] + ) + def test_processes_multiple_batches(self, bot, post, s3_stub): for _ in range(5): factory_archivable_comment(bot, post) From 93a5268d445e1ec17ea55f4c584b31ba55a7ac2e Mon Sep 17 00:00:00 2001 From: lsabor Date: Thu, 20 Aug 2026 11:06:55 -0700 Subject: [PATCH 3/6] adds syncing command --- comments/management/commands/_progress.py | 62 ++++++ .../commands/archive_bot_comment_texts.py | 64 ++---- .../commands/sync_archived_comment_texts.py | 156 +++++++++++++++ comments/services/text_archive.py | 186 +++++++++++++++++- 4 files changed, 422 insertions(+), 46 deletions(-) create mode 100644 comments/management/commands/_progress.py create mode 100644 comments/management/commands/sync_archived_comment_texts.py diff --git a/comments/management/commands/_progress.py b/comments/management/commands/_progress.py new file mode 100644 index 0000000000..7f623c82ad --- /dev/null +++ b/comments/management/commands/_progress.py @@ -0,0 +1,62 @@ +""" +Progress reporting shared by the comment text archive commands. + +Django's command discovery skips modules whose name starts with an +underscore, so this sits alongside the commands without becoming one. +""" + +import time + + +def format_duration(seconds: float) -> str: + seconds = int(seconds) + hours, remainder = divmod(seconds, 3600) + minutes, seconds = divmod(remainder, 60) + + if hours: + return f"{hours}h{minutes:02d}m" + if minutes: + return f"{minutes}m{seconds:02d}s" + + return f"{seconds}s" + + +class ProgressWriter: + """ + Prints a running one-line summary with a rate and an ETA. + + These commands process hundreds of thousands of rows over hours, so the + point is to make a long run observable rather than to look pretty. Output + is one line per batch, not a redrawn line, so it survives being piped to + a log file. + """ + + def __init__(self, stdout, total: int = 0): + self.stdout = stdout + self.total = total + self.started = time.monotonic() + + @property + def elapsed(self) -> float: + return time.monotonic() - self.started + + def write(self, line: str) -> None: + self.stdout.write(line) + self.stdout.flush() + + def update(self, done: int, summary: str, detail: str = "") -> None: + elapsed = self.elapsed + rate = done / elapsed if elapsed else 0 + percent = (done / self.total * 100) if self.total else 0 + remaining = max(self.total - done, 0) + eta = format_duration(remaining / rate) if rate else "?" + + line = ( + f" {done:,}/{self.total:,} ({percent:.1f}%) {summary} " + f"{rate:.1f}/s elapsed {format_duration(elapsed)} eta {eta}" + ) + + if detail: + line += f" [{detail}]" + + self.write(line) diff --git a/comments/management/commands/archive_bot_comment_texts.py b/comments/management/commands/archive_bot_comment_texts.py index 757bff77fe..1808f8607c 100644 --- a/comments/management/commands/archive_bot_comment_texts.py +++ b/comments/management/commands/archive_bot_comment_texts.py @@ -1,4 +1,3 @@ -import time from collections.abc import Callable from django.conf import settings @@ -16,18 +15,7 @@ check_is_enabled, ) - -def format_duration(seconds: float) -> str: - seconds = int(seconds) - hours, remainder = divmod(seconds, 3600) - minutes, seconds = divmod(remainder, 60) - - if hours: - return f"{hours}h{minutes:02d}m" - if minutes: - return f"{minutes}m{seconds:02d}s" - - return f"{seconds}s" +from ._progress import ProgressWriter, format_duration class Command(BaseCommand): @@ -68,26 +56,6 @@ def add_arguments(self, parser): ), ) - def _write_progress(self, stats, started: float): - elapsed = time.monotonic() - started - done = stats.archived + stats.failed + stats.skipped - rate = done / elapsed if elapsed else 0 - percent = (done / stats.total * 100) if stats.total else 0 - remaining = max(stats.total - done, 0) - eta = format_duration(remaining / rate) if rate else "?" - - line = ( - f" {done:,}/{stats.total:,} ({percent:.1f}%) " - f"{stats.chars_reclaimed:,} chars reclaimed " - f"{rate:.1f}/s elapsed {format_duration(elapsed)} eta {eta}" - ) - - if stats.failed or stats.skipped: - line += f" [{stats.failed} failed, {stats.skipped} skipped]" - - self.stdout.write(line) - self.stdout.flush() - def handle(self, *args, **options): dry_run = options["dry_run"] @@ -97,20 +65,32 @@ def handle(self, *args, **options): "comment text archiving is disabled." ) - started = time.monotonic() + progress = ProgressWriter(self.stdout) on_progress: Callable[[ArchiveStats], None] | None = None if not dry_run: - self.stdout.write( + progress.write( f"Archiving to {settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT}/" f"{S3_KEY_PREFIX}/ with concurrency {options['concurrency']}, " f"batches of {options['batch_size']}" ) - self.stdout.write("Counting eligible comments...") - self.stdout.flush() + progress.write("Counting eligible comments...") def write_progress(stats: ArchiveStats) -> None: - self._write_progress(stats, started) + progress.total = stats.total + detail = ", ".join( + f"{count} {label}" + for label, count in ( + ("failed", stats.failed), + ("skipped", stats.skipped), + ) + if count + ) + progress.update( + stats.archived + stats.failed + stats.skipped, + f"{stats.chars_reclaimed:,} chars reclaimed", + detail, + ) on_progress = write_progress @@ -123,17 +103,15 @@ def write_progress(stats: ArchiveStats) -> None: ) verb = "Would archive" if dry_run else "Archived" - elapsed = ( - "" if dry_run else f" in {format_duration(time.monotonic() - started)}" - ) - self.stdout.write( + elapsed = "" if dry_run else f" in {format_duration(progress.elapsed)}" + progress.write( f"{verb} {stats.archived:,} comment(s), " f"reclaiming {stats.chars_reclaimed:,} characters{elapsed}" ) if stats.sample_ids: sample = ", ".join(str(pk) for pk in stats.sample_ids) - self.stdout.write(f"Sample comment ids: {sample}") + progress.write(f"Sample comment ids: {sample}") if stats.skipped: self.stdout.write( diff --git a/comments/management/commands/sync_archived_comment_texts.py b/comments/management/commands/sync_archived_comment_texts.py new file mode 100644 index 0000000000..a7c47810f6 --- /dev/null +++ b/comments/management/commands/sync_archived_comment_texts.py @@ -0,0 +1,156 @@ +from django.conf import settings +from django.core.management.base import BaseCommand, CommandError +from django.utils.dateparse import parse_datetime +from django.utils.timezone import is_naive, make_aware + +from comments.services.text_archive import ( + DEFAULT_BATCH_SIZE, + DEFAULT_CONCURRENCY, + S3_KEY_PREFIX, + SyncStats, + check_is_enabled, + sync_archived_comment_texts, +) + +from ._progress import ProgressWriter, format_duration + + +class Command(BaseCommand): + help = ( + "Truncates comments whose full text is already in the S3 archive, " + "without uploading anything. This is the second half of a one-off " + "migration: `archive_bot_comment_texts` is run once against a copy of " + "the database to populate the bucket, then this brings the real " + "database in line with it. Afterwards the monthly cron job takes over." + ) + + def add_arguments(self, parser): + parser.add_argument( + "--snapshot-at", + required=True, + help=( + "When the database copy used for the uploads was taken, as an " + "ISO-8601 timestamp (e.g. 2026-08-20T17:00:00Z). Rows created " + "or touched after this are left alone, because the archived " + "copy of their text may be stale. Required: there is no safe " + "default." + ), + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Report what would be truncated without writing to the database", + ) + parser.add_argument( + "--verify", + action="store_true", + help=( + "Re-read every archived object and require it to match the row " + "before truncating. Much slower, and downloads the whole " + "archive, but does not rely on the timestamp guards alone" + ), + ) + parser.add_argument( + "--batch-size", + type=int, + default=DEFAULT_BATCH_SIZE, + help=f"Comments per database update (default: {DEFAULT_BATCH_SIZE})", + ) + parser.add_argument( + "--concurrency", + type=int, + default=DEFAULT_CONCURRENCY, + help=( + "Downloads to keep in flight at once, for --verify " + f"(default: {DEFAULT_CONCURRENCY})" + ), + ) + + def handle(self, *args, **options): + if not check_is_enabled(): + raise CommandError( + "AWS_STORAGE_BUCKET_COMMENTS_TEXT is not configured, " + "comment text archiving is disabled." + ) + + snapshot_at = parse_datetime(options["snapshot_at"]) + + if snapshot_at is None: + raise CommandError( + f"Could not parse --snapshot-at {options['snapshot_at']!r} as an " + "ISO-8601 timestamp." + ) + + if is_naive(snapshot_at): + # A naive timestamp here would be compared against tz-aware columns + # and blow up mid-run, after an unknown number of rows + snapshot_at = make_aware(snapshot_at) + + dry_run = options["dry_run"] + progress = ProgressWriter(self.stdout) + + progress.write( + f"Syncing against {settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT}/" + f"{S3_KEY_PREFIX}/ as of {snapshot_at.isoformat()}" + + (" (verifying every object)" if options["verify"] else "") + ) + progress.write("Listing the archive...") + + def on_progress(stats: SyncStats) -> None: + progress.total = stats.total + done = ( + stats.synced + + stats.already_archived + + stats.orphaned + + stats.skipped_stale + + stats.mismatched + ) + detail = ", ".join( + f"{count} {label}" + for label, count in ( + ("already archived", stats.already_archived), + ("orphaned", stats.orphaned), + ("stale", stats.skipped_stale), + ("mismatched", stats.mismatched), + ) + if count + ) + progress.update(done, f"{stats.chars_reclaimed:,} chars reclaimed", detail) + + stats = sync_archived_comment_texts( + snapshot_at=snapshot_at, + dry_run=dry_run, + verify=options["verify"], + batch_size=options["batch_size"], + concurrency=options["concurrency"], + on_progress=on_progress, + ) + + verb = "Would sync" if dry_run else "Synced" + progress.write( + f"\n{verb} {stats.synced:,} of {stats.total:,} archived object(s), " + f"reclaiming {stats.chars_reclaimed:,} characters " + f"in {format_duration(progress.elapsed)}" + ) + + for label, count in ( + ("already truncated", stats.already_archived), + ("orphaned (no such comment)", stats.orphaned), + ("skipped as touched since the snapshot", stats.skipped_stale), + ): + if count: + progress.write(f" {count:,} {label}") + + if stats.sample_ids: + sample = ", ".join(str(pk) for pk in stats.sample_ids) + progress.write(f"Sample comment ids: {sample}") + + if stats.mismatched: + # Not fatal: these keep their text and the monthly job re-archives + # them, but a large number means the snapshot is not what we think + self.stdout.write( + self.style.WARNING( + f"{stats.mismatched:,} archived object(s) did not match the " + "current text and were left alone" + ) + ) diff --git a/comments/services/text_archive.py b/comments/services/text_archive.py index acfedc9006..d3ada23045 100644 --- a/comments/services/text_archive.py +++ b/comments/services/text_archive.py @@ -3,7 +3,7 @@ from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field -from datetime import timedelta +from datetime import datetime, timedelta from botocore.config import Config from django.conf import settings @@ -106,13 +106,13 @@ def upload_text(comment_id: int, text: str, s3=None) -> str: return key -def fetch_text(comment_id: int) -> str | None: +def fetch_text(comment_id: int, s3=None) -> str | None: """ Reads the archived full text of a comment back from S3. Returns None if the object is missing. """ - s3 = get_boto_client("s3") + s3 = s3 or get_archive_s3_client() try: obj = s3.get_object( @@ -325,3 +325,183 @@ def archive_bot_comment_texts( on_progress(stats) return stats + + +def list_archived_comment_ids( + on_progress: Callable[[int], None] | None = None, +) -> set[int]: + """ + Every comment id that already has an object in the archive, read straight + from the bucket. + + The bucket, not the database, is the authority on what has been archived: + the point of the sync below is to reconcile a database that knows nothing + about uploads performed elsewhere. + """ + + s3 = get_archive_s3_client() + prefix = f"{S3_KEY_PREFIX}/" + comment_ids = set() + + for page in s3.get_paginator("list_objects_v2").paginate( + Bucket=settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT, Prefix=prefix + ): + for obj in page.get("Contents", []): + stem = obj["Key"][len(prefix) :].removesuffix(".json") + + if stem.isdigit(): + comment_ids.add(int(stem)) + + if on_progress is not None: + on_progress(len(comment_ids)) + + return comment_ids + + +@dataclass +class SyncStats: + # Objects found in the bucket + total: int = 0 + synced: int = 0 + # Present in the bucket, but the row is already truncated + already_archived: int = 0 + # Present in the bucket with no matching row: deleted since the upload + orphaned: int = 0 + # Touched since the snapshot, so the archived copy may be stale + skipped_stale: int = 0 + # `--verify` only: the archived text no longer matches the row + mismatched: int = 0 + chars_reclaimed: int = 0 + sample_ids: list[int] = field(default_factory=list) + + +def get_syncable_comments(comment_ids, snapshot_at: datetime) -> QuerySet[Comment]: + """ + Rows that may be truncated against an archive uploaded elsewhere. + + Everything here is a guard against the archived copy being stale. The + upload happened against a database snapshot taken at `snapshot_at`; any + row created or touched since then may have text the archive does not + have, so it is left alone. Skipping costs nothing — the monthly job + re-archives it properly — while truncating it would destroy the edit. + """ + + return ( + Comment.objects.rewrite(False) + .filter( + pk__in=comment_ids, + is_text_archived=False, + created_at__lt=snapshot_at, + ) + # `edited_at` is bumped by every save; `text_edited_at` only by an + # edit to the text. Both are nullable on rows that predate them, and + # either one moving past the snapshot disqualifies the row. + .filter(Q(edited_at__lt=snapshot_at) | Q(edited_at__isnull=True)) + .filter(Q(text_edited_at__lt=snapshot_at) | Q(text_edited_at__isnull=True)) + .annotate(text_length=Length(ORIGINAL_TEXT)) + .filter(text_length__gt=ARCHIVE_STUB_LENGTH) + ) + + +def _verify_archived_text(rows, concurrency: int) -> set[int]: + """ + Ids whose archived object still matches the row's text exactly. + """ + + s3 = get_archive_s3_client(concurrency) + verified = set() + + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = {pool.submit(fetch_text, row["id"], s3): row for row in rows} + + for future, row in futures.items(): + try: + archived = future.result() + except Exception: + logger.exception( + "Failed to read archived text of comment %s", row["id"] + ) + + continue + + if archived is not None and archived == ( + row["text_original"] or row["text"] + ): + verified.add(row["id"]) + + return verified + + +def sync_archived_comment_texts( + snapshot_at: datetime, + dry_run: bool = False, + verify: bool = False, + batch_size: int = DEFAULT_BATCH_SIZE, + concurrency: int = DEFAULT_CONCURRENCY, + on_progress: Callable[[SyncStats], None] | None = None, +) -> SyncStats: + """ + Truncates rows whose text is already in the archive, uploading nothing. + + This exists for one migration. The uploads are slow and bandwidth-heavy, + so they are performed once against a copy of the database; this then + brings the real database in line with the bucket without moving the text + a second time. + + `verify` re-reads every object and requires it to match the row before + truncating. That is the safe-but-slow path; without it the timestamp + guards in `get_syncable_comments` are what stand between a stale archive + and a lost edit. + """ + + stats = SyncStats() + archived_ids = sorted(list_archived_comment_ids()) + stats.total = len(archived_ids) + + truncate_kwargs = _build_truncate_kwargs() + columns = ["id", "text_length"] + (["text", "text_original"] if verify else []) + + for start in range(0, len(archived_ids), batch_size): + chunk = archived_ids[start : start + batch_size] + + # Two queries per chunk so the accounting is exact: what the database + # knows about these ids, then which of them are safe to truncate. + states = dict( + Comment.objects.rewrite(False) + .filter(pk__in=chunk) + .values_list("id", "is_text_archived") + ) + rows = list(get_syncable_comments(chunk, snapshot_at).values(*columns)) + + stats.orphaned += len(chunk) - len(states) + already = sum(1 for archived in states.values() if archived) + stats.already_archived += already + stats.skipped_stale += len(states) - already - len(rows) + + if verify and rows: + verified = _verify_archived_text(rows, concurrency) + stats.mismatched += len(rows) - len(verified) + rows = [row for row in rows if row["id"] in verified] + + if rows and not dry_run: + # Re-apply the guards at write time: a row edited between the + # select above and this update must not be truncated. + untouched = get_syncable_comments([row["id"] for row in rows], snapshot_at) + synced_ids = set(untouched.values_list("id", flat=True)) + updated = untouched.update(**truncate_kwargs) + + stats.synced += updated + stats.skipped_stale += len(rows) - updated + rows = [row for row in rows if row["id"] in synced_ids] + elif rows: + stats.synced += len(rows) + + stats.chars_reclaimed += sum( + max(row["text_length"] - ARCHIVE_STUB_LENGTH, 0) for row in rows + ) + stats.sample_ids = (stats.sample_ids + sorted(row["id"] for row in rows))[:5] + + if on_progress is not None: + on_progress(stats) + + return stats From 7af60ece4a0c684fafdbefa4472a306e86e86290 Mon Sep 17 00:00:00 2001 From: lsabor Date: Thu, 20 Aug 2026 11:17:02 -0700 Subject: [PATCH 4/6] test: cover syncing comment texts against an existing archive Exercises the guards that make the one-off migration safe rather than just the happy path: rows edited, text-edited, or created since the snapshot must be left alone, since the archived copy of their text may predate the change. Also covers reading ids back out of the bucket, the `--verify` path accepting a matching object and rejecting a diverged one, orphaned objects with no comment, idempotency, and that the sync uploads nothing and does not bump `edited_at`. The S3 stub grows a `list_objects_v2` paginator to support this. Co-Authored-By: Claude Opus 5 --- tests/unit/test_comments/test_text_archive.py | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) diff --git a/tests/unit/test_comments/test_text_archive.py b/tests/unit/test_comments/test_text_archive.py index cdc876bf3d..068ad62dd1 100644 --- a/tests/unit/test_comments/test_text_archive.py +++ b/tests/unit/test_comments/test_text_archive.py @@ -17,6 +17,9 @@ archive_bot_comment_texts, build_key, get_archivable_comments, + list_archived_comment_ids, + sync_archived_comment_texts, + upload_text, ) from posts.models import Post from projects.permissions import ObjectPermission @@ -80,6 +83,23 @@ def get_object(self, Bucket, Key): return {"Body": mocker.Mock(read=lambda: objects[Key].encode("utf-8"))} + def get_paginator(self, operation_name): + assert operation_name == "list_objects_v2" + + class Paginator: + def paginate(self, Bucket, Prefix): + # One page is enough here; the real paginator's chunking + # is botocore's concern, not ours + yield { + "Contents": [ + {"Key": key} + for key in sorted(objects) + if key.startswith(Prefix) + ] + } + + return Paginator() + mocker.patch( "comments.services.text_archive.get_boto_client", return_value=Client() ) @@ -375,3 +395,184 @@ def test_missing_archive_object_returns_404( ) assert response.status_code == 404 + + +@pytest.fixture() +def archived_elsewhere(bot, post, s3_stub): + """ + A comment whose text is in the archive while the row still holds it in + full: the state the production database is in after the uploads have been + performed against a copy of it. + """ + + comment = factory_archivable_comment(bot, post) + upload_text(comment.pk, LONG_TEXT) + + return comment + + +class TestListArchivedCommentIds: + def test_reads_ids_from_the_bucket(self, bot, post, s3_stub): + comments = [factory_archivable_comment(bot, post) for _ in range(3)] + for comment in comments: + upload_text(comment.pk, LONG_TEXT) + + assert list_archived_comment_ids() == {c.pk for c in comments} + + def test_ignores_keys_that_are_not_comment_ids(self, bot, post, s3_stub): + comment = factory_archivable_comment(bot, post) + upload_text(comment.pk, LONG_TEXT) + s3_stub["comments_text/not-an-id.json"] = "{}" + + assert list_archived_comment_ids() == {comment.pk} + + +class TestSyncArchivedCommentTexts: + def test_truncates_without_uploading(self, archived_elsewhere, s3_stub): + before = dict(s3_stub) + + stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + + assert stats.synced == 1 + assert stats.chars_reclaimed == len(LONG_TEXT) - ARCHIVE_STUB_LENGTH + # Nothing was written back to the archive + assert s3_stub == before + + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.is_text_archived is True + assert archived_elsewhere.text_original == LONG_TEXT[:ARCHIVE_STUB_LENGTH] + assert ( + Comment.objects.rewrite(False) + .filter(pk=archived_elsewhere.pk) + .values_list("text", flat=True)[0] + == LONG_TEXT[:ARCHIVE_STUB_LENGTH] + ) + + def test_does_not_bump_edited_at(self, archived_elsewhere, s3_stub): + edited_at = archived_elsewhere.edited_at + + sync_archived_comment_texts(snapshot_at=timezone.now()) + + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.edited_at == edited_at + + def test_dry_run_writes_nothing(self, archived_elsewhere, s3_stub): + stats = sync_archived_comment_texts(snapshot_at=timezone.now(), dry_run=True) + + assert stats.synced == 1 + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.is_text_archived is False + assert archived_elsewhere.text_original == LONG_TEXT + + def test_skips_rows_edited_since_the_snapshot(self, archived_elsewhere, s3_stub): + """ + The archived copy predates the edit, so truncating would lose it. + """ + + Comment.objects.rewrite(False).filter(pk=archived_elsewhere.pk).update( + edited_at=timezone.now() + ) + + stats = sync_archived_comment_texts( + snapshot_at=timezone.now() - timedelta(hours=1) + ) + + assert stats.synced == 0 + assert stats.skipped_stale == 1 + + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.is_text_archived is False + assert archived_elsewhere.text_original == LONG_TEXT + + def test_skips_rows_whose_text_was_edited_since_the_snapshot( + self, archived_elsewhere, s3_stub + ): + Comment.objects.rewrite(False).filter(pk=archived_elsewhere.pk).update( + text_edited_at=timezone.now(), edited_at=None + ) + + stats = sync_archived_comment_texts( + snapshot_at=timezone.now() - timedelta(hours=1) + ) + + assert stats.synced == 0 + assert stats.skipped_stale == 1 + + def test_skips_rows_created_since_the_snapshot(self, archived_elsewhere, s3_stub): + Comment.objects.rewrite(False).filter(pk=archived_elsewhere.pk).update( + created_at=timezone.now() + ) + + stats = sync_archived_comment_texts( + snapshot_at=timezone.now() - timedelta(hours=1) + ) + + assert stats.synced == 0 + assert stats.skipped_stale == 1 + + def test_counts_already_truncated_rows(self, archived_elsewhere, s3_stub): + sync_archived_comment_texts(snapshot_at=timezone.now()) + + stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + + assert stats.synced == 0 + assert stats.already_archived == 1 + + def test_counts_objects_with_no_comment(self, bot, post, s3_stub): + upload_text(999_999_999, LONG_TEXT) + + stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + + assert stats.synced == 0 + assert stats.orphaned == 1 + + def test_verify_skips_rows_whose_archive_no_longer_matches( + self, archived_elsewhere, s3_stub + ): + upload_text(archived_elsewhere.pk, "something else entirely") + + stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + + assert stats.synced == 0 + assert stats.mismatched == 1 + + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.text_original == LONG_TEXT + + def test_verify_accepts_a_matching_archive(self, archived_elsewhere, s3_stub): + stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + + assert stats.synced == 1 + assert stats.mismatched == 0 + + +class TestSyncCommand: + def test_syncs(self, archived_elsewhere, s3_stub): + out = StringIO() + + call_command( + "sync_archived_comment_texts", + "--snapshot-at", + timezone.now().isoformat(), + stdout=out, + ) + + assert "Synced 1 of 1 archived object(s)" in out.getvalue() + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.is_text_archived is True + + def test_requires_a_snapshot_timestamp(self, s3_stub): + with pytest.raises(CommandError): + call_command("sync_archived_comment_texts") + + def test_rejects_an_unparseable_snapshot_timestamp(self, s3_stub): + with pytest.raises(CommandError): + call_command("sync_archived_comment_texts", "--snapshot-at", "yesterday") + + def test_errors_when_bucket_is_not_configured(self, mocker): + mocker.patch("django.conf.settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT", None) + + with pytest.raises(CommandError): + call_command( + "sync_archived_comment_texts", "--snapshot-at", "2026-01-01T00:00:00Z" + ) From fbb971b0ae480579644278cde97e745c0b866b35 Mon Sep 17 00:00:00 2001 From: lsabor Date: Thu, 20 Aug 2026 12:46:32 -0700 Subject: [PATCH 5/6] fix: address review of the comment text archive - Give the monthly job a 30-minute time limit and cap it at one retry. Dramatiq's defaults are 10 minutes and 20 retries, so a run over the limit was killed and then replayed for hours. - Restore staff access to archived text. `get_comment_permission_for_user` resolves every private comment to no permission but the author's, so archiving otherwise left a bot's full text unreadable through every interface. The admin now renders it read-only from S3, and the full-text endpoint lets staff read any comment. - Re-assert the archiver's own invariants in the sync command: a stray key in the bucket must not be able to truncate a public or human comment. Split into `get_sync_candidates` (eligibility) and `get_syncable_comments` (freshness). - Annotate `original_text` instead of selecting both copies of the text, halving the working set of a batch. This needs an explicit `output_field` on `ORIGINAL_TEXT`: modeltranslation's `TranslationTextField` and `Value("")` only reconcile while the expression stays wrapped in `Length`/`Substr`. - Build one S3 client per sync run rather than one per verify batch. - Separate unreadable archived objects from genuine mismatches, and ineligible rows from stale ones, so the command's report says what actually happened. - Drop the dead `on_progress` parameter from `list_archived_comment_ids`. Co-Authored-By: Claude Opus 5 --- comments/admin.py | 41 +++- .../commands/sync_archived_comment_texts.py | 16 ++ comments/serializers/common.py | 3 + comments/services/text_archive.py | 132 ++++++++---- comments/tasks.py | 6 +- comments/views/common.py | 18 +- tests/unit/test_comments/test_text_archive.py | 189 +++++++++++++++++- 7 files changed, 347 insertions(+), 58 deletions(-) diff --git a/comments/admin.py b/comments/admin.py index c6056e3a51..c5f2017bee 100644 --- a/comments/admin.py +++ b/comments/admin.py @@ -2,7 +2,9 @@ from django.conf import settings from django.contrib import admin from django.contrib.postgres.search import SearchQuery +from django.utils.html import format_html +from comments.services.text_archive import get_full_text from utils.models import CustomTranslationAdmin, uniques_ordered_list from utils.translation import build_supported_localized_fieldname from .models import Comment, KeyFactor, KeyFactorDriver @@ -67,14 +69,49 @@ class CommentAdmin(CustomTranslationAdmin): def should_update_translations(self, obj): return not obj.on_post.is_private() + @admin.display(description="Archived text (read-only, fetched from S3)") + def archived_text(self, obj): + """ + The full text of an archived comment, read back from the archive. + + The admin is where staff investigate a comment, and the row itself + now holds nothing but a 200-character stub. Reading this costs an S3 + round trip per change-page load, which is why `get_fields` only adds + it for rows that are actually archived. + """ + + text = get_full_text(obj) + + if text is None: + return format_html( + "{}", + "The archived text could not be retrieved from S3. " + "Only the stub above remains in the database.", + ) + + return format_html( + '
{}
', + text, + ) + + def get_fields(self, request, obj=None): + fields = list(super().get_fields(request, obj)) + + if obj and obj.is_text_archived: + fields.append("archived_text") + + return uniques_ordered_list(fields) + def get_readonly_fields(self, request, obj=None): readonly_fields = list(super().get_readonly_fields(request, obj)) if obj and obj.is_text_archived: # Only a stub of the text is left in the db, so editing it here # would bypass the `update_comment` guard and leave the row out of - # sync with the archived original - readonly_fields += ["text"] + [ + # sync with the archived original. `archived_text` is not a model + # field at all, so it has to be declared read-only to appear. + readonly_fields += ["text", "archived_text"] + [ build_supported_localized_fieldname("text", lang) for lang, _label in settings.LANGUAGES ] diff --git a/comments/management/commands/sync_archived_comment_texts.py b/comments/management/commands/sync_archived_comment_texts.py index a7c47810f6..935139eae4 100644 --- a/comments/management/commands/sync_archived_comment_texts.py +++ b/comments/management/commands/sync_archived_comment_texts.py @@ -103,7 +103,9 @@ def on_progress(stats: SyncStats) -> None: + stats.already_archived + stats.orphaned + stats.skipped_stale + + stats.ineligible + stats.mismatched + + stats.verify_failed ) detail = ", ".join( f"{count} {label}" @@ -111,7 +113,9 @@ def on_progress(stats: SyncStats) -> None: ("already archived", stats.already_archived), ("orphaned", stats.orphaned), ("stale", stats.skipped_stale), + ("ineligible", stats.ineligible), ("mismatched", stats.mismatched), + ("unreadable", stats.verify_failed), ) if count ) @@ -137,6 +141,7 @@ def on_progress(stats: SyncStats) -> None: ("already truncated", stats.already_archived), ("orphaned (no such comment)", stats.orphaned), ("skipped as touched since the snapshot", stats.skipped_stale), + ("ineligible (not a long private bot comment)", stats.ineligible), ): if count: progress.write(f" {count:,} {label}") @@ -154,3 +159,14 @@ def on_progress(stats: SyncStats) -> None: "current text and were left alone" ) ) + + if stats.verify_failed: + # Distinct from a mismatch: nothing is known about these objects, + # so a non-zero count here means the bucket, not the snapshot, is + # what needs looking at + self.stdout.write( + self.style.ERROR( + f"{stats.verify_failed:,} archived object(s) could not be read " + "back and were left alone" + ) + ) diff --git a/comments/serializers/common.py b/comments/serializers/common.py index c3f334de8f..b7cf427e8c 100644 --- a/comments/serializers/common.py +++ b/comments/serializers/common.py @@ -76,6 +76,9 @@ class Meta: "text_edited_at", "is_soft_deleted", "text", + # TODO: consumed by the front end in a later commit, which will + # replace the stub in `text` with a "load full text" affordance + # backed by the `comment-full-text` endpoint "is_text_archived", "on_post", "on_post_data", diff --git a/comments/services/text_archive.py b/comments/services/text_archive.py index d3ada23045..b61af874dc 100644 --- a/comments/services/text_archive.py +++ b/comments/services/text_archive.py @@ -8,7 +8,7 @@ from botocore.config import Config from django.conf import settings from django.core.serializers.json import DjangoJSONEncoder -from django.db.models import Count, Q, QuerySet, Sum, Value +from django.db.models import Count, Q, QuerySet, Sum, TextField, Value from django.db.models.functions import Coalesce, Length, NullIf, Substr from django.utils import timezone @@ -39,7 +39,13 @@ # `comment.text` resolve to `text_original` through the translation # descriptor). `text_original` may be NULL or empty on rows that were never # saved through the descriptor, so fall back to the base column. -ORIGINAL_TEXT = Coalesce(NullIf("text_original", Value("")), "text") +# `output_field` is required, not decorative: `text_original` is a +# modeltranslation `TranslationTextField` and `Value("")` a `CharField`, which +# Django refuses to reconcile on its own as soon as the expression is selected +# rather than wrapped in `Length`/`Substr`. +ORIGINAL_TEXT = Coalesce( + NullIf("text_original", Value("")), "text", output_field=TextField() +) def check_is_enabled() -> bool: @@ -257,10 +263,14 @@ def archive_bot_comment_texts( if limit is not None: page_size = min(batch_size, limit - stats.archived - stats.failed) + # `original_text` is annotated rather than selecting both columns: + # they hold the same content, and a page of 500 comments that may run + # to 150k characters each is worth not loading twice. rows = list( queryset.filter(id__gt=cursor) .order_by("id") - .values("id", "text", "text_original", "text_length")[:page_size] + .annotate(original_text=ORIGINAL_TEXT) + .values("id", "original_text", "text_length")[:page_size] ) if not rows: @@ -278,9 +288,7 @@ def archive_bot_comment_texts( # upload can never be outrun by its own truncation. with ThreadPoolExecutor(max_workers=concurrency) as pool: futures = { - pool.submit( - upload_text, row["id"], row["text_original"] or row["text"], s3 - ): row["id"] + pool.submit(upload_text, row["id"], row["original_text"], s3): row["id"] for row in rows } @@ -327,19 +335,18 @@ def archive_bot_comment_texts( return stats -def list_archived_comment_ids( - on_progress: Callable[[int], None] | None = None, -) -> set[int]: +def list_archived_comment_ids(s3=None) -> set[int]: """ Every comment id that already has an object in the archive, read straight from the bucket. - The bucket, not the database, is the authority on what has been archived: - the point of the sync below is to reconcile a database that knows nothing - about uploads performed elsewhere. + The bucket is the authority on what has been uploaded: the point of the + sync below is to reconcile a database that knows nothing about uploads + performed elsewhere. It is not, however, the authority on what may be + truncated — see `get_sync_candidates`. """ - s3 = get_archive_s3_client() + s3 = s3 or get_archive_s3_client() prefix = f"{S3_KEY_PREFIX}/" comment_ids = set() @@ -352,9 +359,6 @@ def list_archived_comment_ids( if stem.isdigit(): comment_ids.add(int(stem)) - if on_progress is not None: - on_progress(len(comment_ids)) - return comment_ids @@ -369,47 +373,77 @@ class SyncStats: orphaned: int = 0 # Touched since the snapshot, so the archived copy may be stale skipped_stale: int = 0 + # A row the archiver would never have uploaded, or one already at or + # below the stub length: nothing to reclaim, and a hint that the bucket + # holds keys this command did not put there + ineligible: int = 0 # `--verify` only: the archived text no longer matches the row mismatched: int = 0 + # `--verify` only: the archived object could not be read back at all, + # which says nothing about whether it matches + verify_failed: int = 0 chars_reclaimed: int = 0 sample_ids: list[int] = field(default_factory=list) -def get_syncable_comments(comment_ids, snapshot_at: datetime) -> QuerySet[Comment]: +def get_sync_candidates(comment_ids) -> QuerySet[Comment]: """ - Rows that may be truncated against an archive uploaded elsewhere. + Rows this command is allowed to truncate at all, ignoring freshness. - Everything here is a guard against the archived copy being stale. The - upload happened against a database snapshot taken at `snapshot_at`; any - row created or touched since then may have text the archive does not - have, so it is left alone. Skipping costs nothing — the monthly job - re-archives it properly — while truncating it would destroy the edit. + Anything the archiver uploaded was a long, private bot comment, so those + invariants are re-asserted here rather than trusting the key alone: a + stray or mistyped object in the bucket must not be able to truncate a row + the archiver would never have touched. """ return ( Comment.objects.rewrite(False) .filter( pk__in=comment_ids, + author__is_bot=True, + is_private=True, is_text_archived=False, - created_at__lt=snapshot_at, ) + .annotate(text_length=Length(ORIGINAL_TEXT)) + .filter(text_length__gt=ARCHIVE_STUB_LENGTH) + ) + + +def get_syncable_comments(comment_ids, snapshot_at: datetime) -> QuerySet[Comment]: + """ + Candidates that are also safe to truncate against an archive uploaded + elsewhere. + + Everything added here is a guard against the archived copy being stale. + The upload happened against a database snapshot taken at `snapshot_at`; + any row created or touched since then may have text the archive does not + have, so it is left alone. Skipping costs nothing — the monthly job + re-archives it properly — while truncating it would destroy the edit. + """ + + return ( + get_sync_candidates(comment_ids) + .filter(created_at__lt=snapshot_at) # `edited_at` is bumped by every save; `text_edited_at` only by an # edit to the text. Both are nullable on rows that predate them, and # either one moving past the snapshot disqualifies the row. .filter(Q(edited_at__lt=snapshot_at) | Q(edited_at__isnull=True)) .filter(Q(text_edited_at__lt=snapshot_at) | Q(text_edited_at__isnull=True)) - .annotate(text_length=Length(ORIGINAL_TEXT)) - .filter(text_length__gt=ARCHIVE_STUB_LENGTH) ) -def _verify_archived_text(rows, concurrency: int) -> set[int]: +def _verify_archived_text(rows, s3, concurrency: int) -> tuple[set[int], set[int]]: """ - Ids whose archived object still matches the row's text exactly. + Ids whose archived object still matches the row's text exactly, and ids + whose object could not be read back at all. + + The two are kept apart because they mean different things: a mismatch is + a stale archive, an unreadable object is an S3 problem. Both leave the + row alone. """ - s3 = get_archive_s3_client(concurrency) verified = set() + unreadable = set() with ThreadPoolExecutor(max_workers=concurrency) as pool: futures = {pool.submit(fetch_text, row["id"], s3): row for row in rows} @@ -421,15 +455,17 @@ def _verify_archived_text(rows, concurrency: int) -> set[int]: logger.exception( "Failed to read archived text of comment %s", row["id"] ) + unreadable.add(row["id"]) continue - if archived is not None and archived == ( - row["text_original"] or row["text"] - ): + if archived is None: + # `fetch_text` swallows a missing object and logs it + unreadable.add(row["id"]) + elif archived == row["original_text"]: verified.add(row["id"]) - return verified + return verified, unreadable def sync_archived_comment_texts( @@ -455,32 +491,46 @@ def sync_archived_comment_texts( """ stats = SyncStats() - archived_ids = sorted(list_archived_comment_ids()) + concurrency = max(concurrency, 1) + # One client for the whole run, listing and verification alike: building + # one is expensive, and the verification below would otherwise build a + # fresh one for every batch. + s3 = get_archive_s3_client(concurrency) + archived_ids = sorted(list_archived_comment_ids(s3)) stats.total = len(archived_ids) truncate_kwargs = _build_truncate_kwargs() - columns = ["id", "text_length"] + (["text", "text_original"] if verify else []) + columns = ["id", "text_length"] + (["original_text"] if verify else []) for start in range(0, len(archived_ids), batch_size): chunk = archived_ids[start : start + batch_size] - # Two queries per chunk so the accounting is exact: what the database - # knows about these ids, then which of them are safe to truncate. + # Three queries per chunk so the accounting is exact: what the + # database knows about these ids, which of them this command may + # touch at all, and which of those are also fresh enough to truncate. states = dict( Comment.objects.rewrite(False) .filter(pk__in=chunk) .values_list("id", "is_text_archived") ) - rows = list(get_syncable_comments(chunk, snapshot_at).values(*columns)) + candidate_ids = set(get_sync_candidates(chunk).values_list("id", flat=True)) + syncable = get_syncable_comments(chunk, snapshot_at) + + if verify: + syncable = syncable.annotate(original_text=ORIGINAL_TEXT) + + rows = list(syncable.values(*columns)) stats.orphaned += len(chunk) - len(states) already = sum(1 for archived in states.values() if archived) stats.already_archived += already - stats.skipped_stale += len(states) - already - len(rows) + stats.ineligible += len(states) - already - len(candidate_ids) + stats.skipped_stale += len(candidate_ids) - len(rows) if verify and rows: - verified = _verify_archived_text(rows, concurrency) - stats.mismatched += len(rows) - len(verified) + verified, unreadable = _verify_archived_text(rows, s3, concurrency) + stats.mismatched += len(rows) - len(verified) - len(unreadable) + stats.verify_failed += len(unreadable) rows = [row for row in rows if row["id"] in verified] if rows and not dry_run: diff --git a/comments/tasks.py b/comments/tasks.py index d4ccc1a8b9..777a88dc7a 100644 --- a/comments/tasks.py +++ b/comments/tasks.py @@ -103,7 +103,11 @@ def update_current_top_comments_of_week(): update_top_comments_of_week(week_start_date) -@dramatiq.actor +# The monthly run walks a month of long private bot comments, so it needs far +# more than dramatiq's default 10-minute time limit. Retries are capped at one: +# every batch commits as it goes, so a failed run resumes rather than repeats, +# and the default of 20 would just replay the same failure for hours. +@dramatiq.actor(time_limit=1_800_000, max_retries=1) def job_archive_bot_comment_texts(): # Import here to avoid circular imports from comments.services.text_archive import ( diff --git a/comments/views/common.py b/comments/views/common.py index 2f5b30b18d..e93b4c1dcc 100644 --- a/comments/views/common.py +++ b/comments/views/common.py @@ -243,13 +243,17 @@ def comment_full_text_api_view(request: Request, pk: int): comment = get_object_or_404(Comment, pk=pk) - # Private comments resolve to no permission for anyone but their author - permission = get_comment_permission_for_user(comment, user=request.user) - ObjectPermission.can_view(permission, raise_exception=True) - - if comment.is_soft_deleted: - # Mirrors the comment serializer, which never exposes deleted text - raise PermissionDenied("This comment has been deleted.") + # Staff read any comment, deleted or private. Archiving would otherwise + # take away the only view they had of a bot's full text: + # `get_comment_permission_for_user` resolves every private comment to no + # permission but the author's, and the row itself now holds only a stub. + if not (request.user.is_staff or request.user.is_superuser): + permission = get_comment_permission_for_user(comment, user=request.user) + ObjectPermission.can_view(permission, raise_exception=True) + + if comment.is_soft_deleted: + # Mirrors the comment serializer, which never exposes deleted text + raise PermissionDenied("This comment has been deleted.") text = get_full_text(comment) diff --git a/tests/unit/test_comments/test_text_archive.py b/tests/unit/test_comments/test_text_archive.py index 068ad62dd1..24323d8be7 100644 --- a/tests/unit/test_comments/test_text_archive.py +++ b/tests/unit/test_comments/test_text_archive.py @@ -3,12 +3,14 @@ from io import StringIO import pytest # noqa +from django.contrib import admin from django.core.management import call_command from django.core.management.base import CommandError from django.urls import reverse from django.utils import timezone from rest_framework.exceptions import ValidationError +from comments.admin import CommentAdmin from comments.models import Comment from comments.services.common import update_comment from comments.services.text_archive import ( @@ -37,6 +39,11 @@ def bot(user1): return factory_user(username="bot1", email="bot1@metaculus.com", is_bot=True) +@pytest.fixture() +def staff_user(): + return factory_user(username="staff1", email="staff1@metaculus.com", is_staff=True) + + @pytest.fixture() def post(user1): return factory_post( @@ -62,7 +69,7 @@ def factory_archivable_comment(author, post, text=LONG_TEXT, **kwargs): @pytest.fixture() -def s3_stub(mocker): +def s3_stub(mocker, settings): """ Minimal in-memory stand-in for the S3 client used by the archive service. """ @@ -78,7 +85,9 @@ def put_object(self, Bucket, Key, Body, **kwargs): objects[Key] = Body def get_object(self, Bucket, Key): - if Key not in objects: + # A key mapped to None is one the listing still reports but whose + # object has gone: exactly what S3 answers with NoSuchKey + if objects.get(Key) is None: raise Client.exceptions.NoSuchKey() return {"Body": mocker.Mock(read=lambda: objects[Key].encode("utf-8"))} @@ -103,7 +112,7 @@ def paginate(self, Bucket, Prefix): mocker.patch( "comments.services.text_archive.get_boto_client", return_value=Client() ) - mocker.patch("django.conf.settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT", "test-bucket") + settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT = "test-bucket" return objects @@ -307,8 +316,8 @@ def test_archives(self, bot, post, s3_stub): comment.refresh_from_db() assert comment.is_text_archived is True - def test_errors_when_bucket_is_not_configured(self, bot, post, mocker): - mocker.patch("django.conf.settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT", None) + def test_errors_when_bucket_is_not_configured(self, bot, post, settings): + settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT = None with pytest.raises(CommandError): call_command("archive_bot_comment_texts", "--dry-run") @@ -383,6 +392,50 @@ def test_soft_deleted_comment_is_not_readable( assert response.status_code == 403 + def test_staff_reads_someone_elses_private_archived_text( + self, bot, post, s3_stub, staff_user, create_client_for_user + ): + """ + Archiving must not take away the view staff already had in the admin. + """ + + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + + response = create_client_for_user(staff_user).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 200 + assert response.data["text"] == LONG_TEXT + + def test_staff_reads_soft_deleted_archived_text( + self, bot, post, s3_stub, staff_user, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + Comment.objects.filter(pk=comment.pk).update(is_soft_deleted=True) + + response = create_client_for_user(staff_user).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 200 + assert response.data["text"] == LONG_TEXT + + def test_superuser_reads_someone_elses_private_archived_text( + self, bot, post, s3_stub, user_admin, create_client_for_user + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + + response = create_client_for_user(user_admin).get( + reverse("comment-full-text", kwargs={"pk": comment.pk}) + ) + + assert response.status_code == 200 + assert response.data["text"] == LONG_TEXT + def test_missing_archive_object_returns_404( self, bot, post, s3_stub, create_client_for_user ): @@ -397,6 +450,61 @@ def test_missing_archive_object_returns_404( assert response.status_code == 404 +class TestCommentAdminArchivedText: + """ + The admin is where staff investigate a comment, so it has to show the + archived text rather than the stub the row was left with. + """ + + @pytest.fixture() + def comment_admin(self): + return CommentAdmin(Comment, admin.site) + + def test_renders_the_archived_text(self, bot, post, s3_stub, comment_admin): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + comment.refresh_from_db() + + assert LONG_TEXT in comment_admin.archived_text(comment) + + def test_says_so_when_the_archive_cannot_be_read( + self, bot, post, s3_stub, comment_admin + ): + comment = factory_archivable_comment(bot, post) + archive_bot_comment_texts() + comment.refresh_from_db() + s3_stub.clear() + + assert "could not be retrieved" in comment_admin.archived_text(comment) + + def test_field_is_only_added_for_archived_comments( + self, bot, post, s3_stub, comment_admin + ): + comment = factory_archivable_comment(bot, post) + + assert "archived_text" not in comment_admin.get_fields(None, comment) + + archive_bot_comment_texts() + comment.refresh_from_db() + + assert "archived_text" in comment_admin.get_fields(None, comment) + + def test_text_and_archived_text_are_read_only_once_archived( + self, bot, post, s3_stub, comment_admin + ): + comment = factory_archivable_comment(bot, post) + + assert "text" not in comment_admin.get_readonly_fields(None, comment) + + archive_bot_comment_texts() + comment.refresh_from_db() + + readonly_fields = comment_admin.get_readonly_fields(None, comment) + assert "text" in readonly_fields + assert "text_original" in readonly_fields + assert "archived_text" in readonly_fields + + @pytest.fixture() def archived_elsewhere(bot, post, s3_stub): """ @@ -526,6 +634,37 @@ def test_counts_objects_with_no_comment(self, bot, post, s3_stub): assert stats.synced == 0 assert stats.orphaned == 1 + def test_ignores_keys_for_comments_the_archiver_would_never_upload( + self, user1, bot, post, s3_stub + ): + """ + The bucket says what was uploaded, not what may be truncated. A key + pointing at a public human comment is a mistake, not an instruction. + """ + + human = factory_archivable_comment(user1, post, is_private=False) + upload_text(human.pk, LONG_TEXT) + + stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + + assert stats.synced == 0 + assert stats.ineligible == 1 + assert stats.skipped_stale == 0 + + human.refresh_from_db() + assert human.is_text_archived is False + assert human.text_original == LONG_TEXT + + def test_counts_rows_too_short_to_truncate_as_ineligible(self, bot, post, s3_stub): + short = factory_archivable_comment(bot, post, text="c" * ARCHIVE_STUB_LENGTH) + upload_text(short.pk, short.text) + + stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + + assert stats.synced == 0 + assert stats.ineligible == 1 + assert stats.skipped_stale == 0 + def test_verify_skips_rows_whose_archive_no_longer_matches( self, archived_elsewhere, s3_stub ): @@ -544,6 +683,42 @@ def test_verify_accepts_a_matching_archive(self, archived_elsewhere, s3_stub): assert stats.synced == 1 assert stats.mismatched == 0 + assert stats.verify_failed == 0 + + def test_verify_counts_an_unreadable_object_apart_from_a_mismatch( + self, archived_elsewhere, s3_stub, mocker + ): + """ + A read failure says nothing about whether the archive matches, so it + must not be reported as drift. + """ + + mocker.patch( + "comments.services.text_archive.fetch_text", + side_effect=RuntimeError("s3 is down"), + ) + + stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + + assert stats.synced == 0 + assert stats.mismatched == 0 + assert stats.verify_failed == 1 + + archived_elsewhere.refresh_from_db() + assert archived_elsewhere.text_original == LONG_TEXT + + def test_verify_counts_a_missing_object_as_unreadable( + self, archived_elsewhere, s3_stub + ): + s3_stub.clear() + # Put the key back in the listing without a body behind it + s3_stub[build_key(archived_elsewhere.pk)] = None + + stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + + assert stats.synced == 0 + assert stats.mismatched == 0 + assert stats.verify_failed == 1 class TestSyncCommand: @@ -569,8 +744,8 @@ def test_rejects_an_unparseable_snapshot_timestamp(self, s3_stub): with pytest.raises(CommandError): call_command("sync_archived_comment_texts", "--snapshot-at", "yesterday") - def test_errors_when_bucket_is_not_configured(self, mocker): - mocker.patch("django.conf.settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT", None) + def test_errors_when_bucket_is_not_configured(self, settings): + settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT = None with pytest.raises(CommandError): call_command( From 25ec7e0506e26805eeba9d0572154b9780292383 Mon Sep 17 00:00:00 2001 From: lsabor Date: Thu, 20 Aug 2026 13:27:52 -0700 Subject: [PATCH 6/6] refactor: simplify the comment text archive commands - Inline the progress writer into each command and drop the shared `_progress` module. - Remove `--snapshot-at`. The freshness guards go with it, so `get_sync_candidates` and `get_syncable_comments` collapse into one eligibility queryset and `SyncStats.skipped_stale` is gone; the write-time re-select now feeds `ineligible`. `--verify` becomes the only check that an archived copy is still current. - Lower ARCHIVE_MIN_TEXT_LENGTH from 2000 to 500 characters. Co-Authored-By: Claude Opus 5 --- comments/management/commands/_progress.py | 62 ------------ .../commands/archive_bot_comment_texts.py | 55 ++++++++++- .../commands/sync_archived_comment_texts.py | 97 ++++++++++++------- comments/services/text_archive.py | 69 +++++-------- tests/unit/test_comments/test_text_archive.py | 91 +++-------------- 5 files changed, 151 insertions(+), 223 deletions(-) delete mode 100644 comments/management/commands/_progress.py diff --git a/comments/management/commands/_progress.py b/comments/management/commands/_progress.py deleted file mode 100644 index 7f623c82ad..0000000000 --- a/comments/management/commands/_progress.py +++ /dev/null @@ -1,62 +0,0 @@ -""" -Progress reporting shared by the comment text archive commands. - -Django's command discovery skips modules whose name starts with an -underscore, so this sits alongside the commands without becoming one. -""" - -import time - - -def format_duration(seconds: float) -> str: - seconds = int(seconds) - hours, remainder = divmod(seconds, 3600) - minutes, seconds = divmod(remainder, 60) - - if hours: - return f"{hours}h{minutes:02d}m" - if minutes: - return f"{minutes}m{seconds:02d}s" - - return f"{seconds}s" - - -class ProgressWriter: - """ - Prints a running one-line summary with a rate and an ETA. - - These commands process hundreds of thousands of rows over hours, so the - point is to make a long run observable rather than to look pretty. Output - is one line per batch, not a redrawn line, so it survives being piped to - a log file. - """ - - def __init__(self, stdout, total: int = 0): - self.stdout = stdout - self.total = total - self.started = time.monotonic() - - @property - def elapsed(self) -> float: - return time.monotonic() - self.started - - def write(self, line: str) -> None: - self.stdout.write(line) - self.stdout.flush() - - def update(self, done: int, summary: str, detail: str = "") -> None: - elapsed = self.elapsed - rate = done / elapsed if elapsed else 0 - percent = (done / self.total * 100) if self.total else 0 - remaining = max(self.total - done, 0) - eta = format_duration(remaining / rate) if rate else "?" - - line = ( - f" {done:,}/{self.total:,} ({percent:.1f}%) {summary} " - f"{rate:.1f}/s elapsed {format_duration(elapsed)} eta {eta}" - ) - - if detail: - line += f" [{detail}]" - - self.write(line) diff --git a/comments/management/commands/archive_bot_comment_texts.py b/comments/management/commands/archive_bot_comment_texts.py index 1808f8607c..72527ddc49 100644 --- a/comments/management/commands/archive_bot_comment_texts.py +++ b/comments/management/commands/archive_bot_comment_texts.py @@ -1,3 +1,4 @@ +import time from collections.abc import Callable from django.conf import settings @@ -15,7 +16,59 @@ check_is_enabled, ) -from ._progress import ProgressWriter, format_duration + +def format_duration(seconds: float) -> str: + seconds = int(seconds) + hours, remainder = divmod(seconds, 3600) + minutes, seconds = divmod(remainder, 60) + + if hours: + return f"{hours}h{minutes:02d}m" + if minutes: + return f"{minutes}m{seconds:02d}s" + + return f"{seconds}s" + + +class ProgressWriter: + """ + Prints a running one-line summary with a rate and an ETA. + + This command works through hundreds of thousands of rows over hours, so + the point is to make a long run observable rather than to look pretty. + Output is one line per batch, not a redrawn line, so it survives being + piped to a log file. + """ + + def __init__(self, stdout, total: int = 0): + self.stdout = stdout + self.total = total + self.started = time.monotonic() + + @property + def elapsed(self) -> float: + return time.monotonic() - self.started + + def write(self, line: str) -> None: + self.stdout.write(line) + self.stdout.flush() + + def update(self, done: int, summary: str, detail: str = "") -> None: + elapsed = self.elapsed + rate = done / elapsed if elapsed else 0 + percent = (done / self.total * 100) if self.total else 0 + remaining = max(self.total - done, 0) + eta = format_duration(remaining / rate) if rate else "?" + + line = ( + f" {done:,}/{self.total:,} ({percent:.1f}%) {summary} " + f"{rate:.1f}/s elapsed {format_duration(elapsed)} eta {eta}" + ) + + if detail: + line += f" [{detail}]" + + self.write(line) class Command(BaseCommand): diff --git a/comments/management/commands/sync_archived_comment_texts.py b/comments/management/commands/sync_archived_comment_texts.py index 935139eae4..2e642f7d22 100644 --- a/comments/management/commands/sync_archived_comment_texts.py +++ b/comments/management/commands/sync_archived_comment_texts.py @@ -1,7 +1,7 @@ +import time + from django.conf import settings from django.core.management.base import BaseCommand, CommandError -from django.utils.dateparse import parse_datetime -from django.utils.timezone import is_naive, make_aware from comments.services.text_archive import ( DEFAULT_BATCH_SIZE, @@ -12,7 +12,59 @@ sync_archived_comment_texts, ) -from ._progress import ProgressWriter, format_duration + +def format_duration(seconds: float) -> str: + seconds = int(seconds) + hours, remainder = divmod(seconds, 3600) + minutes, seconds = divmod(remainder, 60) + + if hours: + return f"{hours}h{minutes:02d}m" + if minutes: + return f"{minutes}m{seconds:02d}s" + + return f"{seconds}s" + + +class ProgressWriter: + """ + Prints a running one-line summary with a rate and an ETA. + + This command works through hundreds of thousands of objects over hours, + so the point is to make a long run observable rather than to look pretty. + Output is one line per batch, not a redrawn line, so it survives being + piped to a log file. + """ + + def __init__(self, stdout, total: int = 0): + self.stdout = stdout + self.total = total + self.started = time.monotonic() + + @property + def elapsed(self) -> float: + return time.monotonic() - self.started + + def write(self, line: str) -> None: + self.stdout.write(line) + self.stdout.flush() + + def update(self, done: int, summary: str, detail: str = "") -> None: + elapsed = self.elapsed + rate = done / elapsed if elapsed else 0 + percent = (done / self.total * 100) if self.total else 0 + remaining = max(self.total - done, 0) + eta = format_duration(remaining / rate) if rate else "?" + + line = ( + f" {done:,}/{self.total:,} ({percent:.1f}%) {summary} " + f"{rate:.1f}/s elapsed {format_duration(elapsed)} eta {eta}" + ) + + if detail: + line += f" [{detail}]" + + self.write(line) class Command(BaseCommand): @@ -25,17 +77,6 @@ class Command(BaseCommand): ) def add_arguments(self, parser): - parser.add_argument( - "--snapshot-at", - required=True, - help=( - "When the database copy used for the uploads was taken, as an " - "ISO-8601 timestamp (e.g. 2026-08-20T17:00:00Z). Rows created " - "or touched after this are left alone, because the archived " - "copy of their text may be stale. Required: there is no safe " - "default." - ), - ) parser.add_argument( "--dry-run", action="store_true", @@ -47,7 +88,8 @@ def add_arguments(self, parser): help=( "Re-read every archived object and require it to match the row " "before truncating. Much slower, and downloads the whole " - "archive, but does not rely on the timestamp guards alone" + "archive, but it is the only check that the archived copy is " + "still current" ), ) parser.add_argument( @@ -73,25 +115,12 @@ def handle(self, *args, **options): "comment text archiving is disabled." ) - snapshot_at = parse_datetime(options["snapshot_at"]) - - if snapshot_at is None: - raise CommandError( - f"Could not parse --snapshot-at {options['snapshot_at']!r} as an " - "ISO-8601 timestamp." - ) - - if is_naive(snapshot_at): - # A naive timestamp here would be compared against tz-aware columns - # and blow up mid-run, after an unknown number of rows - snapshot_at = make_aware(snapshot_at) - dry_run = options["dry_run"] progress = ProgressWriter(self.stdout) progress.write( f"Syncing against {settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT}/" - f"{S3_KEY_PREFIX}/ as of {snapshot_at.isoformat()}" + f"{S3_KEY_PREFIX}/" + (" (verifying every object)" if options["verify"] else "") ) progress.write("Listing the archive...") @@ -102,7 +131,6 @@ def on_progress(stats: SyncStats) -> None: stats.synced + stats.already_archived + stats.orphaned - + stats.skipped_stale + stats.ineligible + stats.mismatched + stats.verify_failed @@ -112,7 +140,6 @@ def on_progress(stats: SyncStats) -> None: for label, count in ( ("already archived", stats.already_archived), ("orphaned", stats.orphaned), - ("stale", stats.skipped_stale), ("ineligible", stats.ineligible), ("mismatched", stats.mismatched), ("unreadable", stats.verify_failed), @@ -122,7 +149,6 @@ def on_progress(stats: SyncStats) -> None: progress.update(done, f"{stats.chars_reclaimed:,} chars reclaimed", detail) stats = sync_archived_comment_texts( - snapshot_at=snapshot_at, dry_run=dry_run, verify=options["verify"], batch_size=options["batch_size"], @@ -140,7 +166,6 @@ def on_progress(stats: SyncStats) -> None: for label, count in ( ("already truncated", stats.already_archived), ("orphaned (no such comment)", stats.orphaned), - ("skipped as touched since the snapshot", stats.skipped_stale), ("ineligible (not a long private bot comment)", stats.ineligible), ): if count: @@ -152,7 +177,8 @@ def on_progress(stats: SyncStats) -> None: if stats.mismatched: # Not fatal: these keep their text and the monthly job re-archives - # them, but a large number means the snapshot is not what we think + # them, but a large number means the archive is further out of + # date than expected self.stdout.write( self.style.WARNING( f"{stats.mismatched:,} archived object(s) did not match the " @@ -162,8 +188,7 @@ def on_progress(stats: SyncStats) -> None: if stats.verify_failed: # Distinct from a mismatch: nothing is known about these objects, - # so a non-zero count here means the bucket, not the snapshot, is - # what needs looking at + # so a non-zero count here means the bucket is what needs looking at self.stdout.write( self.style.ERROR( f"{stats.verify_failed:,} archived object(s) could not be read " diff --git a/comments/services/text_archive.py b/comments/services/text_archive.py index b61af874dc..eb7459675a 100644 --- a/comments/services/text_archive.py +++ b/comments/services/text_archive.py @@ -3,7 +3,7 @@ from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field -from datetime import datetime, timedelta +from datetime import timedelta from botocore.config import Config from django.conf import settings @@ -22,7 +22,7 @@ ARCHIVE_AGE_DAYS = 30 # Only archive comments whose text is longer than this. Below this, it's not important # to move. -ARCHIVE_MIN_TEXT_LENGTH = 2000 +ARCHIVE_MIN_TEXT_LENGTH = 500 # Length of the stub left behind in the text columns ARCHIVE_STUB_LENGTH = 200 @@ -343,7 +343,7 @@ def list_archived_comment_ids(s3=None) -> set[int]: The bucket is the authority on what has been uploaded: the point of the sync below is to reconcile a database that knows nothing about uploads performed elsewhere. It is not, however, the authority on what may be - truncated — see `get_sync_candidates`. + truncated — see `get_syncable_comments`. """ s3 = s3 or get_archive_s3_client() @@ -371,8 +371,6 @@ class SyncStats: already_archived: int = 0 # Present in the bucket with no matching row: deleted since the upload orphaned: int = 0 - # Touched since the snapshot, so the archived copy may be stale - skipped_stale: int = 0 # A row the archiver would never have uploaded, or one already at or # below the stub length: nothing to reclaim, and a hint that the bucket # holds keys this command did not put there @@ -386,14 +384,18 @@ class SyncStats: sample_ids: list[int] = field(default_factory=list) -def get_sync_candidates(comment_ids) -> QuerySet[Comment]: +def get_syncable_comments(comment_ids) -> QuerySet[Comment]: """ - Rows this command is allowed to truncate at all, ignoring freshness. + Rows this command is allowed to truncate against an archive uploaded + elsewhere. Anything the archiver uploaded was a long, private bot comment, so those invariants are re-asserted here rather than trusting the key alone: a stray or mistyped object in the bucket must not be able to truncate a row the archiver would never have touched. + + Nothing here can tell whether the archived copy is still current — that + is what `--verify` is for. """ return ( @@ -409,29 +411,6 @@ def get_sync_candidates(comment_ids) -> QuerySet[Comment]: ) -def get_syncable_comments(comment_ids, snapshot_at: datetime) -> QuerySet[Comment]: - """ - Candidates that are also safe to truncate against an archive uploaded - elsewhere. - - Everything added here is a guard against the archived copy being stale. - The upload happened against a database snapshot taken at `snapshot_at`; - any row created or touched since then may have text the archive does not - have, so it is left alone. Skipping costs nothing — the monthly job - re-archives it properly — while truncating it would destroy the edit. - """ - - return ( - get_sync_candidates(comment_ids) - .filter(created_at__lt=snapshot_at) - # `edited_at` is bumped by every save; `text_edited_at` only by an - # edit to the text. Both are nullable on rows that predate them, and - # either one moving past the snapshot disqualifies the row. - .filter(Q(edited_at__lt=snapshot_at) | Q(edited_at__isnull=True)) - .filter(Q(text_edited_at__lt=snapshot_at) | Q(text_edited_at__isnull=True)) - ) - - def _verify_archived_text(rows, s3, concurrency: int) -> tuple[set[int], set[int]]: """ Ids whose archived object still matches the row's text exactly, and ids @@ -469,7 +448,6 @@ def _verify_archived_text(rows, s3, concurrency: int) -> tuple[set[int], set[int def sync_archived_comment_texts( - snapshot_at: datetime, dry_run: bool = False, verify: bool = False, batch_size: int = DEFAULT_BATCH_SIZE, @@ -485,9 +463,9 @@ def sync_archived_comment_texts( a second time. `verify` re-reads every object and requires it to match the row before - truncating. That is the safe-but-slow path; without it the timestamp - guards in `get_syncable_comments` are what stand between a stale archive - and a lost edit. + truncating. That is the safe-but-slow path, and the only thing standing + between an archive that has gone stale and a lost edit: without it a row + is truncated on the strength of its key being in the bucket. """ stats = SyncStats() @@ -505,16 +483,14 @@ def sync_archived_comment_texts( for start in range(0, len(archived_ids), batch_size): chunk = archived_ids[start : start + batch_size] - # Three queries per chunk so the accounting is exact: what the - # database knows about these ids, which of them this command may - # touch at all, and which of those are also fresh enough to truncate. + # Two queries per chunk so the accounting is exact: what the database + # knows about these ids, then which of them may be truncated. states = dict( Comment.objects.rewrite(False) .filter(pk__in=chunk) .values_list("id", "is_text_archived") ) - candidate_ids = set(get_sync_candidates(chunk).values_list("id", flat=True)) - syncable = get_syncable_comments(chunk, snapshot_at) + syncable = get_syncable_comments(chunk) if verify: syncable = syncable.annotate(original_text=ORIGINAL_TEXT) @@ -524,8 +500,7 @@ def sync_archived_comment_texts( stats.orphaned += len(chunk) - len(states) already = sum(1 for archived in states.values() if archived) stats.already_archived += already - stats.ineligible += len(states) - already - len(candidate_ids) - stats.skipped_stale += len(candidate_ids) - len(rows) + stats.ineligible += len(states) - already - len(rows) if verify and rows: verified, unreadable = _verify_archived_text(rows, s3, concurrency) @@ -534,14 +509,14 @@ def sync_archived_comment_texts( rows = [row for row in rows if row["id"] in verified] if rows and not dry_run: - # Re-apply the guards at write time: a row edited between the - # select above and this update must not be truncated. - untouched = get_syncable_comments([row["id"] for row in rows], snapshot_at) - synced_ids = set(untouched.values_list("id", flat=True)) - updated = untouched.update(**truncate_kwargs) + # Re-select at write time: a row archived or shortened between the + # select above and this update must not be truncated again. + eligible = get_syncable_comments([row["id"] for row in rows]) + synced_ids = set(eligible.values_list("id", flat=True)) + updated = eligible.update(**truncate_kwargs) stats.synced += updated - stats.skipped_stale += len(rows) - updated + stats.ineligible += len(rows) - updated rows = [row for row in rows if row["id"] in synced_ids] elif rows: stats.synced += len(rows) diff --git a/tests/unit/test_comments/test_text_archive.py b/tests/unit/test_comments/test_text_archive.py index 24323d8be7..c8cfea164d 100644 --- a/tests/unit/test_comments/test_text_archive.py +++ b/tests/unit/test_comments/test_text_archive.py @@ -539,7 +539,7 @@ class TestSyncArchivedCommentTexts: def test_truncates_without_uploading(self, archived_elsewhere, s3_stub): before = dict(s3_stub) - stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + stats = sync_archived_comment_texts() assert stats.synced == 1 assert stats.chars_reclaimed == len(LONG_TEXT) - ARCHIVE_STUB_LENGTH @@ -559,69 +559,23 @@ def test_truncates_without_uploading(self, archived_elsewhere, s3_stub): def test_does_not_bump_edited_at(self, archived_elsewhere, s3_stub): edited_at = archived_elsewhere.edited_at - sync_archived_comment_texts(snapshot_at=timezone.now()) + sync_archived_comment_texts() archived_elsewhere.refresh_from_db() assert archived_elsewhere.edited_at == edited_at def test_dry_run_writes_nothing(self, archived_elsewhere, s3_stub): - stats = sync_archived_comment_texts(snapshot_at=timezone.now(), dry_run=True) + stats = sync_archived_comment_texts(dry_run=True) assert stats.synced == 1 archived_elsewhere.refresh_from_db() assert archived_elsewhere.is_text_archived is False assert archived_elsewhere.text_original == LONG_TEXT - def test_skips_rows_edited_since_the_snapshot(self, archived_elsewhere, s3_stub): - """ - The archived copy predates the edit, so truncating would lose it. - """ - - Comment.objects.rewrite(False).filter(pk=archived_elsewhere.pk).update( - edited_at=timezone.now() - ) - - stats = sync_archived_comment_texts( - snapshot_at=timezone.now() - timedelta(hours=1) - ) - - assert stats.synced == 0 - assert stats.skipped_stale == 1 - - archived_elsewhere.refresh_from_db() - assert archived_elsewhere.is_text_archived is False - assert archived_elsewhere.text_original == LONG_TEXT - - def test_skips_rows_whose_text_was_edited_since_the_snapshot( - self, archived_elsewhere, s3_stub - ): - Comment.objects.rewrite(False).filter(pk=archived_elsewhere.pk).update( - text_edited_at=timezone.now(), edited_at=None - ) - - stats = sync_archived_comment_texts( - snapshot_at=timezone.now() - timedelta(hours=1) - ) - - assert stats.synced == 0 - assert stats.skipped_stale == 1 - - def test_skips_rows_created_since_the_snapshot(self, archived_elsewhere, s3_stub): - Comment.objects.rewrite(False).filter(pk=archived_elsewhere.pk).update( - created_at=timezone.now() - ) - - stats = sync_archived_comment_texts( - snapshot_at=timezone.now() - timedelta(hours=1) - ) - - assert stats.synced == 0 - assert stats.skipped_stale == 1 - def test_counts_already_truncated_rows(self, archived_elsewhere, s3_stub): - sync_archived_comment_texts(snapshot_at=timezone.now()) + sync_archived_comment_texts() - stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + stats = sync_archived_comment_texts() assert stats.synced == 0 assert stats.already_archived == 1 @@ -629,7 +583,7 @@ def test_counts_already_truncated_rows(self, archived_elsewhere, s3_stub): def test_counts_objects_with_no_comment(self, bot, post, s3_stub): upload_text(999_999_999, LONG_TEXT) - stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + stats = sync_archived_comment_texts() assert stats.synced == 0 assert stats.orphaned == 1 @@ -645,11 +599,10 @@ def test_ignores_keys_for_comments_the_archiver_would_never_upload( human = factory_archivable_comment(user1, post, is_private=False) upload_text(human.pk, LONG_TEXT) - stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + stats = sync_archived_comment_texts() assert stats.synced == 0 assert stats.ineligible == 1 - assert stats.skipped_stale == 0 human.refresh_from_db() assert human.is_text_archived is False @@ -659,18 +612,17 @@ def test_counts_rows_too_short_to_truncate_as_ineligible(self, bot, post, s3_stu short = factory_archivable_comment(bot, post, text="c" * ARCHIVE_STUB_LENGTH) upload_text(short.pk, short.text) - stats = sync_archived_comment_texts(snapshot_at=timezone.now()) + stats = sync_archived_comment_texts() assert stats.synced == 0 assert stats.ineligible == 1 - assert stats.skipped_stale == 0 def test_verify_skips_rows_whose_archive_no_longer_matches( self, archived_elsewhere, s3_stub ): upload_text(archived_elsewhere.pk, "something else entirely") - stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + stats = sync_archived_comment_texts(verify=True) assert stats.synced == 0 assert stats.mismatched == 1 @@ -679,7 +631,7 @@ def test_verify_skips_rows_whose_archive_no_longer_matches( assert archived_elsewhere.text_original == LONG_TEXT def test_verify_accepts_a_matching_archive(self, archived_elsewhere, s3_stub): - stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + stats = sync_archived_comment_texts(verify=True) assert stats.synced == 1 assert stats.mismatched == 0 @@ -698,7 +650,7 @@ def test_verify_counts_an_unreadable_object_apart_from_a_mismatch( side_effect=RuntimeError("s3 is down"), ) - stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + stats = sync_archived_comment_texts(verify=True) assert stats.synced == 0 assert stats.mismatched == 0 @@ -714,7 +666,7 @@ def test_verify_counts_a_missing_object_as_unreadable( # Put the key back in the listing without a body behind it s3_stub[build_key(archived_elsewhere.pk)] = None - stats = sync_archived_comment_texts(snapshot_at=timezone.now(), verify=True) + stats = sync_archived_comment_texts(verify=True) assert stats.synced == 0 assert stats.mismatched == 0 @@ -725,29 +677,14 @@ class TestSyncCommand: def test_syncs(self, archived_elsewhere, s3_stub): out = StringIO() - call_command( - "sync_archived_comment_texts", - "--snapshot-at", - timezone.now().isoformat(), - stdout=out, - ) + call_command("sync_archived_comment_texts", stdout=out) assert "Synced 1 of 1 archived object(s)" in out.getvalue() archived_elsewhere.refresh_from_db() assert archived_elsewhere.is_text_archived is True - def test_requires_a_snapshot_timestamp(self, s3_stub): - with pytest.raises(CommandError): - call_command("sync_archived_comment_texts") - - def test_rejects_an_unparseable_snapshot_timestamp(self, s3_stub): - with pytest.raises(CommandError): - call_command("sync_archived_comment_texts", "--snapshot-at", "yesterday") - def test_errors_when_bucket_is_not_configured(self, settings): settings.AWS_STORAGE_BUCKET_COMMENTS_TEXT = None with pytest.raises(CommandError): - call_command( - "sync_archived_comment_texts", "--snapshot-at", "2026-01-01T00:00:00Z" - ) + call_command("sync_archived_comment_texts")