fix(members): make backfill remote state visible and stop-safe

The remote state object was first written only at the
--checkpoint-every-th success (default 25), so short or slow runs
appeared to write nothing to S3 until exit — and the exit flush relied
on a finally block that never runs under SIGTERM, meaning a container
stop would silently discard all progress since the last checkpoint.

Remote state now flushes immediately on the first record (the object
appears within seconds and access problems surface at profile 1),
announces every checkpoint on stdout with the exact URI, and SIGTERM
is converted to a graceful exit (143) that persists progress before
the process dies. SIGKILL remains the only loss window, bounded at
checkpoint-every - 1 profiles.

Author: MartinPaulEve

docs/newprofile/backfill_memberships_guide.md

@@ -123,11 +123,24 @@ or `AWS_*` environment variables). The role needs `s3:GetObject` and
 `s3:PutObject` on the state key.
 
 Object stores cannot append, so remote state rewrites the whole object
-every `--checkpoint-every` N processed profiles (default 25) and once
-more on completion. Consequences:
+at checkpoints: immediately after the **first** processed profile (so
+the object appears within seconds of starting and S3 access problems
+surface at profile 1), every `--checkpoint-every` N processed profiles
+thereafter (default 25), and once more on exit. Each checkpoint prints
+a confirmation:
 
-- A hard kill (OOM, SIGKILL) loses at most N−1 profiles of progress;
-  those profiles are simply re-processed on resume. The operation is
+```
+checkpoint: 25 username(s) -> s3://kc-profiles-ops/backfill/state.txt
+```
+
+Consequences:
+
+- **SIGTERM (container stop, `docker stop`, ECS/Kubernetes shutdown)
+  is handled gracefully**: progress is flushed before the process
+  exits (status 143). Only SIGKILL — e.g. the OOM killer, or the
+  follow-up kill after a stop grace period expires — can lose
+  progress, and then at most N−1 profiles since the last checkpoint;
+  those are simply re-processed on resume. The operation is
   idempotent, so this is safe.
 - Lower N for more durability at the cost of one S3 PUT per N
   profiles. `--checkpoint-every 1` writes after every profile, which
@@ -159,7 +172,10 @@ abort the run.
 - **Run resumed but reprocessed some profiles.** Expected with remote
   state: up to `--checkpoint-every − 1` profiles since the last
   checkpoint are retried.
-- **State object never appears in S3.** Check for an exception in the
-  command output: state is first written after the
-  `--checkpoint-every`-th profile, so very short runs with large N
-  only write on completion. Verify the role has `s3:PutObject`.
+- **State object never appears in S3.** The first write happens
+  immediately after the first successfully processed profile, and each
+  checkpoint prints a `checkpoint: … -> s3://…` line. If no checkpoint
+  line appears, no profile has completed yet (look for `Failed:` lines
+  — errored profiles are never recorded); if the line appears but the
+  object is missing, you are looking at the wrong key — the line
+  prints the exact URI being written.

knowledge_commons_profiles/newprofile/management/commands/backfill_memberships.py

@@ -20,6 +20,7 @@
 
 import json
 import logging
+import signal
 import time
 from pathlib import Path
 
@@ -66,16 +67,20 @@ class _RemoteStateStore:
     Resume state in an object store (e.g. s3://) via smart_open.
 
     Object stores cannot append, so the whole state object is rewritten
-    every ``checkpoint_every`` records and once more on close. A hard
-    kill therefore loses at most ``checkpoint_every - 1`` records of
-    progress; those profiles are simply re-processed on resume.
+    on the first record (so the object appears immediately and access
+    problems surface at profile 1), every ``checkpoint_every`` records
+    thereafter, and once more on close. A hard kill therefore loses at
+    most ``checkpoint_every - 1`` records of progress; those profiles
+    are simply re-processed on resume.
     """
 
-    def __init__(self, uri: str, checkpoint_every: int):
+    def __init__(self, uri: str, checkpoint_every: int, announce=None):
         self._uri = uri
         self._every = max(1, checkpoint_every)
+        self._announce = announce
         self._done: set[str] = set()
         self._unflushed = 0
+        self._primed = False
 
     def load(self) -> set[str]:
         try:
@@ -91,14 +96,19 @@ def load(self) -> set[str]:
     def record(self, username: str) -> None:
         self._done.add(username)
         self._unflushed += 1
-        if self._unflushed >= self._every:
+        if not self._primed or self._unflushed >= self._every:
             self._flush()
 
     def _flush(self) -> None:
         content = "".join(f"{name}\n" for name in sorted(self._done))
         with smart_open.open(self._uri, "w") as handle:
             handle.write(content)
+        self._primed = True
         self._unflushed = 0
+        if self._announce:
+            self._announce(
+                f"checkpoint: {len(self._done)} username(s) -> {self._uri}"
+            )
 
     def close(self) -> None:
         if self._unflushed:
@@ -213,18 +223,27 @@ def _build_queryset(options):
 
         return qs
 
-    @staticmethod
-    def _make_store(options):
+    def _make_store(self, options):
         """
         Build the resume-state store for --state-file, if requested.
         """
         target = options["state_file"]
         if not target:
             return None
         if "://" in target:
-            return _RemoteStateStore(target, options["checkpoint_every"])
+            return _RemoteStateStore(
+                target,
+                options["checkpoint_every"],
+                announce=self.stdout.write,
+            )
         return _LocalStateStore(target)
 
+    @staticmethod
+    def _exit_on_sigterm(signum, frame):
+        # container stops deliver SIGTERM; raise so the finally block
+        # persists resume state before the process dies
+        raise SystemExit(143)
+
     def _backfill_profile(self, profile, options) -> bool:
         """
         Backfill one profile; return True if its memberships changed.
@@ -249,6 +268,10 @@ def handle(self, *args, **options):
         state_store = self._make_store(options)
         done = state_store.load() if state_store else set()
 
+        previous_sigterm = signal.signal(
+            signal.SIGTERM, self._exit_on_sigterm
+        )
+
         try:
             for profile in self._build_queryset(options).iterator():
                 if profile.username in done:
@@ -291,6 +314,7 @@ def handle(self, *args, **options):
                 if options["sleep"]:
                     time.sleep(options["sleep"])
         finally:
+            signal.signal(signal.SIGTERM, previous_sigterm)
             if state_store:
                 state_store.close()
 

knowledge_commons_profiles/newprofile/tests/test_backfill_memberships.py

@@ -10,7 +10,10 @@
 """
 
 import json
+import os
+import signal
 import tempfile
+import time
 from io import StringIO
 from pathlib import Path
 from unittest.mock import patch
@@ -362,6 +365,70 @@ def flaky(profile):
             remote.objects[self.URI].split(), ["clyde"]
         )
 
+    def test_remote_state_first_record_flushes_immediately(self):
+        # even with a large checkpoint interval the object must appear
+        # as soon as the first profile is done, so operators can see
+        # the run is working and IAM problems surface at profile 1
+        self._profile_with_role("bonnie")
+        self._profile_with_role("clyde")
+        remote = self._fake_remote()
+
+        self._call("--state-file", self.URI, "--checkpoint-every", "100")
+
+        self.assertEqual(remote.history[0].split(), ["bonnie"])
+
+    def test_remote_state_checkpoints_are_announced(self):
+        self._profile_with_role("bonnie")
+        remote = self._fake_remote()
+
+        out, _ = self._call("--state-file", self.URI)
+
+        self.assertIn(self.URI, out)
+        self.assertTrue(remote.history)
+
+    def test_sigterm_flushes_remote_state_and_exits(self):
+        # container stops deliver SIGTERM; the run must persist its
+        # progress before exiting instead of dying silently
+        self._profile_with_role("bonnie")
+        self._profile_with_role("clyde")
+        remote = self._fake_remote()
+
+        def sentinel(signum, frame):
+            msg = "SIGTERM was not handled gracefully"
+            raise AssertionError(msg)
+
+        previous = signal.signal(signal.SIGTERM, sentinel)
+        self.addCleanup(signal.signal, signal.SIGTERM, previous)
+
+        real_refresh = (
+            "knowledge_commons_profiles.rest_api.sync."
+            "ExternalSync.refresh_local_memberships"
+        )
+        original = ExternalSync.refresh_local_memberships
+
+        def stopped_mid_run(profile):
+            if profile.username == "clyde":
+                os.kill(os.getpid(), signal.SIGTERM)
+                time.sleep(1)  # the handler interrupts this sleep
+            return original(profile)
+
+        with (
+            patch(real_refresh, side_effect=stopped_mid_run),
+            self.assertRaises(SystemExit),
+        ):
+            call_command(
+                "backfill_memberships",
+                "--state-file",
+                self.URI,
+                "--checkpoint-every",
+                "100",
+                "--no-notify",
+                stdout=StringIO(),
+            )
+
+        # bonnie's progress survived the stop; clyde retries on resume
+        self.assertEqual(remote.objects[self.URI].split(), ["bonnie"])
+
     def test_remote_state_dry_run_writes_nothing(self):
         self._profile_with_role("bonnie")
         remote = self._fake_remote()