fix(memhub): keep unknown retries off the event loop

The retry-unknown endpoint resolved and requeued every unknown row through
its own connection and commit, on the loop. Behind a model outage that is a
few thousand rows, so the whole Runtime stopped answering and SIGTERM could
not be handled until the process was killed.

Resolve and requeue a set in one transaction each, with the state machine
still deciding every row, and run the database work on a worker thread.

Measured on 1,000 unknown rows: 0.313s per-row handles -> 0.028s per set.

CHANGELOG.md

@@ -4,6 +4,10 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- Kept a confirmed unknown-retry from taking the event loop: resolution and
+  requeue now share one database handle per pass over the set instead of one
+  per row, and the database work runs on a worker thread, so other endpoints
+  and shutdown keep answering while a backlog drains.
 - Fixed legacy-to-SDK startup migration to adopt native Coding Experience and
   Working Fact IDs instead of importing them as Chat layers. Raw migration now
   rebuilds bounded user/assistant turns from the latest three days, excludes

memhub/apps/core/src/one_memory/api.py

@@ -3939,19 +3939,29 @@ def create_app(
     async def retry_unknown_memory_tasks(
         request: RetryUnknownMemoryTasksRequest,
     ) -> RetryUnknownMemoryTasksReceipt:
-        """Retry unknown writes only behind the user's explicit confirmation."""
+        """Retry unknown writes only behind the user's explicit confirmation.
+
+        The database work runs on a worker thread: it is a bounded set of rows,
+        not a handful, and doing it on the loop is what made every other
+        endpoint (and the shutdown signal) wait behind it.
+        """
         del request
         reconciled = await _sdk_memory_runtime(app).reconcile_unknown_outputs()
         from one_memory.memory_server.durable_queue import DurableWriteQueue
 
         queue = DurableWriteQueue(app.state.memory_write_queue_path)
-        unknown = queue.unknown_jobs(limit=2048)
-        if app.state.memory_authority is MemoryAuthority.LEGACY:
-            for job in unknown:
-                queue.resolve_unknown_from_linked_output(job.job_id)
-        requeued = sum(queue.retry_unknown_confirmed(job.job_id) for job in unknown)
-        if app.state.memory_authority is MemoryAuthority.LEGACY:
-            requeued += app.state.memory_task_store.retry_unknown_confirmed()
+        legacy_authority = app.state.memory_authority is MemoryAuthority.LEGACY
+
+        def retry_unknown() -> int:
+            job_ids = tuple(job.job_id for job in queue.unknown_jobs(limit=2048))
+            if legacy_authority:
+                queue.resolve_unknown_outputs(job_ids)
+            requeued = queue.retry_unknown_confirmed_many(job_ids)
+            if legacy_authority:
+                requeued += app.state.memory_task_store.retry_unknown_confirmed()
+            return requeued
+
+        requeued = await asyncio.to_thread(retry_unknown)
         return RetryUnknownMemoryTasksReceipt(reconciled=reconciled, requeued=requeued)
 
     @app.get(

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -20,7 +20,7 @@ import re
 import shutil
 import sqlite3
 import time as _time
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterable
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -1767,6 +1767,21 @@ class DurableWriteQueue:
         Returns the new state, or `None` when the row is missing or not
         `unknown` (an already-resolved row is left untouched).
         """
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            return self._resolve_unknown_in(connection, job_id, found=found, detail=detail)
+
+    def _resolve_unknown_in(
+        self,
+        connection: sqlite3.Connection,
+        job_id: str,
+        *,
+        found: bool,
+        detail: str,
+    ) -> str | None:
+        """The inquiry decision itself, so a caller's batch can share one transaction."""
         from one_memory.memory_server.contracts import WriteJobState as JobState
         from one_memory.memory_server.job_state import (
             IllegalTransition,
@@ -1777,31 +1792,27 @@ class DurableWriteQueue:
         event = JobEvent.RESOLVED if found else JobEvent.GIVE_UP
         target = next_state(JobState.UNKNOWN, event)
         now = datetime.now(UTC).isoformat()
-        with (
-            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
-            connection,
-        ):
-            row = connection.execute(
-                "SELECT state FROM memory_write_jobs WHERE job_id=?", (job_id,)
-            ).fetchone()
-            if row is None or str(row["state"]) != UNKNOWN:
-                return None
-            # Re-affirm the transition against the row we actually hold, so a
-            # race that changed it cannot be forced into an illegal state.
-            try:
-                next_state(JobState(str(row["state"])), event)
-            except IllegalTransition:
-                return None
-            connection.execute(
-                "UPDATE memory_write_jobs SET state=?, lease_owner=NULL, "
-                "lease_expires_at=NULL, last_error=?, updated_at=? WHERE job_id=?",
-                (
-                    str(target),
-                    detail or (None if found else "model_content_outcome_not_found"),
-                    now,
-                    job_id,
-                ),
-            )
+        row = connection.execute(
+            "SELECT state FROM memory_write_jobs WHERE job_id=?", (job_id,)
+        ).fetchone()
+        if row is None or str(row["state"]) != UNKNOWN:
+            return None
+        # Re-affirm the transition against the row we actually hold, so a
+        # race that changed it cannot be forced into an illegal state.
+        try:
+            next_state(JobState(str(row["state"])), event)
+        except IllegalTransition:
+            return None
+        connection.execute(
+            "UPDATE memory_write_jobs SET state=?, lease_owner=NULL, "
+            "lease_expires_at=NULL, last_error=?, updated_at=? WHERE job_id=?",
+            (
+                str(target),
+                detail or (None if found else "model_content_outcome_not_found"),
+                now,
+                job_id,
+            ),
+        )
         return str(target)
 
     def retry_unknown_confirmed(self, job_id: str) -> int:
@@ -1814,45 +1825,64 @@ class DurableWriteQueue:
         short transaction, so the unified queue remains the only execution
         entry and a half-reset attempt cannot become permanently stranded.
         """
-        now = datetime.now(UTC).isoformat()
+        return self.retry_unknown_confirmed_many((job_id,))
+
+    def retry_unknown_confirmed_many(self, job_ids: Iterable[str]) -> int:
+        """Requeue a confirmed set in **one** transaction.
+
+        Same rule as the single-row form, applied to a set: the state machine
+        is still consulted per row, but the set shares one connection and one
+        commit. A per-row connection is what turned a 2,048-row retry into
+        hours of synchronous database work on the caller's thread.
+        """
+        ids = tuple(job_ids)
+        if not ids:
+            return 0
+        requeued = 0
         with (
             main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
             connection,
         ):
-            row = connection.execute(
-                "SELECT state,last_error FROM memory_write_jobs WHERE job_id=?", (job_id,)
-            ).fetchone()
-            if (
-                row is None
-                or str(row["state"]) != UNKNOWN
-                or str(row["last_error"]) in NON_RETRYABLE_POST_WRITE_CODES
-            ):
-                return 0
-            connection.execute(
-                """
-                UPDATE memory_tasks
-                SET status='pending', runner_token=NULL, attempts=0,
-                    last_error_code='model_content_manual_retry_requested', updated_at=?
-                WHERE status='unknown' AND task_id IN (
-                    SELECT inputs.task_id
-                    FROM memory_task_inputs AS inputs, json_each(inputs.refs_json) AS ref
-                    JOIN memory_write_job_evidence AS linked
-                      ON linked.evidence_id=json_extract(ref.value, '$.evidence_id')
-                    WHERE linked.job_id=?
-                )
-                """,
-                (now, job_id),
+            for job_id in ids:
+                requeued += self._retry_unknown_confirmed_in(connection, job_id)
+        return requeued
+
+    def _retry_unknown_confirmed_in(self, connection: sqlite3.Connection, job_id: str) -> int:
+        now = datetime.now(UTC).isoformat()
+        row = connection.execute(
+            "SELECT state,last_error FROM memory_write_jobs WHERE job_id=?", (job_id,)
+        ).fetchone()
+        if (
+            row is None
+            or str(row["state"]) != UNKNOWN
+            or str(row["last_error"]) in NON_RETRYABLE_POST_WRITE_CODES
+        ):
+            return 0
+        connection.execute(
+            """
+            UPDATE memory_tasks
+            SET status='pending', runner_token=NULL, attempts=0,
+                last_error_code='model_content_manual_retry_requested', updated_at=?
+            WHERE status='unknown' AND task_id IN (
+                SELECT inputs.task_id
+                FROM memory_task_inputs AS inputs, json_each(inputs.refs_json) AS ref
+                JOIN memory_write_job_evidence AS linked
+                  ON linked.evidence_id=json_extract(ref.value, '$.evidence_id')
+                WHERE linked.job_id=?
             )
-            changed = connection.execute(
-                """
-                UPDATE memory_write_jobs
-                SET state='pending', attempts=0, lease_owner=NULL,
-                    lease_expires_at=NULL, hy_invoked=0,
-                    last_error='model_content_manual_retry_requested', updated_at=?
-                WHERE job_id=? AND state='unknown'
-                """,
-                (now, job_id),
-            ).rowcount
+            """,
+            (now, job_id),
+        )
+        changed = connection.execute(
+            """
+            UPDATE memory_write_jobs
+            SET state='pending', attempts=0, lease_owner=NULL,
+                lease_expires_at=NULL, hy_invoked=0,
+                last_error='model_content_manual_retry_requested', updated_at=?
+            WHERE job_id=? AND state='unknown'
+            """,
+            (now, job_id),
+        ).rowcount
         return changed
 
     def readmit_cancelled_permission_confirmed(self, job_id: str, request: WriteRequest) -> int:
@@ -2020,35 +2050,57 @@ class DurableWriteQueue:
                 )
         return len(job_ids)
 
-    def resolve_unknown_from_linked_output(self, job_id: str) -> bool:
-        """Close an unknown durable row when local governance output proves success."""
-        with main_database(self.database_path).read(row_factory=sqlite3.Row) as connection:
-            row = connection.execute(
-                """
-                SELECT 1
-                FROM memory_write_job_evidence AS linked
-                JOIN memory_task_inputs AS inputs
-                  ON EXISTS (
-                    SELECT 1 FROM json_each(inputs.refs_json) AS ref
-                    WHERE json_extract(ref.value, '$.evidence_id')=linked.evidence_id
-                  )
-                JOIN memory_tasks AS task ON task.task_id=inputs.task_id
-                JOIN memory_task_outputs AS output ON output.task_id=task.task_id
-                WHERE linked.job_id=? AND task.status='completed'
-                LIMIT 1
-                """,
-                (job_id,),
-            ).fetchone()
-        if row is None:
-            return False
-        return (
-            self.resolve_unknown(
-                job_id,
-                found=True,
-                detail="model_content_outcome_reconciled",
-            )
-            == COMPLETED
-        )
+    def resolve_unknown_outputs(self, job_ids: Iterable[str]) -> int:
+        """Close the unknown rows whose local governance output proves success.
+
+        One transaction for the whole set. The state machine still decides each
+        row's next state, so this is the same decision the single-row form made
+        — the only thing removed is the connection-and-commit per job that made
+        a few thousand rows block the Runtime for hours.
+        """
+        ids = tuple(job_ids)
+        if not ids:
+            return 0
+        resolved = 0
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            for job_id in ids:
+                if not self._linked_output_proves_success(connection, job_id):
+                    continue
+                if (
+                    self._resolve_unknown_in(
+                        connection,
+                        job_id,
+                        found=True,
+                        detail="model_content_outcome_reconciled",
+                    )
+                    == COMPLETED
+                ):
+                    resolved += 1
+        return resolved
+
+    @staticmethod
+    def _linked_output_proves_success(connection: sqlite3.Connection, job_id: str) -> bool:
+        """Whether a completed governance task already wrote this Evidence."""
+        row = connection.execute(
+            """
+            SELECT 1
+            FROM memory_write_job_evidence AS linked
+            JOIN memory_task_inputs AS inputs
+              ON EXISTS (
+                SELECT 1 FROM json_each(inputs.refs_json) AS ref
+                WHERE json_extract(ref.value, '$.evidence_id')=linked.evidence_id
+              )
+            JOIN memory_tasks AS task ON task.task_id=inputs.task_id
+            JOIN memory_task_outputs AS output ON output.task_id=task.task_id
+            WHERE linked.job_id=? AND task.status='completed'
+            LIMIT 1
+            """,
+            (job_id,),
+        ).fetchone()
+        return row is not None
 
     def unknown_jobs(
         self,

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -2394,6 +2394,83 @@ def test_user_confirmed_unknown_retry_requeues_job_and_linked_tasks(tmp_path) ->
     assert queue.retry_unknown_confirmed(record.job_id) == 0
 
 
+def test_confirmed_retry_batch_resolves_and_requeues_with_one_handle(
+    tmp_path, monkeypatch
+) -> None:
+    """A confirmed retry set pays one database handle, not one per row.
+
+    Regression: the endpoint resolved and requeued every unknown row through
+    its own connection and commit. A few thousand unknowns therefore held the
+    event loop — and with it every other endpoint and the shutdown signal —
+    for hours, while the same work in one transaction finishes in seconds.
+    """
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    proven = queue.submit(_request(evidence=("e1",), key="idem-proven"))
+    missing = queue.submit(_request(evidence=("e2",), key="idem-missing"))
+    untouched = queue.submit(_request(evidence=("e3",), key="idem-pending"))
+
+    with sqlite3.connect(path) as connection, connection:
+        connection.execute(
+            "UPDATE memory_write_jobs SET state='unknown', attempts=2, hy_invoked=1, "
+            "last_error='hy_memory_timeout' WHERE job_id IN (?,?)",
+            (proven.job_id, missing.job_id),
+        )
+        connection.executescript(
+            """
+            CREATE TABLE memory_tasks (
+                task_id TEXT PRIMARY KEY,
+                status TEXT NOT NULL,
+                runner_token TEXT,
+                attempts INTEGER NOT NULL,
+                last_error_code TEXT,
+                updated_at TEXT NOT NULL
+            );
+            CREATE TABLE memory_task_inputs (
+                task_id TEXT PRIMARY KEY,
+                refs_json TEXT NOT NULL
+            );
+            CREATE TABLE memory_task_outputs (
+                task_id TEXT NOT NULL,
+                memory_id TEXT NOT NULL,
+                body TEXT NOT NULL
+            );
+            """
+        )
+        connection.execute(
+            "INSERT INTO memory_tasks VALUES (?,?,?,?,?,?)",
+            ("task-1", "completed", None, 1, None, "old"),
+        )
+        connection.execute(
+            "INSERT INTO memory_task_inputs VALUES (?,?)",
+            ("task-1", '[{"evidence_id":"e1"}]'),
+        )
+        connection.execute(
+            "INSERT INTO memory_task_outputs VALUES (?,?,?)",
+            ("task-1", "m-1", "{}"),
+        )
+
+    import one_memory.memory_server.durable_queue as durable_queue_module
+
+    handles: list[int] = []
+    real_main_database = durable_queue_module.main_database
+
+    def counting_main_database(*args, **kwargs):
+        handles.append(1)
+        return real_main_database(*args, **kwargs)
+
+    monkeypatch.setattr(durable_queue_module, "main_database", counting_main_database)
+    job_ids = (proven.job_id, missing.job_id, untouched.job_id)
+
+    assert queue.resolve_unknown_outputs(job_ids) == 1
+    assert queue.retry_unknown_confirmed_many(job_ids) == 1
+    assert handles == [1, 1], "one handle per pass over the set, not one per row"
+
+    assert queue.get(proven.request_id).state == COMPLETED
+    assert queue.get(missing.request_id).state == PENDING
+    assert queue.get(untouched.request_id).state == PENDING
+
+
 def test_restart_redrains_a_job_claimed_by_a_worker_that_died(tmp_path) -> None:
     """End-to-end restart recovery: a claimed-but-not-run job runs once.