diff --git a/app/db.py b/app/db.py index bf8d10c..b2dc5cb 100644 --- a/app/db.py +++ b/app/db.py @@ -468,6 +468,9 @@ def latest_project(sid): # (table, channel) pairs whose row-level INSERT/UPDATE/DELETE bump a counter. _REV_TRIGGERS = [ ("student_state", "states"), + # Goal profiles are written after the student_state row, so they signal the + # same channel; else an open sheet refetches before the newest run's evidence. + ("goal_profile", "states"), ("trigger_event", "triggers"), ("switch_event", "switches"), ("tracked_student", "roster"), @@ -822,6 +825,7 @@ def _workspace_export_queries(workspace_id): "student_state": (f"SELECT * FROM student_state WHERE {roster}", (ws,)), "trigger_event": (f"SELECT * FROM trigger_event WHERE {roster}", (ws,)), "switch_event": (f"SELECT * FROM switch_event WHERE {roster}", (ws,)), + "goal_profile": (f"SELECT * FROM goal_profile WHERE {roster}", (ws,)), "vex_log": (f"SELECT * FROM vex_log WHERE {vex_roster}", (ws,)), } @@ -995,7 +999,9 @@ def triggers_feed(cutoff, limit=100, workspace_id=None, since=None): params.append(workspace_id) after = "" if since is not None: - after = "AND te.started_at >= ? " + # A reset hides earlier alerts, except one still open: that is the + # student's current state (e.g. idle right now), not history. + after = "AND (te.resolved_at IS NULL OR te.started_at >= ?) " params.append(dt_to_db(since)) params += [dt_to_db(cutoff), limit] rows = _query( @@ -1255,6 +1261,15 @@ def tracked_add(sid, workspace_id=None): ) +def mark_workspace_for_backfill(workspace_id=None): + """Ask the daemon to backfill every student on this workspace's roster again. + Used when a paused board resumes: while it was paused its students weren't + fetched, but the shared cursor moved on if another board was live.""" + _execute( + "UPDATE tracked_student SET backfilled = 0 WHERE workspace_id = ?", (_ws(workspace_id),) + ) + + def mark_backfilled(sid): # Backfill pulls a student's shared history once, so mark every workspace's # roster row for that student (the daemon ingests the union, per student). @@ -1290,6 +1305,10 @@ def tracked_remove(sid, workspace_id=None): con.executemany("DELETE FROM message WHERE id = ?", [(m,) for m in msg_ids]) con.execute("DELETE FROM student_state WHERE studentID = ?", (sid,)) con.execute("DELETE FROM trigger_event WHERE studentID = ?", (sid,)) + # Everything else derived from their events goes too, or a re-add would + # mix these stale goal runs and switches into the rebuilt ones. + con.execute("DELETE FROM goal_profile WHERE studentID = ?", (sid,)) + con.execute("DELETE FROM switch_event WHERE studentID = ?", (sid,)) # ========================================================================== diff --git a/app/main.py b/app/main.py index d9aa9d7..7a2153e 100644 --- a/app/main.py +++ b/app/main.py @@ -893,7 +893,12 @@ def set_polling(body: PollingBody, wsid: int = Depends(current_workspace_id)): paused the daemon makes no prod requests on this board's behalf; it keeps running and picks back up within about a second of being re-enabled. Purely a control flag, prod is untouched either way.""" + was_enabled = _polling_enabled(wsid) db.set_workspace_setting(wsid, "polling_enabled", "1" if body.enabled else "0") + if body.enabled and not was_enabled: + # Refill the pause gap: backfill is idempotent, and the daemon rebuilds + # each student's worker from the log afterwards. + db.mark_workspace_for_backfill(wsid) return {"enabled": body.enabled} diff --git a/app/pipeline/poller.py b/app/pipeline/poller.py index dd1e95e..7cf09d0 100644 --- a/app/pipeline/poller.py +++ b/app/pipeline/poller.py @@ -105,6 +105,11 @@ def drain(client, cursor, limit=500, overlap_seconds=2, tracked=None, since=None max_et = cursor.last_event_time max_id = cursor.last_source_id or 0 + # Collect every page first. Prod pages newest-first, and the workers must see + # each student's events in the order they happened (a run's outcome follows + # its run; run numbers count up), so events are persisted and routed only + # after sorting them oldest-first. + wanted = [] while True: results = client.page_by_time(date_from, limit, offset) if not results: @@ -120,28 +125,33 @@ def drain(client, cursor, limit=500, overlap_seconds=2, tracked=None, since=None continue if since is not None and et is not None and et < since: continue # before the session cutoff -- skip, but the cursor still advanced above - inserted, norm = persist(ev) - if inserted: - try: - route(norm) - new_count += 1 - except Exception: - # The row is already in vex_log, but routing it into the - # in-memory worker failed. Discard that worker so the next - # tick rebuilds it from the DB (which now has this event), - # rather than keep serving a buffer that's silently missing it. - from app.pipeline import workers - - workers._workers.pop(db.canon_id(norm["studentID"]), None) - logger.exception( - "route failed for %s after persist; dropped worker for rehydrate", - norm["studentID"], - ) - raise + wanted.append((et, sid or 0, ev)) if len(results) < limit: break offset += limit + _EPOCH = datetime.min.replace(tzinfo=UTC) + wanted.sort(key=lambda w: (w[0] or _EPOCH, w[1])) + for _, _, ev in wanted: + inserted, norm = persist(ev) + if inserted: + try: + route(norm) + new_count += 1 + except Exception: + # The row is already in vex_log, but routing it into the + # in-memory worker failed. Discard that worker so the next + # tick rebuilds it from the DB (which now has this event), + # rather than keep serving a buffer that's silently missing it. + from app.pipeline import workers + + workers._workers.pop(db.canon_id(norm["studentID"]), None) + logger.exception( + "route failed for %s after persist; dropped worker for rehydrate", + norm["studentID"], + ) + raise + # Drain finished and everything is durably written, now advance the cursor. if max_et != cursor.last_event_time or max_id != (cursor.last_source_id or 0): cursor.last_event_time = max_et diff --git a/docs/concepts/write-path.md b/docs/concepts/write-path.md index fa53369..09c98fc 100644 --- a/docs/concepts/write-path.md +++ b/docs/concepts/write-path.md @@ -70,6 +70,10 @@ correctness machinery in the whole thing. - Each drain pages prod with `dateFrom = last_event_time - overlap`, where overlap is a 2-second window, so events sitting right on a timestamp boundary don't slip through. +- Prod returns pages newest-first, so a drain collects every page, sorts the events + oldest-first, and only then persists and routes them. Workers must see a student's + events in the order they happened: a run's outcome follows its run, and run + numbers count up. - It persists, then advances. The cursor only moves after a full drain is safely written. - Inserts are idempotent. Every event has a unique `source_event_id`, so re-fetched diff --git a/frontend/src/CohortDashboard.jsx b/frontend/src/CohortDashboard.jsx index 0689dd5..7ab3ab2 100644 --- a/frontend/src/CohortDashboard.jsx +++ b/frontend/src/CohortDashboard.jsx @@ -2245,10 +2245,11 @@ const CohortDashboard = () => { try { setNotes((await api.get("/api/notes/", { params: { studentID: sid } })).data.notes || []); } catch { - setNotes([]); + /* keep what's shown: a blip must not look like "no notes" */ } }, []); React.useEffect(() => { + setNotes([]); // a different student: never show the previous one's notes fetchNotes(selected); }, [selected, fetchNotes]); @@ -2256,6 +2257,9 @@ const CohortDashboard = () => { // keeps the grid current while the modal is open: whenever the live feed // refetches (a new alert, an ack), the history follows on the same beat. const [history, setHistory] = React.useState([]); + React.useEffect(() => { + setHistory([]); // a different student: never show the previous one's history + }, [selected]); React.useEffect(() => { if (!selected) { setHistory([]); @@ -2268,7 +2272,7 @@ const CohortDashboard = () => { if (alive) setHistory(r.data.history || []); }) .catch(() => { - if (alive) setHistory([]); + /* keep what's shown: a blip must not empty the grid */ }); return () => { alive = false; @@ -2301,7 +2305,8 @@ const CohortDashboard = () => { const d = (await api.get(`/api/student_states/${encodeURIComponent(selected)}/`)).data; if (alive) setDetailFull(d); } catch { - if (alive) setDetailFull(null); + /* keep the last good payload; the selection effect above already + cleared it if this is a different student */ } if (alive) setDetailFor(selected); })(); diff --git a/tests/test_audit_fixes.py b/tests/test_audit_fixes.py new file mode 100644 index 0000000..93ce79f --- /dev/null +++ b/tests/test_audit_fixes.py @@ -0,0 +1,94 @@ +"""Regression tests for the 2026-09-25 bug audit.""" + +import io +import zipfile +from datetime import timedelta + +from app import db +from app.pipeline import poller, workers + + +class _TimeClient: + """Serves page_by_time pages the way prod does: newest-first.""" + + def __init__(self, pages): + self.pages = pages + + def page_by_time(self, date_from, limit, offset): + idx = offset // limit + return self.pages[idx] if idx < len(self.pages) else [] + + +def _ev(eid, sid, ts): + return { + "id": eid, + "studentID": sid, + "classCode": "C1", + "eventType": "runProject", + "project": "{}", + "raw_message": "{}", + "recieved_at": ts, + } + + +def test_drain_feeds_a_newest_first_page_oldest_first(): + """Prod pages newest-first. Routing in page order fed the worker backwards, + so runs in one tick were numbered newest-first and an outcome could land on + the wrong run.""" + db.tracked_add("d1") + db.mark_backfilled("d1") + stamps = ["2026-09-25T10:00:00Z", "2026-09-25T10:00:05Z", "2026-09-25T10:00:10Z"] + page = [_ev(i + 1, "d1", t) for i, t in enumerate(stamps)][::-1] # newest-first + cursor = poller.get_cursor() + poller.drain(_TimeClient([page]), cursor, limit=500, tracked={"d1"}) + w = workers.get_worker("d1") + w.recompute_and_write() + ts = [r["ts"] for r in db.list_student_states(["d1"])[0]["runs"]["runs"]] + assert len(ts) == 3 and ts == sorted(ts) + + +def test_untracking_purges_goal_evidence_and_switches(): + """Removing a student from the last board must purge everything derived from + their events, or a re-add mixes stale goal runs into the rebuilt ones.""" + db.tracked_add("u1") + db.upsert_goal_profile("u1", 7, {"index": 7, "status": "profiled", "ts": 1.0}) + db.record_switch("u1", "casing", "U1", "u1") + db.tracked_remove("u1") + assert db.list_goal_profiles("u1", session=None) == [] + assert db.list_switches() == [] + + +def test_goal_evidence_writes_signal_the_states_channel(): + """Goal profiles are written after student_state, so without their own signal + an open sheet refetches before the newest run's evidence exists.""" + before = db.data_fingerprint()["states"] + db.upsert_goal_profile("g1", 0, {"index": 0, "status": "profiled", "ts": 1.0}) + assert db.data_fingerprint()["states"] != before + + +def test_reset_keeps_alerts_that_are_still_open(client, seed_state): + """Reset hides history, but an alert still holding (an idle student) is the + student's current state and must stay in the feed.""" + seed_state("o1") + started = db.now() - timedelta(minutes=5) + db.create_trigger("o1", "inactive", started, started, None, {"label": "Inactive"}) + client.post("/api/reset/") + feed = client.get("/api/triggers/").json()["triggers"] + assert [t["trigger_type"] for t in feed] == ["inactive"] + + +def test_export_includes_goal_evidence(seed_state): + seed_state("e1") + db.upsert_goal_profile("e1", 0, {"index": 0, "status": "profiled", "ts": 1.0}) + names = zipfile.ZipFile(io.BytesIO(db.export_zip_bytes())).namelist() + assert any(n.endswith("goal_profile.csv") for n in names) + + +def test_resuming_polling_rebackfills_the_boards_students(client, seed_state): + """While a board is paused its students aren't fetched, but the shared cursor + moves on if another board is live. Resuming must pull the gap back in.""" + seed_state("p1") + db.mark_backfilled("p1") + client.post("/api/polling/", json={"enabled": False}) + client.post("/api/polling/", json={"enabled": True}) + assert db.tracked_list()[0]["backfilled"] is False