Bound scheduled todo reminder sweeps

src/msg/workers/maintenance.py

@@ -1,19 +1,21 @@
 """Retention and rebuilding apply deployment-wide, never personal storage quotas."""
 from __future__ import annotations
+
 import asyncio
-from dataclasses import replace
-from datetime import timedelta
 import hashlib
-import os
 import time
+from dataclasses import replace
+from datetime import timedelta
 from pathlib import Path
 
-from msg.constants import ROOT_SPACE, ONLINE_CA
-from msg.core.codec import canonical, decode, digest, loads, wire, parse_time
+from msg.constants import ONLINE_CA, ROOT_SPACE
+from msg.core.codec import canonical, decode, digest, loads, parse_time, wire
 from msg.core.errors import Failure, require
-from msg.core.models import AuditEvent, Event, ResourceRef, Revision, Resource, TransferSession
+from msg.core.models import AuditEvent, Event, Resource, ResourceRef, Revision, TransferSession
 from msg.plugins.common import new_id
 
+TODO_DELIVERY_BATCH_SIZE = 64
+
 
 async def purge_revisions(app, tx, resource, *, actor, request_id, reason):
     revisions = [decode(Revision,loads(raw)) for (raw,) in tx.execute(
@@ -136,7 +138,18 @@ async def _deliver_due_todos(app,tx):
     """
     now=app.clock()
     delivered=0
-    for (raw,) in tx.execute("SELECT body FROM resources WHERE type='todo' AND state='active' ORDER BY id"):
+    # Reading a todo body may invoke the Git content store. Keep that work
+    # bounded while this write transaction holds the deployment-wide writer
+    # lock, and rotate the cursor so a large account cannot starve later todos.
+    cursor=tx.setting('todo_delivery_cursor','')
+    rows=tx.rows("SELECT id,body FROM resources WHERE type='todo' AND state='active' "
+        "AND id>? ORDER BY id LIMIT ?",(cursor,TODO_DELIVERY_BATCH_SIZE))
+    if len(rows)<TODO_DELIVERY_BATCH_SIZE:
+        rows+=tx.rows("SELECT id,body FROM resources WHERE type='todo' AND state='active' "
+            "AND id<=? ORDER BY id LIMIT ?",(cursor,TODO_DELIVERY_BATCH_SIZE-len(rows)))
+    if rows:
+        tx.set_setting('todo_delivery_cursor',rows[-1][0])
+    for _resource_id,raw in rows:
         resource=decode(Resource,loads(raw))
         if resource.revision is None or resource.mode&0o077:
             continue

tests/test_todo_due_inbox.py

@@ -2,10 +2,10 @@
 from datetime import timedelta
 
 import pytest
+from test_service import NOW, call, register
 
 from msg.core.codec import wire
-from msg.workers.maintenance import run_maintenance
-from test_service import NOW, call, register
+from msg.workers.maintenance import TODO_DELIVERY_BATCH_SIZE, run_maintenance
 
 
 @pytest.mark.asyncio
@@ -83,3 +83,37 @@ async def test_only_current_pending_due_revision_can_be_delivered(installed):
         row = tx.one('SELECT body FROM messages WHERE resource=?', (rid,))
         assert row is not None and future in row[0] and past not in row[0]
         assert tx.one('SELECT COUNT(*) FROM messages WHERE resource=?', (archived_id,))[0] == 0
+
+
+@pytest.mark.asyncio
+async def test_due_todo_sweep_is_bounded_and_rotates_through_backlog(installed, monkeypatch):
+    app, _ = installed
+    key, owner, _ = await register(app, 'due-backlog')
+    past = wire(NOW - timedelta(hours=1))
+    total = TODO_DELIVERY_BATCH_SIZE + 3
+    for index in range(total):
+        result = await call(app, 'identity.todo_put',
+            {'name': f'due-{index:03}', 'title': f'Due {index}', 'due_at': past},
+            key=key, subject=owner)
+        assert result.status == 'ok', wire(result)
+
+    reads = 0
+    original_read = app.contents.read_bytes
+
+    async def counted_read(content):
+        nonlocal reads
+        reads += 1
+        return await original_read(content)
+
+    monkeypatch.setattr(app.contents, 'read_bytes', counted_read)
+    first = await run_maintenance(app, 'deliver_due_todos', scheduled=True)
+    assert reads == TODO_DELIVERY_BATCH_SIZE
+    assert first == {'delivered_todo_reminders': TODO_DELIVERY_BATCH_SIZE}
+
+    reads = 0
+    second = await run_maintenance(app, 'deliver_due_todos', scheduled=True)
+    assert reads == TODO_DELIVERY_BATCH_SIZE
+    assert second == {'delivered_todo_reminders': total-TODO_DELIVERY_BATCH_SIZE}
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one("SELECT COUNT(*) FROM messages WHERE recipient=? AND event_id LIKE 'm_%'",
+                      (owner,))[0] == total