From 3ea24b6bf7011eb7a1b6b26a9eaf743ddf80b90d Mon Sep 17 00:00:00 2001 From: Sykez <276981287+Sykezzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 03:01:12 -0500 Subject: [PATCH 1/3] perf(oneroster): skip unchanged managed roster reads --- gamgui/components/oneroster/executor.py | 375 +++++++++++-- gamgui/components/oneroster/models.py | 83 +++ gamgui/components/oneroster/planner.py | 664 +++++++++++++++++++++--- gamgui/components/oneroster/semantic.py | 62 +++ gamgui/components/oneroster/store.py | 462 ++++++++++++++++- tests/test_oneroster_delta.py | 463 +++++++++++++++++ 6 files changed, 1983 insertions(+), 126 deletions(-) create mode 100644 gamgui/components/oneroster/semantic.py create mode 100644 tests/test_oneroster_delta.py diff --git a/gamgui/components/oneroster/executor.py b/gamgui/components/oneroster/executor.py index a39b314..4b02bed 100644 --- a/gamgui/components/oneroster/executor.py +++ b/gamgui/components/oneroster/executor.py @@ -26,11 +26,14 @@ GateState, ImportAction, LivePlanningResult, + ManagedCourseDirty, + ManagedCourseVerification, OneRosterError, StudentEnrollmentGate, canonical_hash, ) from .planner import OneRosterPlanner +from .semantic import metadata_hash, student_hash, teacher_hash from .store import OneRosterStore, action_sequence_hash @@ -93,6 +96,12 @@ def observe( self.clean_batches = 0 +@dataclass(frozen=True) +class _VerificationRead: + results: Mapping[str, bool] + verifications: tuple[ManagedCourseVerification, ...] + + class ExecutableClassroomConnector(Protocol): async def run_classroom_batch( self, @@ -209,6 +218,7 @@ async def execute( manifest.import_id, limited_import=manifest.plan_kind == "limited", now=now, + required_metadata_aliases=tuple(manifest.live_evidence), ) self.store.record_execution_planning_receipt( run.id, @@ -403,6 +413,7 @@ async def observe() -> LivePlanningResult: manifest.import_id, limited_import=manifest.plan_kind == "limited", now=now, + required_metadata_aliases=tuple(manifest.live_evidence), ) first = await observe() @@ -474,6 +485,7 @@ async def revalidate_scheduled_gate( manifest.import_id, limited_import=manifest.plan_kind == "limited", now=now, + required_metadata_aliases=tuple(manifest.live_evidence), ) await self._validate_preflight( manifest, @@ -535,7 +547,22 @@ async def reconcile_interrupted( actions = self.store.get_manifest_actions_by_id( manifest_id, batch.action_ids ) - verified = await self._verify(actions) + verification_read = await self._verify(actions) + verified = verification_read.results + self.store.record_managed_course_verifications( + verification_read.verifications + ) + self.store.mark_managed_course_dirty( + _dirty_for_actions( + tuple( + action + for action in actions + if not verified.get(action.id, False) + ) + ), + error_code="OR-RECOVERY-VERIFY-INCOMPLETE", + recovery_required=True, + ) for action in actions: if not verified.get(action.id, False): continue @@ -910,6 +937,11 @@ async def _apply_chunk( durable_batch.id, worker_count=worker_count, ) + self.store.mark_managed_course_dirty( + _dirty_for_actions(chunk), + error_code="OR-EXECUTION-UNCERTAIN", + recovery_required=True, + ) commands = [_command_for(action) for action in chunk] batch_failed = False throttling_count = 0 @@ -951,8 +983,12 @@ async def pulse() -> None: if batch_receipt is not None: throttling_count = int(batch_receipt.throttling_count) try: - verified, verification_attempts, verification_seconds = ( - await self._verify_with_retries(chunk, owner_ids) + verified, verification_attempts, verification_seconds, verifications = ( + await self._verify_with_retries( + chunk, + owner_ids, + include_verifications=True, + ) ) applied = guarded_applied failed = guarded_failed @@ -980,6 +1016,9 @@ async def pulse() -> None: }: blocked_courses.add(action.subject.casefold()) if durable_batch is not None: + dirty = _dirty_for_actions( + tuple(action for action in chunk if not verified.get(action.id)) + ) completed_batch = self.store.complete_verified_batch( durable_batch.id, manifest_id, @@ -990,10 +1029,19 @@ async def pulse() -> None: verification_attempts=verification_attempts, worker_count=worker_count, throttling_count=throttling_count, + managed_verifications=verifications, + managed_dirty=dirty, ) persistence_seconds = completed_batch.persistence_seconds else: persistence_started = time.perf_counter() + self.store.record_managed_course_verifications(verifications) + self.store.mark_managed_course_dirty( + _dirty_for_actions( + tuple(action for action in chunk if not verified.get(action.id)) + ), + error_code="OR-BATCH-VERIFY-FAILED", + ) for action_id, (status, detail) in results.items(): self.store.mark_action_result( manifest_id, @@ -1076,10 +1124,13 @@ async def _verify_with_retries( self, actions: Sequence[ImportAction], owner_ids: Optional[Mapping[str, str]], - ) -> tuple[dict[str, bool], int, float]: + *, + include_verifications: bool = False, + ) -> Any: started = time.perf_counter() remaining = list(actions) verified = {action.id: False for action in actions} + verification_by_alias: dict[str, ManagedCourseVerification] = {} attempts = 0 retry_offsets = (0.0, *self.stabilization_delays) for retry_offset in retry_offsets: @@ -1088,16 +1139,37 @@ async def _verify_with_retries( wait_seconds = retry_offset - (time.perf_counter() - started) if wait_seconds > 0: await asyncio.sleep(wait_seconds) - result = await self._verify(remaining, owner_ids) + raw_read = await self._verify(remaining, owner_ids) + if isinstance(raw_read, _VerificationRead): + read_results = raw_read.results + read_verifications = raw_read.verifications + else: + # Keep the focused private-method test seam compatible with + # mapping-only verification fakes. + read_results = raw_read + read_verifications = () + for verification in read_verifications: + key = verification.alias.casefold() + verification_by_alias[key] = _merge_verification( + verification_by_alias.get(key), + verification, + ) attempts += 1 next_remaining: list[ImportAction] = [] for action in remaining: - if bool(result.get(action.id)): + if bool(read_results.get(action.id)): verified[action.id] = True else: next_remaining.append(action) remaining = next_remaining - return verified, attempts, time.perf_counter() - started + result = ( + verified, + attempts, + time.perf_counter() - started, + ) + if include_verifications: + return (*result, tuple(verification_by_alias.values())) + return result def _record_performance_audit( self, @@ -1221,7 +1293,7 @@ async def _verify( self, actions: Sequence[ImportAction], owner_ids: Optional[Mapping[str, str]] = None, - ) -> dict[str, bool]: + ) -> _VerificationRead: aliases = sorted({action.subject for action in actions}) bulk_courses = getattr( self.connector, @@ -1262,12 +1334,21 @@ async def read(alias: str) -> tuple[str, Optional[Any]]: return alias.casefold(), detail details = dict(await asyncio.gather(*(read(alias) for alias in aliases))) - roster_aliases = { + teacher_aliases = { + action.subject.casefold() + for action in actions + if action.kind in {"teacher_add", "teacher_remove"} + } + student_aliases = { action.subject.casefold() for action in actions - if action.kind in {"teacher_add", "teacher_remove", "student_add", "student_remove"} + if action.kind in {"student_add", "student_remove"} } - rosters: dict[str, tuple[set[str], set[str]]] = {} + roster_aliases = teacher_aliases | student_aliases + rosters: dict[ + str, + tuple[Optional[set[str]], Optional[set[str]]], + ] = {} roster_details = { alias: detail for alias in sorted(roster_aliases) @@ -1280,42 +1361,63 @@ async def read(alias: str) -> tuple[str, Optional[Any]]: None, ) if roster_details and callable(bulk_rosters): - requested_course_ids = [ - _text(detail, "id") for detail in roster_details.values() - ] - try: - participants = await bulk_rosters( - requested_course_ids, - "all", - ) - if ( - not isinstance(participants, CourseRosterSnapshot) - or not participants.covers(requested_course_ids) - ): + by_course = { + _text(detail, "id"): alias + for alias, detail in roster_details.items() + } + both_ids = sorted( + course_id + for course_id, alias in by_course.items() + if alias in teacher_aliases and alias in student_aliases + ) + teacher_ids = sorted( + course_id + for course_id, alias in by_course.items() + if alias in teacher_aliases and alias not in student_aliases + ) + student_ids = sorted( + course_id + for course_id, alias in by_course.items() + if alias in student_aliases and alias not in teacher_aliases + ) + for requested_course_ids, role in ( + (both_ids, "all"), + (teacher_ids, "teachers"), + (student_ids, "students"), + ): + if not requested_course_ids: + continue + try: + participants = await bulk_rosters(requested_course_ids, role) + complete = bool( + isinstance(participants, CourseRosterSnapshot) + and participants.covers(requested_course_ids) + ) + except Exception: + complete = False participants = None - except Exception: - participants = None - if participants is not None: - by_course = { - _text(detail, "id"): alias - for alias, detail in roster_details.items() - } - rosters.update( - { - alias: ( - set(participants.for_course(course_id)[0]), - set(participants.for_course(course_id)[1]), - ) - for course_id, alias in by_course.items() - } - ) + if not complete or participants is None: + continue + for course_id in requested_course_ids: + alias = by_course[course_id] + teachers, students = participants.for_course(course_id) + rosters[alias] = ( + set(teachers) if role in {"all", "teachers"} else None, + set(students) if role in {"all", "students"} else None, + ) elif roster_details: for alias, detail in roster_details.items(): course_id = _text(detail, "id") + teachers_needed = alias in teacher_aliases + students_needed = alias in student_aliases try: teachers, students = await asyncio.gather( - self.connector.list_course_participants(course_id, "teachers"), - self.connector.list_course_participants(course_id, "students"), + self.connector.list_course_participants(course_id, "teachers") + if teachers_needed + else asyncio.sleep(0, result=()), + self.connector.list_course_participants(course_id, "students") + if students_needed + else asyncio.sleep(0, result=()), ) except Exception: continue @@ -1326,18 +1428,124 @@ async def read(alias: str) -> tuple[str, Optional[Any]]: # A course-only GAM row is not proof of a complete roster. continue rosters[alias] = ( - {_participant_email(item) for item in teachers if _participant_email(item)}, - {_participant_email(item) for item in students if _participant_email(item)}, + {_participant_email(item) for item in teachers if _participant_email(item)} + if teachers_needed + else None, + {_participant_email(item) for item in students if _participant_email(item)} + if students_needed + else None, ) - return { - action.id: _verify_action( + action_results: dict[str, bool] = {} + for action in actions: + alias = action.subject.casefold() + roster = rosters.get(alias) + if action.kind in {"teacher_add", "teacher_remove"} and ( + roster is None or roster[0] is None + ): + action_roster = None + elif action.kind in {"student_add", "student_remove"} and ( + roster is None or roster[1] is None + ): + action_roster = None + else: + action_roster = ( + set(roster[0] or ()), + set(roster[1] or ()), + ) if roster is not None else None + action_results[action.id] = _verify_action( action, - details.get(action.subject.casefold()), - rosters.get(action.subject.casefold()), + details.get(alias), + action_roster, owner_ids or {}, ) - for action in actions + + states = self.store.get_managed_course_states(aliases) + kinds_by_alias: dict[str, set[str]] = {} + for action in actions: + kinds_by_alias.setdefault(action.subject.casefold(), set()).add(action.kind) + owners_by_id = { + str(user_id): str(email).casefold() + for email, user_id in (owner_ids or {}).items() + if str(user_id) + } + verifications: list[ManagedCourseVerification] = [] + metadata_kinds = { + "course_create", + "course_update", + "course_activate", + "course_archive", + "owner_transfer", } + for alias in aliases: + key = alias.casefold() + if key not in states: + continue + detail = details.get(key) + if detail is None or not _has_alias(detail, alias): + continue + course_id = _text(detail, "id") + if not course_id: + continue + kinds = kinds_by_alias.get(key, set()) + metadata_needed = bool(kinds & metadata_kinds) + teachers_needed = bool(kinds & {"teacher_add", "teacher_remove"}) + students_needed = bool(kinds & {"student_add", "student_remove"}) + roster = rosters.get(key) + owner_email = _text(detail, "owner_email", "ownerEmail").casefold() + if not owner_email: + owner_email = owners_by_id.get( + _text(detail, "owner_id", "ownerId"), + "", + ) + state = _text(detail, "course_state", "courseState").upper() + metadata_covered = bool(metadata_needed and owner_email and state) + teacher_members = ( + tuple(sorted(roster[0])) + if teachers_needed and roster is not None and roster[0] is not None + else None + ) + student_members = ( + tuple(sorted(roster[1])) + if students_needed and roster is not None and roster[1] is not None + else None + ) + if not (metadata_covered or teacher_members is not None or student_members is not None): + continue + verifications.append( + ManagedCourseVerification( + alias=alias, + course_id=course_id, + metadata_hash=( + metadata_hash( + alias, + _text(detail, "name"), + owner_email, + _text(detail, "room"), + _text(detail, "section"), + state, + ) + if metadata_covered + else None + ), + teacher_hash=( + teacher_hash(teacher_members) + if teacher_members is not None + else None + ), + student_hash=( + student_hash(student_members) + if student_members is not None + else None + ), + teacher_members=teacher_members, + student_members=student_members, + course_state=state if metadata_covered else None, + ) + ) + return _VerificationRead( + results=action_results, + verifications=tuple(verifications), + ) def _command_for(action: ImportAction) -> list[str]: @@ -1440,6 +1648,77 @@ def _verify_action( return False +def _dirty_for_actions( + actions: Sequence[ImportAction], +) -> tuple[ManagedCourseDirty, ...]: + by_alias: dict[str, dict[str, bool]] = {} + metadata_kinds = { + "course_create", + "course_update", + "course_activate", + "course_archive", + "owner_transfer", + } + for action in actions: + scopes = by_alias.setdefault( + action.subject, + {"metadata": False, "teachers": False, "students": False}, + ) + if action.kind in metadata_kinds: + scopes["metadata"] = True + elif action.kind in {"teacher_add", "teacher_remove"}: + scopes["teachers"] = True + elif action.kind in {"student_add", "student_remove"}: + scopes["students"] = True + return tuple( + ManagedCourseDirty(alias=alias, **scopes) + for alias, scopes in sorted(by_alias.items(), key=lambda item: item[0].casefold()) + if any(scopes.values()) + ) + + +def _merge_verification( + previous: Optional[ManagedCourseVerification], + current: ManagedCourseVerification, +) -> ManagedCourseVerification: + if previous is None: + return current + return ManagedCourseVerification( + alias=current.alias, + course_id=current.course_id or previous.course_id, + source_class_id=current.source_class_id or previous.source_class_id, + last_seen_import_id=current.last_seen_import_id or previous.last_seen_import_id, + metadata_hash=( + current.metadata_hash + if current.metadata_hash is not None + else previous.metadata_hash + ), + teacher_hash=( + current.teacher_hash + if current.teacher_hash is not None + else previous.teacher_hash + ), + student_hash=( + current.student_hash + if current.student_hash is not None + else previous.student_hash + ), + teacher_members=( + current.teacher_members + if current.teacher_members is not None + else previous.teacher_members + ), + student_members=( + current.student_members + if current.student_members is not None + else previous.student_members + ), + course_state=current.course_state or previous.course_state, + verified_at=max(previous.verified_at, current.verified_at), + clear_recovery=previous.clear_recovery and current.clear_recovery, + ) + + def _gate_allows( gate: StudentEnrollmentGate, manifest: ClassroomImportManifest, diff --git a/gamgui/components/oneroster/models.py b/gamgui/components/oneroster/models.py index 12f53c0..1a140c1 100644 --- a/gamgui/components/oneroster/models.py +++ b/gamgui/components/oneroster/models.py @@ -463,6 +463,11 @@ class LivePlanningResult: compare=False, repr=False, ) + desired_course_hashes: Mapping[str, "ManagedCourseDesired"] = field( + default_factory=dict, + compare=False, + repr=False, + ) def actions_for(self, plan_kind: str) -> Tuple[ImportAction, ...]: kind = str(plan_kind or "").strip().casefold() @@ -608,6 +613,84 @@ class PlanningPerformanceReceipt: directory_snapshot_seconds: float = 0.0 classroom_snapshot_seconds: float = 0.0 roster_snapshot_seconds: float = 0.0 + total_managed_aliases: int = 0 + candidate_aliases: int = 0 + unchanged_aliases: int = 0 + metadata_reads_requested: int = 0 + teacher_rosters_requested: int = 0 + student_rosters_requested: int = 0 + cached_metadata_scopes: int = 0 + cached_teacher_scopes: int = 0 + cached_student_scopes: int = 0 + audit_courses_requested: int = 0 + + +@dataclass(frozen=True) +class ManagedCourseState: + """Durable, independently scoped verification state for one managed alias.""" + + domain: str + alias: str + course_id: str + source_class_id: str + last_seen_import_id: str + desired_metadata_hash: str = "" + desired_teacher_hash: str = "" + desired_student_hash: str = "" + verified_metadata_hash: str = "" + verified_teacher_hash: str = "" + verified_student_hash: str = "" + last_metadata_verified_at: float = 0.0 + last_teacher_verified_at: float = 0.0 + last_student_verified_at: float = 0.0 + verified_course_state: str = "" + metadata_dirty: bool = True + teacher_roster_dirty: bool = True + student_roster_dirty: bool = True + recovery_required: bool = False + last_error_code: str = "" + version: int = 1 + updated_at: float = 0.0 + + +@dataclass(frozen=True) +class ManagedCourseDesired: + """Accepted desired hashes; recording these never establishes live proof.""" + + alias: str + import_id: str + metadata_hash: str + teacher_hash: str + student_hash: str + source_class_id: str = "" + + +@dataclass(frozen=True) +class ManagedCourseVerification: + """Fresh live evidence for only the explicitly populated verification scopes.""" + + alias: str + course_id: str + source_class_id: str = "" + last_seen_import_id: str = "" + metadata_hash: Optional[str] = None + teacher_hash: Optional[str] = None + student_hash: Optional[str] = None + teacher_members: Optional[Tuple[str, ...]] = None + student_members: Optional[Tuple[str, ...]] = None + course_state: Optional[str] = None + verified_at: float = 0.0 + clear_recovery: bool = True + + +@dataclass(frozen=True) +class ManagedCourseDirty: + """Exact registry scopes whose live outcome is incomplete or uncertain.""" + + alias: str + metadata: bool = False + teachers: bool = False + students: bool = False @dataclass(frozen=True) diff --git a/gamgui/components/oneroster/planner.py b/gamgui/components/oneroster/planner.py index 5a90bd2..8fdb98e 100644 --- a/gamgui/components/oneroster/planner.py +++ b/gamgui/components/oneroster/planner.py @@ -28,10 +28,15 @@ ImportIssue, IssueSeverity, LivePlanningResult, + ManagedCourseDesired, + ManagedCourseDirty, + ManagedCourseState, + ManagedCourseVerification, OneRosterError, PlanningPerformanceReceipt, canonical_hash, ) +from .semantic import metadata_hash, student_hash, teacher_hash from .store import OneRosterStore from .thresholds import evaluate_thresholds @@ -39,6 +44,7 @@ PLANNER_SCHEMA_VERSION = 3 DIRECTORY_CONCURRENCY = 12 COURSE_CONCURRENCY = 8 +DEFAULT_AUDIT_SAMPLE_SIZE = 25 # The current planner uses compact in-memory participant/action models. Keep # district planning comfortably below the measured memory cliff; larger valid # snapshots remain available for inspection and CSV export. @@ -93,6 +99,17 @@ class _LiveCourse: students: tuple[str, ...] = () owner_email: str = "" unresolved_members: tuple[str, ...] = () + metadata_loaded: bool = True + teachers_loaded: bool = True + students_loaded: bool = True + + +@dataclass(frozen=True, slots=True) +class _CourseReadIntent: + metadata: bool + teachers: bool + students: bool + reasons: tuple[str, ...] @dataclass(frozen=True, slots=True) @@ -135,11 +152,18 @@ def __init__( *, directory_concurrency: int = DIRECTORY_CONCURRENCY, course_concurrency: int = COURSE_CONCURRENCY, + audit_sample_size: Optional[int] = None, ) -> None: self.store = store self.connector = connector self.directory_concurrency = max(1, min(int(directory_concurrency), 32)) self.course_concurrency = max(1, min(int(course_concurrency), 16)) + configured_audit = ( + getattr(connector, "oneroster_audit_sample_size", DEFAULT_AUDIT_SAMPLE_SIZE) + if audit_sample_size is None + else audit_sample_size + ) + self.audit_sample_size = max(0, min(int(configured_audit), 25)) async def plan( self, @@ -147,6 +171,7 @@ async def plan( *, limited_import: bool = False, now: Optional[datetime] = None, + required_metadata_aliases: Sequence[str] = (), ) -> LivePlanningResult: planning_started = time.perf_counter() snapshot = self.store.refresh_schedule_scope( @@ -183,7 +208,6 @@ async def plan( protected_aliases = { alias.casefold() for alias in protected_alias_values } - previous_id = self.store.previous_accepted_import_id(import_id) previous_aliases = self.store.previous_accepted_aliases(import_id) relevant_aliases = tuple( sorted( @@ -191,24 +215,39 @@ async def plan( key=str.casefold, ) ) + managed_states = self.store.get_managed_course_states(relevant_aliases) + bootstrap_metadata_aliases = tuple( + alias + for alias in relevant_aliases + if (state := managed_states.get(alias.casefold())) is None + or not state.course_id + or state.metadata_dirty + or state.recovery_required + ) + async def timed_directory_snapshot() -> tuple[Optional[dict[str, Any]], float]: started = time.perf_counter() value = await self._read_directory_snapshot() return value, time.perf_counter() - started - async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: + async def timed_classroom_snapshot( + aliases: Sequence[str], + ) -> tuple[Optional[dict[str, Any]], float]: started = time.perf_counter() - value = await self._read_managed_course_snapshot(relevant_aliases) + value = await self._read_managed_course_snapshot(aliases) return value, time.perf_counter() - started - # Independent live reads use separate private GAMCFGDIRs. GAMRunner - # serializes only refreshed-token persistence, not the read processes. - directory_result, classroom_result = await asyncio.gather( - timed_directory_snapshot(), - timed_classroom_snapshot(), - ) + if bootstrap_metadata_aliases: + directory_result, bootstrap_result = await asyncio.gather( + timed_directory_snapshot(), + timed_classroom_snapshot(bootstrap_metadata_aliases), + ) + bootstrap_courses, classroom_snapshot_seconds = bootstrap_result + else: + directory_result = await timed_directory_snapshot() + bootstrap_courses = {} + classroom_snapshot_seconds = 0.0 directory_snapshot, directory_snapshot_seconds = directory_result - managed_courses, classroom_snapshot_seconds = classroom_result resolved, resolution_issues = await self._resolve_directory( desired, directory_snapshot, @@ -222,19 +261,124 @@ async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: ) issues.extend(course_issues) - roster_started = time.perf_counter() - live_courses = await self._read_live_courses( + desired_states = _desired_course_states(eligible, import_id) + read_plan, unchanged_aliases = _build_course_read_plan( eligible, - managed_courses, - directory_snapshot, + desired_states, + managed_states, + previous_aliases=previous_aliases, + protected_aliases=protected_alias_values, ) - live_courses, live_resolution_issues = await self._resolve_live_participants( - live_courses, - resolved, - directory_snapshot, + audit_aliases = self.store.select_managed_audit_aliases( + tuple( + course.alias + for course in eligible + if course.alias.casefold() in unchanged_aliases + ), + limit=self.audit_sample_size, + ) + for alias in audit_aliases: + key = alias.casefold() + read_plan[key] = _CourseReadIntent( + metadata=True, + teachers=True, + students=True, + reasons=("bounded_audit",), + ) + unchanged_aliases.discard(key) + required_metadata_keys = { + key + for alias in required_metadata_aliases + if (key := _managed_alias_key(alias)) + } + for alias in relevant_aliases: + key = alias.casefold() + if key not in required_metadata_keys: + continue + current = read_plan.get( + key, + _CourseReadIntent(False, False, False, ()), + ) + read_plan[key] = _CourseReadIntent( + metadata=True, + teachers=current.teachers, + students=current.students, + reasons=(*current.reasons, "execution_identity_guard"), + ) + unchanged_aliases.discard(key) + + # Desired hashes establish intent, not live proof. New aliases enter as + # dirty stubs until exact live evidence supplies a stable course ID. + self.store.record_managed_course_desired(tuple(desired_states.values())) + managed_states = self.store.get_managed_course_states(relevant_aliases) + cached_members = self.store.verified_managed_members_many( + tuple(course.alias for course in eligible) ) - issues.extend(live_resolution_issues) - roster_snapshot_seconds = time.perf_counter() - roster_started + + metadata_aliases = tuple( + alias + for alias in relevant_aliases + if (intent := read_plan.get(alias.casefold())) is not None + and intent.metadata + ) + try: + remaining_metadata_aliases = tuple( + alias + for alias in metadata_aliases + if alias.casefold() + not in {item.casefold() for item in bootstrap_metadata_aliases} + ) + managed_courses = bootstrap_courses + if remaining_metadata_aliases and managed_courses is not None: + additional_courses, additional_seconds = ( + await timed_classroom_snapshot(remaining_metadata_aliases) + ) + classroom_snapshot_seconds += additional_seconds + if additional_courses is None: + managed_courses = None + else: + managed_courses = {**managed_courses, **additional_courses} + roster_started = time.perf_counter() + live_courses = await self._read_live_courses( + eligible, + managed_courses, + directory_snapshot, + read_plan=read_plan, + managed_states=managed_states, + cached_members=cached_members, + ) + live_courses, live_resolution_issues = await self._resolve_live_participants( + live_courses, + resolved, + directory_snapshot, + ) + issues.extend(live_resolution_issues) + roster_snapshot_seconds = time.perf_counter() - roster_started + self._persist_live_verification( + import_id, + eligible, + live_courses, + read_plan, + ) + except BaseException: + dirty = tuple( + ManagedCourseDirty( + alias=alias, + metadata=intent.metadata, + teachers=intent.teachers, + students=intent.students, + ) + for alias, intent in ( + (course.alias, read_plan.get(course.alias.casefold())) + for course in eligible + ) + if intent is not None + ) + self.store.mark_managed_course_dirty( + dirty, + error_code="OR-LIVE-READ-INCOMPLETE", + ) + raise archive_actions, archive_basis, archive_issues = await self._archive_actions( import_id, @@ -293,7 +437,40 @@ async def timed_classroom_snapshot() -> tuple[Optional[dict[str, Any]], float]: directory_snapshot_seconds=directory_snapshot_seconds, classroom_snapshot_seconds=classroom_snapshot_seconds, roster_snapshot_seconds=roster_snapshot_seconds, + total_managed_aliases=len(relevant_aliases), + candidate_aliases=len(read_plan), + unchanged_aliases=len(unchanged_aliases), + metadata_reads_requested=len(metadata_aliases), + teacher_rosters_requested=sum( + intent.teachers for intent in read_plan.values() + ), + student_rosters_requested=sum( + intent.students for intent in read_plan.values() + ), + cached_metadata_scopes=sum( + not bool( + (intent := read_plan.get(course.alias.casefold())) + and intent.metadata + ) + for course in eligible + ), + cached_teacher_scopes=sum( + not bool( + (intent := read_plan.get(course.alias.casefold())) + and intent.teachers + ) + for course in eligible + ), + cached_student_scopes=sum( + not bool( + (intent := read_plan.get(course.alias.casefold())) + and intent.students + ) + for course in eligible + ), + audit_courses_requested=len(audit_aliases), ), + desired_course_hashes=desired_states, ) async def _read_directory_snapshot(self) -> Optional[dict[str, Any]]: bulk = getattr(self.connector, "list_oneroster_directory", None) @@ -314,6 +491,8 @@ async def _read_managed_course_snapshot( self, aliases: Sequence[str], ) -> Optional[dict[str, Any]]: + if not aliases: + return {} bulk = getattr(self.connector, "list_oneroster_managed_courses", None) if not callable(bulk): return None @@ -378,16 +557,44 @@ async def _read_live_courses( courses: Sequence[_DesiredCourse], managed_courses: Optional[Mapping[str, Any]] = None, directory_snapshot: Optional[Mapping[str, Any]] = None, + *, + read_plan: Optional[Mapping[str, _CourseReadIntent]] = None, + managed_states: Optional[Mapping[str, ManagedCourseState]] = None, + cached_members: Optional[ + Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]] + ] = None, ) -> dict[str, Optional[_LiveCourse]]: + intents = read_plan or {} + states = managed_states or {} + retained_members = cached_members or {} if managed_courses is not None: details = { - course.alias.casefold(): managed_courses.get(course.alias.casefold()) + course.alias.casefold(): ( + managed_courses.get(course.alias.casefold()) + if bool( + (intent := intents.get(course.alias.casefold())) + and intent.metadata + ) + else _cached_course_detail( + course, + states.get(course.alias.casefold()), + ) + ) for course in courses } else: semaphore = asyncio.Semaphore(self.course_concurrency) async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: + intent = intents.get(course.alias.casefold()) + if not intent or not intent.metadata: + return ( + course.alias.casefold(), + _cached_course_detail( + course, + states.get(course.alias.casefold()), + ), + ) async with semaphore: try: detail = await self.connector.get_course( @@ -406,8 +613,7 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: return course.alias.casefold(), detail details = dict(await asyncio.gather(*(read(course) for course in courses))) - present = [detail for detail in details.values() if detail is not None] - rosters = await self._read_rosters(present) + rosters = await self._read_rosters(details, intents) result: dict[str, Optional[_LiveCourse]] = {} for course in courses: key = course.alias.casefold() @@ -416,7 +622,18 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: result[key] = None continue course_id = _text(detail, "id") - teachers, students = rosters.get(course_id, ((), ())) + cached_teachers, cached_students = retained_members.get(key, ((), ())) + read_teachers, read_students = rosters.get(course_id, (None, None)) + teachers = ( + tuple(read_teachers) + if read_teachers is not None + else tuple(cached_teachers) + ) + students = ( + tuple(read_students) + if read_students is not None + else tuple(cached_students) + ) owner_email = _text( detail, "owner_email", @@ -435,6 +652,11 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: teachers, students, owner_email, + metadata_loaded=bool( + (intent := intents.get(key)) and intent.metadata + ), + teachers_loaded=bool(intent and intent.teachers), + students_loaded=bool(intent and intent.students), ) result[key] = live return result @@ -513,49 +735,100 @@ async def resolve(email: str) -> tuple[str, str]: ), owner_email=mapping.get(live.owner_email, ""), unresolved_members=unresolved, + metadata_loaded=live.metadata_loaded, + teachers_loaded=live.teachers_loaded, + students_loaded=live.students_loaded, ) return normalized, issues async def _read_rosters( self, - details: Sequence[Any], - ) -> dict[str, tuple[tuple[str, ...], tuple[str, ...]]]: - if not details: + details: Mapping[str, Optional[Any]], + read_plan: Mapping[str, _CourseReadIntent], + ) -> dict[ + str, + tuple[Optional[tuple[str, ...]], Optional[tuple[str, ...]]], + ]: + requested: dict[str, tuple[bool, bool]] = {} + for alias, detail in details.items(): + intent = read_plan.get(alias) + course_id = _text(detail, "id") if detail is not None else "" + if not intent or not course_id or not (intent.teachers or intent.students): + continue + requested[course_id] = (intent.teachers, intent.students) + if not requested: return {} - course_ids = [_text(detail, "id") for detail in details if _text(detail, "id")] + both_ids = sorted( + course_id + for course_id, (teachers, students) in requested.items() + if teachers and students + ) + teacher_ids = sorted( + course_id + for course_id, (teachers, students) in requested.items() + if teachers and not students + ) + student_ids = sorted( + course_id + for course_id, (teachers, students) in requested.items() + if students and not teachers + ) + result: dict[ + str, + tuple[Optional[tuple[str, ...]], Optional[tuple[str, ...]]], + ] = {} bulk = getattr(self.connector, "list_course_participants_many", None) if callable(bulk): - try: - participants = await bulk(course_ids, "all") - if ( - not isinstance(participants, CourseRosterSnapshot) - or not participants.covers(course_ids) - ): - raise ValueError( - "The Classroom roster snapshot did not prove complete " - "coverage of the requested courses." + for course_ids, role in ( + (both_ids, "all"), + (teacher_ids, "teachers"), + (student_ids, "students"), + ): + if not course_ids: + continue + try: + participants = await bulk(course_ids, role) + if ( + not isinstance(participants, CourseRosterSnapshot) + or not participants.covers(course_ids) + ): + raise ValueError( + "The Classroom roster snapshot did not prove complete " + "coverage of the requested courses." + ) + except Exception as exc: + raise OneRosterError( + "OR-CLASSROOM-ROSTER-READ", + "Live Classroom rosters could not be read safely.", + ) from exc + for course_id in course_ids: + teachers, students = participants.for_course(course_id) + result[course_id] = ( + tuple(sorted(teachers)) if role in {"all", "teachers"} else None, + tuple(sorted(students)) if role in {"all", "students"} else None, ) - except Exception as exc: - raise OneRosterError( - "OR-CLASSROOM-ROSTER-READ", - "Live Classroom rosters could not be read safely.", - ) from exc - return { - course_id: ( - tuple(sorted(participants.for_course(course_id)[0])), - tuple(sorted(participants.for_course(course_id)[1])), - ) - for course_id in course_ids - } + return result semaphore = asyncio.Semaphore(self.course_concurrency) - async def read(course_id: str) -> tuple[str, tuple[str, ...], tuple[str, ...]]: + async def read( + course_id: str, + teachers_needed: bool, + students_needed: bool, + ) -> tuple[ + str, + Optional[tuple[str, ...]], + Optional[tuple[str, ...]], + ]: async with semaphore: try: teachers, students = await asyncio.gather( - self.connector.list_course_participants(course_id, "teachers"), - self.connector.list_course_participants(course_id, "students"), + self.connector.list_course_participants(course_id, "teachers") + if teachers_needed + else asyncio.sleep(0, result=()), + self.connector.list_course_participants(course_id, "students") + if students_needed + else asyncio.sleep(0, result=()), ) except Exception as exc: raise OneRosterError( @@ -564,17 +837,127 @@ async def read(course_id: str) -> tuple[str, tuple[str, ...], tuple[str, ...]]: ) from exc return ( course_id, - tuple(sorted({_participant_email(item) for item in teachers if _participant_email(item)})), - tuple(sorted({_participant_email(item) for item in students if _participant_email(item)})), + tuple(sorted({_participant_email(item) for item in teachers if _participant_email(item)})) + if teachers_needed + else None, + tuple(sorted({_participant_email(item) for item in students if _participant_email(item)})) + if students_needed + else None, ) return { course_id: (teachers, students) for course_id, teachers, students in await asyncio.gather( - *(read(course_id) for course_id in course_ids) + *( + read(course_id, teachers, students) + for course_id, (teachers, students) in requested.items() + ) ) } + def _persist_live_verification( + self, + import_id: str, + courses: Sequence[_DesiredCourse], + live_courses: Mapping[str, Optional[_LiveCourse]], + read_plan: Mapping[str, _CourseReadIntent], + ) -> None: + verified: list[ManagedCourseVerification] = [] + dirty: list[ManagedCourseDirty] = [] + identity_dirty: list[ManagedCourseDirty] = [] + known_states = self.store.get_managed_course_states( + tuple(course.alias for course in courses) + ) + for course in courses: + key = course.alias.casefold() + intent = read_plan.get(key) + live = live_courses.get(key) + if intent is None or live is None: + continue + course_id = _text(live.detail, "id") + known = known_states.get(key) + if known is not None and known.course_id and course_id != known.course_id: + identity_dirty.append( + ManagedCourseDirty( + course.alias, + metadata=True, + teachers=intent.teachers, + students=intent.students, + ) + ) + continue + exact_alias = _has_exact_alias(live.detail, course.alias) + state = _text(live.detail, "course_state", "courseState").upper() + metadata_covered = bool( + intent.metadata + and course_id + and exact_alias + and live.owner_email + and state + ) + roster_covered = bool( + course_id and exact_alias and not live.unresolved_members + ) + if intent.metadata and not metadata_covered: + dirty.append(ManagedCourseDirty(course.alias, metadata=True)) + if intent.teachers and not roster_covered: + dirty.append(ManagedCourseDirty(course.alias, teachers=True)) + if intent.students and not roster_covered: + dirty.append(ManagedCourseDirty(course.alias, students=True)) + if not (metadata_covered or (intent.teachers and roster_covered) or (intent.students and roster_covered)): + continue + verified.append( + ManagedCourseVerification( + alias=course.alias, + course_id=course_id, + source_class_id=course.class_id, + last_seen_import_id=import_id, + metadata_hash=( + metadata_hash( + course.alias, + _text(live.detail, "name"), + live.owner_email, + _text(live.detail, "room"), + _text(live.detail, "section"), + state, + ) + if metadata_covered + else None + ), + teacher_hash=( + teacher_hash(live.teachers) + if intent.teachers and roster_covered + else None + ), + student_hash=( + student_hash(live.students) + if intent.students and roster_covered + else None + ), + teacher_members=( + tuple(live.teachers) + if intent.teachers and roster_covered + else None + ), + student_members=( + tuple(live.students) + if intent.students and roster_covered + else None + ), + course_state=state if metadata_covered else None, + ) + ) + self.store.record_managed_course_verifications(tuple(verified)) + self.store.mark_managed_course_dirty( + tuple(dirty), + error_code="OR-LIVE-READ-INCOMPLETE", + ) + self.store.mark_managed_course_dirty( + tuple(identity_dirty), + error_code="OR-MANAGED-COURSE-ID-DRIFT", + recovery_required=True, + ) + async def _archive_actions( self, import_id: str, @@ -1007,6 +1390,126 @@ def _managed_alias_key(value: Any) -> str: return alias.casefold() +def _desired_course_states( + courses: Sequence[_DesiredCourse], + import_id: str, +) -> dict[str, ManagedCourseDesired]: + return { + course.alias.casefold(): ManagedCourseDesired( + alias=course.alias, + import_id=import_id, + source_class_id=course.class_id, + metadata_hash=metadata_hash( + course.alias, + course.name, + course.owner_email, + course.room, + course.section, + "ACTIVE", + ), + teacher_hash=teacher_hash(_desired_members(course, "teacher")), + student_hash=student_hash(_desired_members(course, "student")), + ) + for course in courses + } + + +def _build_course_read_plan( + courses: Sequence[_DesiredCourse], + desired: Mapping[str, ManagedCourseDesired], + managed: Mapping[str, ManagedCourseState], + *, + previous_aliases: Sequence[str], + protected_aliases: Sequence[str], +) -> tuple[dict[str, _CourseReadIntent], set[str]]: + plan: dict[str, _CourseReadIntent] = {} + unchanged: set[str] = set() + for course in courses: + key = course.alias.casefold() + target = desired[key] + state = managed.get(key) + reasons: list[str] = [] + if state is None: + plan[key] = _CourseReadIntent( + metadata=True, + teachers=True, + students=True, + reasons=("new_course",), + ) + continue + metadata_changed = target.metadata_hash != state.desired_metadata_hash + teachers_changed = target.teacher_hash != state.desired_teacher_hash + students_changed = target.student_hash != state.desired_student_hash + metadata_dirty = bool( + state.metadata_dirty + or not state.course_id + or target.metadata_hash != state.verified_metadata_hash + ) + teachers_dirty = bool( + state.teacher_roster_dirty + or target.teacher_hash != state.verified_teacher_hash + ) + students_dirty = bool( + state.student_roster_dirty + or target.student_hash != state.verified_student_hash + ) + if metadata_changed: + reasons.append("metadata_changed") + if teachers_changed: + reasons.append("teachers_changed") + if students_changed: + reasons.append("students_changed") + if metadata_dirty and not metadata_changed: + reasons.append("dirty_metadata") + if teachers_dirty and not teachers_changed: + reasons.append("dirty_teachers") + if students_dirty and not students_changed: + reasons.append("dirty_students") + if state.recovery_required: + reasons.append("recovery_required") + metadata_dirty = teachers_dirty = students_dirty = True + intent = _CourseReadIntent( + metadata=metadata_changed or metadata_dirty, + teachers=teachers_changed or teachers_dirty, + students=students_changed or students_dirty, + reasons=tuple(reasons), + ) + if intent.metadata or intent.teachers or intent.students: + plan[key] = intent + else: + unchanged.add(key) + + protected = {alias.casefold() for alias in protected_aliases} + for alias in previous_aliases: + key = alias.casefold() + if key not in protected: + plan[key] = _CourseReadIntent( + metadata=True, + teachers=False, + students=False, + reasons=("removed_from_source",), + ) + unchanged.discard(key) + return plan, unchanged + + +def _cached_course_detail( + course: _DesiredCourse, + state: Optional[ManagedCourseState], +) -> Optional[dict[str, Any]]: + if state is None or not state.course_id: + return None + return { + "id": state.course_id, + "aliases": (course.alias, f"d:{course.alias}"), + "name": course.name, + "section": course.section, + "room": course.room, + "owner_email": course.owner_email, + "course_state": state.verified_course_state or "ACTIVE", + } + + def _read_desired_courses( path: Any, domain: str, @@ -1336,7 +1839,11 @@ def _append_existing_course_actions( "section": course.section, "room": course.room, } - if not limited_import and current_metadata != desired_metadata: + if ( + live.metadata_loaded + and not limited_import + and current_metadata != desired_metadata + ): _append_action( ordinary, _action( @@ -1349,7 +1856,7 @@ def _append_existing_course_actions( ) state = _text(detail, "course_state", "courseState").upper() - if not limited_import and state != "ACTIVE": + if live.metadata_loaded and not limited_import and state != "ACTIVE": _append_action( ordinary, _action( @@ -1367,27 +1874,30 @@ def _append_existing_course_actions( current_students = set(live.students) current_owner = live.owner_email - for teacher in sorted(desired_teachers - current_teachers): - _append_action(ordinary, _action("teacher_add", course.alias, teacher)) - if not limited_import: - for teacher in sorted(current_teachers - desired_teachers): - if teacher == current_owner: - continue - _append_action( - ordinary, - _action("teacher_remove", course.alias, teacher), - ) - for student in sorted(desired_students - current_students): - _append_action(ordinary, _action("student_add", course.alias, student)) - if not limited_import: - for student in sorted(current_students - desired_students): - _append_action( - ordinary, - _action("student_remove", course.alias, student), - ) + if live.teachers_loaded: + for teacher in sorted(desired_teachers - current_teachers): + _append_action(ordinary, _action("teacher_add", course.alias, teacher)) + if not limited_import: + for teacher in sorted(current_teachers - desired_teachers): + if teacher == current_owner: + continue + _append_action( + ordinary, + _action("teacher_remove", course.alias, teacher), + ) + if live.students_loaded: + for student in sorted(desired_students - current_students): + _append_action(ordinary, _action("student_add", course.alias, student)) + if not limited_import: + for student in sorted(current_students - desired_students): + _append_action( + ordinary, + _action("student_remove", course.alias, student), + ) if ( - not limited_import + live.metadata_loaded + and not limited_import and current_owner and current_owner != course.owner_email ): @@ -1448,7 +1958,7 @@ def _live_basis(alias: str, live: _LiveCourse) -> dict[str, Any]: "alias": alias, "exists": True, "id": _text(detail, "id"), - "aliases": sorted(_aliases(detail)), + "aliases": [alias] if _has_exact_alias(detail, alias) else sorted(_aliases(detail)), "name": _text(detail, "name"), "section": _text(detail, "section"), "room": _text(detail, "room"), diff --git a/gamgui/components/oneroster/semantic.py b/gamgui/components/oneroster/semantic.py new file mode 100644 index 0000000..ffefa07 --- /dev/null +++ b/gamgui/components/oneroster/semantic.py @@ -0,0 +1,62 @@ +"""Pure canonical semantic hashing for OneRoster managed-course state.""" + +from __future__ import annotations + +from typing import Any, Iterable + +from .models import canonical_hash + + +def normalize_email_values(values: Iterable[object]) -> tuple[str, ...]: + """Return deterministic membership semantics without retaining identities twice.""" + + return tuple( + sorted( + { + str(value or "").strip().casefold() + for value in values + if str(value or "").strip() + } + ) + ) + + +def metadata_basis( + alias: object, + name: object, + owner_email: object, + room: object, + section: object, + state: object = "ACTIVE", +) -> dict[str, Any]: + """Build the exact desired metadata basis without touching authored descriptions.""" + + return { + "alias": str(alias or "").strip(), + "name": str(name or "").strip(), + "owner_email": str(owner_email or "").strip().casefold(), + "room": str(room or "").strip(), + "section": str(section or "").strip(), + "state": str(state or "").strip().upper(), + } + + +def metadata_hash( + alias: object, + name: object, + owner_email: object, + room: object, + section: object, + state: object = "ACTIVE", +) -> str: + return canonical_hash( + metadata_basis(alias, name, owner_email, room, section, state) + ) + + +def teacher_hash(values: Iterable[object]) -> str: + return canonical_hash(normalize_email_values(values)) + + +def student_hash(values: Iterable[object]) -> str: + return canonical_hash(normalize_email_values(values)) diff --git a/gamgui/components/oneroster/store.py b/gamgui/components/oneroster/store.py index 26383fe..75f3b2f 100644 --- a/gamgui/components/oneroster/store.py +++ b/gamgui/components/oneroster/store.py @@ -44,6 +44,10 @@ ImportAction, ImportIssue, IssueSeverity, + ManagedCourseDesired, + ManagedCourseDirty, + ManagedCourseState, + ManagedCourseVerification, ManifestPage, MAX_PAGE_SIZE, OneRosterError, @@ -67,6 +71,7 @@ preview_total, source_rows, ) +from .semantic import normalize_email_values, student_hash, teacher_hash from .thresholds import evaluation_hash @@ -97,7 +102,6 @@ def default_component_data_root() -> Path: ), ) - def _progress_phase(kind: str) -> tuple[str, str]: normalized = str(kind or "").casefold() for key, label, kinds in _PROGRESS_PHASES: @@ -560,6 +564,346 @@ def mark_accepted(self, import_id: str, *, when: Optional[float] = None) -> None (accepted_at, self.domain, import_id), ) + def get_managed_course_states( + self, + aliases: Optional[Sequence[str]] = None, + ) -> dict[str, ManagedCourseState]: + """Return durable verification state keyed by case-folded exact alias.""" + + requested = ( + {_managed_alias(alias).casefold() for alias in aliases} + if aliases is not None + else None + ) + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT * FROM managed_course_state + WHERE domain = ? ORDER BY alias COLLATE NOCASE + """, + (self.domain,), + ).fetchall() + return { + str(row["alias"]).casefold(): _managed_course_state_from_row(row) + for row in rows + if requested is None or str(row["alias"]).casefold() in requested + } + + def record_managed_course_desired( + self, + values: Sequence[ManagedCourseDesired], + *, + now: Optional[float] = None, + ) -> None: + """Persist authoritative desired hashes without claiming live verification.""" + + timestamp = float(now if now is not None else time.time()) + normalized: list[ManagedCourseDesired] = [] + for value in values: + normalized.append( + ManagedCourseDesired( + alias=_managed_alias(value.alias), + import_id=str(value.import_id or "").strip(), + metadata_hash=_validate_digest(value.metadata_hash), + teacher_hash=_validate_digest(value.teacher_hash), + student_hash=_validate_digest(value.student_hash), + source_class_id=str(value.source_class_id or "").strip(), + ) + ) + if not normalized: + return + with closing(self._conn()) as conn, conn: + conn.executemany( + """ + INSERT INTO managed_course_state ( + domain, alias, course_id, source_class_id, last_seen_import_id, + desired_metadata_hash, desired_teacher_hash, desired_student_hash, + verified_metadata_hash, verified_teacher_hash, verified_student_hash, + last_metadata_verified_at, last_teacher_verified_at, + last_student_verified_at, verified_course_state, + metadata_dirty, teacher_roster_dirty, student_roster_dirty, + recovery_required, last_error_code, version, updated_at + ) VALUES ( + ?, ?, '', ?, ?, ?, ?, ?, '', '', '', 0, 0, 0, '', + 1, 1, 1, 0, '', 1, ? + ) + ON CONFLICT(domain, alias) DO UPDATE SET + source_class_id = excluded.source_class_id, + last_seen_import_id = excluded.last_seen_import_id, + metadata_dirty = CASE + WHEN managed_course_state.verified_metadata_hash != excluded.desired_metadata_hash + THEN 1 ELSE managed_course_state.metadata_dirty END, + teacher_roster_dirty = CASE + WHEN managed_course_state.verified_teacher_hash != excluded.desired_teacher_hash + THEN 1 ELSE managed_course_state.teacher_roster_dirty END, + student_roster_dirty = CASE + WHEN managed_course_state.verified_student_hash != excluded.desired_student_hash + THEN 1 ELSE managed_course_state.student_roster_dirty END, + desired_metadata_hash = excluded.desired_metadata_hash, + desired_teacher_hash = excluded.desired_teacher_hash, + desired_student_hash = excluded.desired_student_hash, + version = managed_course_state.version + 1, + updated_at = excluded.updated_at + """, + ( + ( + self.domain, + value.alias, + value.source_class_id, + value.import_id, + value.metadata_hash, + value.teacher_hash, + value.student_hash, + timestamp, + ) + for value in normalized + ), + ) + + def record_managed_course_verifications( + self, + values: Sequence[ManagedCourseVerification], + *, + now: Optional[float] = None, + ) -> None: + """Persist only explicitly covered live scopes in one transaction.""" + + if not values: + return + timestamp = float(now if now is not None else time.time()) + with closing(self._conn()) as conn, conn: + for value in values: + self._record_managed_course_verification( + conn, + value, + timestamp=timestamp, + ) + + def mark_managed_course_dirty( + self, + values: Sequence[ManagedCourseDirty], + *, + error_code: str, + recovery_required: bool = False, + now: Optional[float] = None, + ) -> None: + """Mark only uncertain scopes dirty; unrelated verification stays valid.""" + + if not values: + return + timestamp = float(now if now is not None else time.time()) + with closing(self._conn()) as conn, conn: + for value in values: + alias = _managed_alias(value.alias) + conn.execute( + """ + UPDATE managed_course_state SET + metadata_dirty = CASE WHEN ? THEN 1 ELSE metadata_dirty END, + teacher_roster_dirty = CASE WHEN ? THEN 1 ELSE teacher_roster_dirty END, + student_roster_dirty = CASE WHEN ? THEN 1 ELSE student_roster_dirty END, + recovery_required = CASE WHEN ? THEN 1 ELSE recovery_required END, + last_error_code = ?, version = version + 1, updated_at = ? + WHERE domain = ? AND alias = ? + """, + ( + bool(value.metadata), + bool(value.teachers), + bool(value.students), + bool(recovery_required), + str(error_code or "")[:100], + timestamp, + self.domain, + alias, + ), + ) + + def verified_managed_members(self, alias: str, role: str) -> tuple[str, ...]: + managed_alias = _managed_alias(alias) + normalized_role = str(role or "").strip().casefold() + if normalized_role not in {"teachers", "students"}: + raise ValueError("Managed-course role must be teachers or students.") + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT email FROM managed_course_members + WHERE domain = ? AND alias = ? AND role = ? + ORDER BY email + """, + (self.domain, managed_alias, normalized_role), + ).fetchall() + return tuple(str(row["email"]) for row in rows) + + def verified_managed_members_many( + self, + aliases: Sequence[str], + ) -> dict[str, tuple[tuple[str, ...], tuple[str, ...]]]: + requested = {_managed_alias(alias).casefold() for alias in aliases} + result: dict[str, tuple[list[str], list[str]]] = { + alias: ([], []) for alias in requested + } + if not requested: + return {} + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT alias, role, email FROM managed_course_members + WHERE domain = ? ORDER BY alias COLLATE NOCASE, role, email + """, + (self.domain,), + ).fetchall() + for row in rows: + key = str(row["alias"]).casefold() + if key not in result: + continue + teachers, students = result[key] + (teachers if str(row["role"]) == "teachers" else students).append( + str(row["email"]) + ) + return { + alias: (tuple(teachers), tuple(students)) + for alias, (teachers, students) in result.items() + } + + def select_managed_audit_aliases( + self, + aliases: Sequence[str], + *, + limit: int = 25, + ) -> tuple[str, ...]: + """Choose the oldest fully verified eligible aliases deterministically.""" + + cap = max(0, min(int(limit), 25)) + if not cap: + return () + requested = {_managed_alias(alias).casefold() for alias in aliases} + with closing(self._conn()) as conn: + rows = conn.execute( + """ + SELECT alias, + MIN(last_metadata_verified_at, + last_teacher_verified_at, + last_student_verified_at) AS oldest + FROM managed_course_state + WHERE domain = ? + AND metadata_dirty = 0 + AND teacher_roster_dirty = 0 + AND student_roster_dirty = 0 + AND recovery_required = 0 + ORDER BY oldest ASC, alias COLLATE NOCASE ASC + """, + (self.domain,), + ).fetchall() + return tuple( + str(row["alias"]) + for row in rows + if str(row["alias"]).casefold() in requested + )[:cap] + + def _record_managed_course_verification( + self, + conn: sqlite3.Connection, + value: ManagedCourseVerification, + *, + timestamp: float, + ) -> None: + alias = _managed_alias(value.alias) + row = conn.execute( + "SELECT * FROM managed_course_state WHERE domain = ? AND alias = ?", + (self.domain, alias), + ).fetchone() + if row is None: + raise OneRosterError( + "OR-MANAGED-STATE-MISSING", + "Live verification cannot adopt an unregistered Classroom course.", + ) + course_id = str(value.course_id or "").strip() + existing_id = str(row["course_id"] or "").strip() + if existing_id and course_id and existing_id != course_id: + raise OneRosterError( + "OR-MANAGED-COURSE-ID-DRIFT", + "The exact managed alias resolved to a different Classroom course.", + ) + effective_course_id = course_id or existing_id + verified_at = float(value.verified_at or timestamp) + updates: dict[str, Any] = { + "course_id": effective_course_id, + "source_class_id": str(value.source_class_id or row["source_class_id"] or ""), + "last_seen_import_id": str(value.last_seen_import_id or row["last_seen_import_id"] or ""), + "last_error_code": "", + "updated_at": timestamp, + } + if value.metadata_hash is not None: + updates.update( + verified_metadata_hash=_validate_digest(value.metadata_hash), + last_metadata_verified_at=verified_at, + verified_course_state=str(value.course_state or "").strip().upper(), + metadata_dirty=0, + ) + for role, supplied_hash, supplied_members, hash_function, dirty_column, time_column, hash_column in ( + ( + "teachers", + value.teacher_hash, + value.teacher_members, + teacher_hash, + "teacher_roster_dirty", + "last_teacher_verified_at", + "verified_teacher_hash", + ), + ( + "students", + value.student_hash, + value.student_members, + student_hash, + "student_roster_dirty", + "last_student_verified_at", + "verified_student_hash", + ), + ): + if supplied_hash is None: + if supplied_members is not None: + raise ValueError("Managed-course members require a matching verified hash.") + continue + if supplied_members is None: + raise ValueError("A verified roster hash requires covered member evidence.") + members = normalize_email_values(supplied_members) + digest = _validate_digest(supplied_hash) + if hash_function(members) != digest: + raise ValueError("Managed-course roster members do not match the verified hash.") + conn.execute( + "DELETE FROM managed_course_members WHERE domain = ? AND alias = ? AND role = ?", + (self.domain, alias, role), + ) + conn.executemany( + """ + INSERT INTO managed_course_members(domain, alias, role, email, verified_at) + VALUES (?, ?, ?, ?, ?) + """, + ((self.domain, alias, role, email, verified_at) for email in members), + ) + updates[hash_column] = digest + updates[time_column] = verified_at + updates[dirty_column] = 0 + assignments = ", ".join(f"{column} = ?" for column in updates) + conn.execute( + f""" + UPDATE managed_course_state SET {assignments}, version = version + 1 + WHERE domain = ? AND alias = ? + """, + (*updates.values(), self.domain, alias), + ) + if value.clear_recovery: + conn.execute( + """ + UPDATE managed_course_state + SET recovery_required = CASE + WHEN metadata_dirty = 0 AND teacher_roster_dirty = 0 + AND student_roster_dirty = 0 THEN 0 + ELSE recovery_required END + WHERE domain = ? AND alias = ? + """, + (self.domain, alias), + ) + def maybe_mark_import_accepted( self, manifest_id: str, @@ -1519,6 +1863,8 @@ def complete_verified_batch( verification_attempts: int, worker_count: int, throttling_count: int = 0, + managed_verifications: Sequence[ManagedCourseVerification] = (), + managed_dirty: Sequence[ManagedCourseDirty] = (), now: Optional[float] = None, ) -> ExecutionBatch: """Atomically save exact verified results and complete their durable batch.""" @@ -1605,6 +1951,33 @@ def complete_verified_batch( "OR-BATCH-ACTIONS-CHANGED", "Not every immutable batch action could be updated.", ) + for verification in managed_verifications: + self._record_managed_course_verification( + conn, + verification, + timestamp=timestamp, + ) + for dirty in managed_dirty: + alias = _managed_alias(dirty.alias) + conn.execute( + """ + UPDATE managed_course_state SET + metadata_dirty = CASE WHEN ? THEN 1 ELSE metadata_dirty END, + teacher_roster_dirty = CASE WHEN ? THEN 1 ELSE teacher_roster_dirty END, + student_roster_dirty = CASE WHEN ? THEN 1 ELSE student_roster_dirty END, + last_error_code = 'OR-BATCH-VERIFY-FAILED', + version = version + 1, updated_at = ? + WHERE domain = ? AND alias = ? + """, + ( + bool(dirty.metadata), + bool(dirty.teachers), + bool(dirty.students), + timestamp, + self.domain, + alias, + ), + ) failed = any(status == "failed" for status, _detail in normalized.values()) persistence_seconds = time.perf_counter() - persist_started conn.execute( @@ -2866,6 +3239,52 @@ def _init_state(self) -> None: ); CREATE INDEX IF NOT EXISTS accepted_aliases_domain_time ON accepted_managed_aliases(domain, accepted_at DESC, alias); + CREATE TABLE IF NOT EXISTS managed_course_state ( + domain TEXT NOT NULL, + alias TEXT NOT NULL, + course_id TEXT NOT NULL DEFAULT '', + source_class_id TEXT NOT NULL DEFAULT '', + last_seen_import_id TEXT NOT NULL DEFAULT '', + desired_metadata_hash TEXT NOT NULL DEFAULT '', + desired_teacher_hash TEXT NOT NULL DEFAULT '', + desired_student_hash TEXT NOT NULL DEFAULT '', + verified_metadata_hash TEXT NOT NULL DEFAULT '', + verified_teacher_hash TEXT NOT NULL DEFAULT '', + verified_student_hash TEXT NOT NULL DEFAULT '', + last_metadata_verified_at REAL NOT NULL DEFAULT 0, + last_teacher_verified_at REAL NOT NULL DEFAULT 0, + last_student_verified_at REAL NOT NULL DEFAULT 0, + verified_course_state TEXT NOT NULL DEFAULT '', + metadata_dirty INTEGER NOT NULL DEFAULT 1, + teacher_roster_dirty INTEGER NOT NULL DEFAULT 1, + student_roster_dirty INTEGER NOT NULL DEFAULT 1, + recovery_required INTEGER NOT NULL DEFAULT 0, + last_error_code TEXT NOT NULL DEFAULT '', + version INTEGER NOT NULL DEFAULT 1, + updated_at REAL NOT NULL DEFAULT 0, + PRIMARY KEY(domain, alias) + ); + CREATE UNIQUE INDEX IF NOT EXISTS managed_course_state_course_id + ON managed_course_state(domain, course_id) + WHERE course_id != ''; + CREATE INDEX IF NOT EXISTS managed_course_state_audit + ON managed_course_state( + domain, metadata_dirty, teacher_roster_dirty, + student_roster_dirty, recovery_required, + last_metadata_verified_at, last_teacher_verified_at, + last_student_verified_at, alias + ); + CREATE TABLE IF NOT EXISTS managed_course_members ( + domain TEXT NOT NULL, + alias TEXT NOT NULL, + role TEXT NOT NULL CHECK(role IN ('teachers', 'students')), + email TEXT NOT NULL, + verified_at REAL NOT NULL, + PRIMARY KEY(domain, alias, role, email), + FOREIGN KEY(domain, alias) + REFERENCES managed_course_state(domain, alias) + ON DELETE CASCADE + ); """ ) _ensure_column( @@ -3070,6 +3489,47 @@ def _snapshot_write_conn(path: Path) -> sqlite3.Connection: raise +def _managed_alias(value: object) -> str: + alias = str(value or "").strip() + if ( + not alias.startswith("Section_") + or len(alias) == len("Section_") + or any(character in alias for character in "\r\n\x00") + ): + raise OneRosterError( + "OR-ALIAS-INVALID", + "A managed Classroom alias must use the exact Section_ form.", + ) + return alias + + +def _managed_course_state_from_row(row: sqlite3.Row) -> ManagedCourseState: + return ManagedCourseState( + domain=str(row["domain"]), + alias=str(row["alias"]), + course_id=str(row["course_id"] or ""), + source_class_id=str(row["source_class_id"] or ""), + last_seen_import_id=str(row["last_seen_import_id"] or ""), + desired_metadata_hash=str(row["desired_metadata_hash"] or ""), + desired_teacher_hash=str(row["desired_teacher_hash"] or ""), + desired_student_hash=str(row["desired_student_hash"] or ""), + verified_metadata_hash=str(row["verified_metadata_hash"] or ""), + verified_teacher_hash=str(row["verified_teacher_hash"] or ""), + verified_student_hash=str(row["verified_student_hash"] or ""), + last_metadata_verified_at=float(row["last_metadata_verified_at"] or 0), + last_teacher_verified_at=float(row["last_teacher_verified_at"] or 0), + last_student_verified_at=float(row["last_student_verified_at"] or 0), + verified_course_state=str(row["verified_course_state"] or ""), + metadata_dirty=bool(row["metadata_dirty"]), + teacher_roster_dirty=bool(row["teacher_roster_dirty"]), + student_roster_dirty=bool(row["student_roster_dirty"]), + recovery_required=bool(row["recovery_required"]), + last_error_code=str(row["last_error_code"] or ""), + version=int(row["version"] or 1), + updated_at=float(row["updated_at"] or 0), + ) + + def _managed_aliases_from_snapshot(path: Path, domain: str) -> tuple[str, ...]: with closing(_snapshot_conn(path)) as conn: rows = conn.execute( diff --git a/tests/test_oneroster_delta.py b/tests/test_oneroster_delta.py new file mode 100644 index 0000000..5d67396 --- /dev/null +++ b/tests/test_oneroster_delta.py @@ -0,0 +1,463 @@ +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from gamgui.components.oneroster import OneRosterService, ThresholdProfile +from gamgui.components.oneroster.executor import OneRosterExecutor +from gamgui.components.oneroster.models import ( + ManagedCourseDesired, + ManagedCourseDirty, + OneRosterError, +) +from gamgui.components.oneroster.semantic import student_hash, teacher_hash +from gamgui.components.oneroster.store import OneRosterStore +from gamgui.core.classroom.models import ( + CourseDetail, + CourseParticipant, + CourseRosterSnapshot, +) +from gamgui.core.gam.models import BatchExecutionReceipt, GAMUser +from tests.test_oneroster_helpers import valid_files, zip_bytes + + +class DeltaConnector: + oneroster_audit_sample_size = 0 + oneroster_stabilization_delays = (0.0, 0.0) + + def __init__(self) -> None: + self.users = { + email: GAMUser(email, user_id=user_id) + for email, user_id in { + "teacher@example.org": "teacher-id", + "teacher2@example.org": "teacher2-id", + "student@example.org": "student-id", + "student-2@example.org": "student-2-id", + }.items() + } + detail = CourseDetail( + id="1000", + name="Algebra I – P1 (2026-27)", + section="P1", + room="101", + owner_id="teacher-id", + owner_email="teacher@example.org", + course_state="ACTIVE", + aliases=("d:Section_101",), + ) + self.courses = {"Section_101": detail} + self.teachers = {"1000": {"teacher@example.org"}} + self.students = {"1000": {"student@example.org"}} + self.metadata_calls: list[tuple[str, ...]] = [] + self.roster_calls: list[tuple[tuple[str, ...], str]] = [] + self.directory_calls = 0 + self.batches: list[tuple[tuple[str, ...], ...]] = [] + self.incomplete_rosters = False + self.unrelated_courses: dict[str, CourseDetail] = {} + + def reset_course_reads(self) -> None: + self.metadata_calls.clear() + self.roster_calls.clear() + + async def list_oneroster_directory(self): + self.directory_calls += 1 + return dict(self.users) + + async def list_oneroster_managed_courses(self, aliases): + requested = tuple(str(alias) for alias in aliases) + self.metadata_calls.append(requested) + requested_keys = {alias.removeprefix("d:").casefold() for alias in requested} + return [ + detail + for alias, detail in self.courses.items() + if alias.casefold() in requested_keys + ] + + async def get_course(self, course_id: str, **_kwargs): + alias = str(course_id).removeprefix("d:") + try: + return self.courses[alias] + except KeyError: + raise KeyError(alias) from None + + async def list_course_participants_many(self, course_ids, role="all"): + requested = tuple(str(course_id) for course_id in course_ids) + self.roster_calls.append((requested, str(role))) + if self.incomplete_rosters: + return CourseRosterSnapshot.empty() + participants: list[CourseParticipant] = [] + for course_id in requested: + if role in {"all", "teachers"}: + participants.extend( + CourseParticipant( + course_id=course_id, + email=email, + role="teachers", + ) + for email in sorted(self.teachers.get(course_id, set())) + ) + if role in {"all", "students"}: + participants.extend( + CourseParticipant( + course_id=course_id, + email=email, + role="students", + ) + for email in sorted(self.students.get(course_id, set())) + ) + return CourseRosterSnapshot.from_participants(participants, requested) + + async def list_course_participants(self, course_id: str, role: str): + source = self.teachers if role == "teachers" else self.students + return [ + CourseParticipant(course_id=course_id, email=email, role=role) + for email in sorted(source.get(course_id, set())) + ] + + async def run_classroom_batch( + self, + commands, + *, + max_commands=50, + worker_count=5, + ): + assert 0 < len(commands) <= max_commands <= 50 + self.batches.append(tuple(tuple(command) for command in commands)) + for raw_command in commands: + command = list(raw_command) + alias = command[1].removeprefix("d:") + detail = self.courses[alias] + operation, role, email = command[2], command[3], command[4] + members = ( + self.teachers[detail.id] + if role == "teachers" + else self.students[detail.id] + ) + if operation == "add": + members.add(email) + else: + members.discard(email) + return BatchExecutionReceipt( + duration_seconds=0.001, + worker_count=worker_count, + outcome="completed", + ) + + +def _service(tmp_path: Path, files: dict[str, str] | None = None): + service = OneRosterService("example.org", tmp_path / "component") + snapshot = service.upload(zip_bytes(files or valid_files())) + service.save_threshold_profile(ThresholdProfile(configured=True)) + return service, snapshot.id + + +async def _baseline(tmp_path: Path): + service, import_id = _service(tmp_path) + connector = DeltaConnector() + plan = await service.build_live_plan(connector, import_id) + return service, connector, import_id, plan + + +def _teacher_change_files() -> dict[str, str]: + files = valid_files() + files["users.csv"] += ( + "teacher-2,active,teacher2,teacher2@example.org,Grace,Teacher,t2,school-1\n" + ) + files["enrollments.csv"] += ( + "enrollment-teacher-2,active,101,school-1,teacher-2,teacher,false,,\n" + ) + return files + + +def _metadata_change_files() -> dict[str, str]: + files = valid_files() + files["courses.csv"] = files["courses.csv"].replace("Algebra I", "Geometry") + return files + + +@pytest.mark.asyncio +async def test_first_baseline_then_identical_plan_skips_live_course_reads( + tmp_path: Path, +): + service, connector, import_id, first = await _baseline(tmp_path) + + assert first.actions == () + assert connector.metadata_calls == [("Section_101",)] + assert connector.roster_calls == [(('1000',), "all")] + state = service.store.get_managed_course_states(["Section_101"])["section_101"] + assert state.course_id == "1000" + assert state.desired_metadata_hash == state.verified_metadata_hash + assert state.desired_teacher_hash == state.verified_teacher_hash + assert state.desired_student_hash == state.verified_student_hash + assert not ( + state.metadata_dirty + or state.teacher_roster_dirty + or state.student_roster_dirty + or state.recovery_required + ) + assert service.store.verified_managed_members("Section_101", "teachers") == ( + "teacher@example.org", + ) + assert service.store.verified_managed_members("Section_101", "students") == ( + "student@example.org", + ) + + connector.reset_course_reads() + second = await service.build_live_plan(connector, import_id) + + assert second.actions == () + assert connector.metadata_calls == [] + assert connector.roster_calls == [] + assert second.performance.total_managed_aliases == 1 + assert second.performance.candidate_aliases == 0 + assert second.performance.unchanged_aliases == 1 + assert second.performance.metadata_reads_requested == 0 + assert second.performance.teacher_rosters_requested == 0 + assert second.performance.student_rosters_requested == 0 + assert second.performance.cached_metadata_scopes == 1 + assert second.performance.cached_teacher_scopes == 1 + assert second.performance.cached_student_scopes == 1 + assert second.performance.audit_courses_requested == 0 + + +@pytest.mark.parametrize( + ("changed_files", "expected_role", "expected_kind"), + [ + (lambda: valid_files(extra_users=1), "students", "student_add"), + (_teacher_change_files, "teachers", "teacher_add"), + (_metadata_change_files, None, "course_update"), + ], + ids=("students", "teachers", "metadata"), +) +@pytest.mark.asyncio +async def test_changed_scope_requests_only_its_live_evidence( + tmp_path: Path, + changed_files, + expected_role: str | None, + expected_kind: str, +): + service, connector, _first_id, _first = await _baseline(tmp_path) + second = service.upload(zip_bytes(changed_files())) + connector.reset_course_reads() + + plan = await service.build_live_plan(connector, second.id) + + if expected_role is None: + assert connector.metadata_calls == [("Section_101",)] + assert connector.roster_calls == [] + else: + assert connector.metadata_calls == [] + assert connector.roster_calls == [(('1000',), expected_role)] + assert expected_kind in {action.kind for action in plan.actions} + + +@pytest.mark.asyncio +async def test_dirty_student_scope_is_reread_without_other_live_scopes(tmp_path: Path): + service, connector, import_id, _first = await _baseline(tmp_path) + service.store.mark_managed_course_dirty( + (ManagedCourseDirty("Section_101", students=True),), + error_code="TEST-INCOMPLETE", + ) + connector.reset_course_reads() + + await service.build_live_plan(connector, import_id) + + assert connector.metadata_calls == [] + assert connector.roster_calls == [(('1000',), "students")] + state = service.store.get_managed_course_states(["Section_101"])["section_101"] + assert not state.student_roster_dirty + + +@pytest.mark.asyncio +async def test_removed_managed_alias_gets_fresh_metadata_read(tmp_path: Path): + service, connector, first_id, _first = await _baseline(tmp_path) + service.mark_accepted(first_id) + files = valid_files() + files["classes.csv"] = files["classes.csv"].replace( + "101,active,Algebra Section", + "202,active,Algebra Section", + ) + files["enrollments.csv"] = files["enrollments.csv"].replace( + ",101,school-1,", + ",202,school-1,", + ) + second = service.upload(zip_bytes(files)) + connector.reset_course_reads() + + plan = await service.build_live_plan(connector, second.id) + + requested_aliases = { + alias.casefold() + for call in connector.metadata_calls + for alias in call + } + assert "section_101" in requested_aliases + assert connector.roster_calls == [] + assert [(action.kind, action.subject) for action in plan.archive_actions] == [ + ("course_archive", "Section_101") + ] + + +@pytest.mark.asyncio +async def test_ten_thousand_unrelated_tenant_courses_add_zero_managed_reads( + tmp_path: Path, +): + service, connector, import_id, _first = await _baseline(tmp_path) + connector.unrelated_courses = { + f"Unmanaged_{index}": CourseDetail( + id=f"unrelated-{index}", + name="Unmanaged", + aliases=(f"d:Unmanaged_{index}",), + ) + for index in range(10_000) + } + assert len(connector.unrelated_courses) == 10_000 + connector.reset_course_reads() + + await service.build_live_plan(connector, import_id) + + assert connector.metadata_calls == [] + assert connector.roster_calls == [] + + +async def _student_execution_setup(tmp_path: Path): + service, connector, _first_id, _first = await _baseline(tmp_path) + second = service.upload(zip_bytes(valid_files(extra_users=1))) + connector.reset_course_reads() + planning = await service.build_live_plan(connector, second.id) + assert [action.kind for action in planning.actions] == ["student_add"] + manifest = service.persist_live_plan(planning).ordinary + executor = OneRosterExecutor( + service.store, + connector, + stabilization_delays=(0.0, 0.0), + ) + service.store.confirm_manifest(manifest.id, manifest.import_id) + service.store.claim_manifest( + manifest.id, + owner_id=executor._operation_owner, + owner_identity=executor._operation_identity, + ) + return service, connector, planning, manifest, executor + + +@pytest.mark.asyncio +async def test_executor_verification_updates_only_student_scope(tmp_path: Path): + service, connector, planning, manifest, executor = await _student_execution_setup( + tmp_path + ) + before = service.store.get_managed_course_states(["Section_101"])["section_101"] + connector.reset_course_reads() + + applied, failed = await executor._apply_chunk( + manifest.id, + manifest.actions, + planning.owner_ids, + set(), + ) + + after = service.store.get_managed_course_states(["Section_101"])["section_101"] + assert (applied, failed) == (1, 0) + assert connector.roster_calls == [(('1000',), "students")] + assert after.verified_student_hash == after.desired_student_hash + assert after.verified_student_hash != before.verified_student_hash + assert after.verified_metadata_hash == before.verified_metadata_hash + assert after.verified_teacher_hash == before.verified_teacher_hash + assert after.last_metadata_verified_at == before.last_metadata_verified_at + assert after.last_teacher_verified_at == before.last_teacher_verified_at + assert not after.student_roster_dirty + assert not after.recovery_required + + +@pytest.mark.asyncio +async def test_incomplete_executor_verification_dirties_only_student_scope( + tmp_path: Path, +): + service, connector, planning, manifest, executor = await _student_execution_setup( + tmp_path + ) + before = service.store.get_managed_course_states(["Section_101"])["section_101"] + connector.incomplete_rosters = True + connector.reset_course_reads() + + applied, failed = await executor._apply_chunk( + manifest.id, + manifest.actions, + planning.owner_ids, + set(), + ) + + after = service.store.get_managed_course_states(["Section_101"])["section_101"] + assert (applied, failed) == (0, 1) + assert connector.roster_calls == [ + (('1000',), "students"), + (('1000',), "students"), + (('1000',), "students"), + ] + assert after.student_roster_dirty + assert after.recovery_required + assert not after.metadata_dirty + assert not after.teacher_roster_dirty + assert after.verified_student_hash == before.verified_student_hash + assert after.verified_metadata_hash == before.verified_metadata_hash + assert after.verified_teacher_hash == before.verified_teacher_hash + + +def test_managed_state_migration_is_additive_and_idempotent(tmp_path: Path): + root = tmp_path / "legacy-component" + root.mkdir() + state_path = root / "state.db" + with sqlite3.connect(state_path) as conn: + conn.execute("CREATE TABLE legacy_marker (id INTEGER PRIMARY KEY)") + conn.execute("INSERT INTO legacy_marker(id) VALUES (1)") + + first = OneRosterStore("example.org", root) + second = OneRosterStore("example.org", root) + + assert first.state_path == second.state_path + with sqlite3.connect(state_path) as conn: + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ) + } + columns = { + row[1] + for row in conn.execute("PRAGMA table_info(managed_course_state)") + } + assert conn.execute("SELECT id FROM legacy_marker").fetchone() == (1,) + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] + assert {"managed_course_state", "managed_course_members"} <= tables + assert { + "desired_metadata_hash", + "verified_teacher_hash", + "student_roster_dirty", + "recovery_required", + } <= columns + + digest = "0" * 64 + with pytest.raises(OneRosterError) as invalid: + second.record_managed_course_desired( + ( + ManagedCourseDesired( + "Arbitrary_1", + "import-id", + digest, + digest, + digest, + ), + ) + ) + assert invalid.value.code == "OR-ALIAS-INVALID" + + +def test_semantic_roster_hashes_are_role_specific_and_normalized(): + assert teacher_hash([" Teacher@Example.org ", "teacher@example.org"]) == ( + teacher_hash(["teacher@example.org"]) + ) + assert student_hash(["student@example.org"]) != teacher_hash( + ["teacher@example.org"] + ) From f7e1a5b473c69564bd5c68a6e239918f1a165da3 Mon Sep 17 00:00:00 2001 From: Sykez <276981287+Sykezzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 04:01:49 -0500 Subject: [PATCH 2/3] perf(oneroster): chunk and resume managed live reads --- gamgui/components/oneroster/models.py | 20 + gamgui/components/oneroster/planner.py | 1214 ++++++++++++++++++++--- gamgui/core/connectors/gam_connector.py | 12 +- tests/test_oneroster_bulk_connector.py | 8 +- tests/test_oneroster_bulk_planner.py | 51 +- tests/test_oneroster_delta.py | 5 +- tests/test_oneroster_read_chunks.py | 556 +++++++++++ 7 files changed, 1691 insertions(+), 175 deletions(-) create mode 100644 tests/test_oneroster_read_chunks.py diff --git a/gamgui/components/oneroster/models.py b/gamgui/components/oneroster/models.py index 1a140c1..f4a4b7b 100644 --- a/gamgui/components/oneroster/models.py +++ b/gamgui/components/oneroster/models.py @@ -623,6 +623,26 @@ class PlanningPerformanceReceipt: cached_teacher_scopes: int = 0 cached_student_scopes: int = 0 audit_courses_requested: int = 0 + metadata_chunk_count: int = 0 + teacher_roster_chunk_count: int = 0 + student_roster_chunk_count: int = 0 + completed_chunk_count: int = 0 + retried_chunk_count: int = 0 + failed_chunk_count: int = 0 + rate_limit_count: int = 0 + timeout_retry_count: int = 0 + incomplete_coverage_count: int = 0 + latency_regression_count: int = 0 + largest_metadata_chunk: int = 0 + largest_roster_chunk: int = 0 + maximum_observed_read_concurrency: int = 0 + final_recommended_read_concurrency: int = 1 + metadata_final_read_concurrency: int = 1 + teacher_final_read_concurrency: int = 1 + student_final_read_concurrency: int = 1 + metadata_chunk_worker_levels: Tuple[int, ...] = () + teacher_roster_chunk_worker_levels: Tuple[int, ...] = () + student_roster_chunk_worker_levels: Tuple[int, ...] = () @dataclass(frozen=True) diff --git a/gamgui/components/oneroster/planner.py b/gamgui/components/oneroster/planner.py index 8fdb98e..f83aae5 100644 --- a/gamgui/components/oneroster/planner.py +++ b/gamgui/components/oneroster/planner.py @@ -11,15 +11,20 @@ import asyncio import hashlib import json +import random import sqlite3 import time from collections import Counter, defaultdict from contextlib import closing -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime -from typing import Any, Iterable, Mapping, Optional, Protocol, Sequence +from typing import Any, Awaitable, Callable, Iterable, Mapping, Optional, Protocol, Sequence from gamgui.core.classroom.models import CourseRosterSnapshot +from gamgui.core.connectors.gam_connector import ( + ONEROSTER_MANAGED_ALIAS_CHUNK_CAP, + ONEROSTER_ROSTER_CHUNK_CAP, +) from gamgui.core.gam.errors import GAMError, GAMErrorKind from .ingest import district_roster_date @@ -45,6 +50,13 @@ DIRECTORY_CONCURRENCY = 12 COURSE_CONCURRENCY = 8 DEFAULT_AUDIT_SAMPLE_SIZE = 25 +ONEROSTER_READ_MAX_CONCURRENCY = 4 +ONEROSTER_READ_MAX_ATTEMPTS = 3 +ONEROSTER_READ_BACKOFF_BASE_SECONDS = 1.0 +ONEROSTER_READ_BACKOFF_CAP_SECONDS = 8.0 +ONEROSTER_READ_BACKOFF_JITTER_RATIO = 0.25 +ONEROSTER_READ_LATENCY_REGRESSION_FACTOR = 2.0 +ONEROSTER_READ_LATENCY_REGRESSION_FLOOR_SECONDS = 1.0 # The current planner uses compact in-memory participant/action models. Keep # district planning comfortably below the measured memory cliff; larger valid # snapshots remain available for inspection and CSV export. @@ -124,6 +136,109 @@ class _ActionPayload: live_evidence: Mapping[str, Any] +class _IncompleteReadCoverage(ValueError): + """A bounded read returned data that cannot prove its requested scope.""" + + +@dataclass(frozen=True, slots=True) +class _CompletedReadChunk: + value: Any + duration_seconds: float + clean: bool + + +@dataclass(slots=True) +class _ReadTuner: + maximum: int + level: int = 1 + clean_streak: int = 0 + baseline_seconds: Optional[float] = None + + def penalize(self) -> None: + self.level = max(1, self.level - 1) + self.clean_streak = 0 + + def completed_cleanly(self, duration_seconds: float) -> bool: + duration = max(0.0, float(duration_seconds)) + regressed = bool( + self.baseline_seconds is not None + and duration + > max( + ONEROSTER_READ_LATENCY_REGRESSION_FLOOR_SECONDS, + self.baseline_seconds * ONEROSTER_READ_LATENCY_REGRESSION_FACTOR, + ) + ) + if regressed: + self.penalize() + else: + self.clean_streak += 1 + if self.clean_streak >= 2 and self.level < self.maximum: + self.level += 1 + self.clean_streak = 0 + self.baseline_seconds = ( + duration + if self.baseline_seconds is None + else (self.baseline_seconds * 0.75) + (duration * 0.25) + ) + return regressed + + +@dataclass(slots=True) +class _ReadPerformance: + metadata_chunk_count: int = 0 + teacher_roster_chunk_count: int = 0 + student_roster_chunk_count: int = 0 + completed_chunk_count: int = 0 + retried_chunk_count: int = 0 + failed_chunk_count: int = 0 + rate_limit_count: int = 0 + timeout_retry_count: int = 0 + incomplete_coverage_count: int = 0 + latency_regression_count: int = 0 + largest_metadata_chunk: int = 0 + largest_roster_chunk: int = 0 + active_chunks: int = 0 + maximum_observed_read_concurrency: int = 0 + metadata_final_read_concurrency: int = 1 + teacher_final_read_concurrency: int = 1 + student_final_read_concurrency: int = 1 + metadata_chunk_worker_levels: list[int] = field(default_factory=list) + teacher_roster_chunk_worker_levels: list[int] = field(default_factory=list) + student_roster_chunk_worker_levels: list[int] = field(default_factory=list) + + def scheduled(self, category: str, size: int, worker_level: int) -> None: + if category == "metadata": + self.metadata_chunk_count += 1 + self.largest_metadata_chunk = max(self.largest_metadata_chunk, size) + self.metadata_chunk_worker_levels.append(worker_level) + elif category == "teachers": + self.teacher_roster_chunk_count += 1 + self.largest_roster_chunk = max(self.largest_roster_chunk, size) + self.teacher_roster_chunk_worker_levels.append(worker_level) + else: + self.student_roster_chunk_count += 1 + self.largest_roster_chunk = max(self.largest_roster_chunk, size) + self.student_roster_chunk_worker_levels.append(worker_level) + + def set_final_level(self, category: str, level: int) -> None: + if category == "metadata": + self.metadata_final_read_concurrency = level + elif category == "teachers": + self.teacher_final_read_concurrency = level + else: + self.student_final_read_concurrency = level + + def final_recommendation(self) -> int: + levels: list[int] = [] + if self.metadata_chunk_count: + levels.append(self.metadata_final_read_concurrency) + if self.teacher_roster_chunk_count: + levels.append(self.teacher_final_read_concurrency) + if self.student_roster_chunk_count: + levels.append(self.student_final_read_concurrency) + return min(levels) if levels else 1 + + def planner_configuration_hash( *, limited_import: bool, @@ -153,6 +268,12 @@ def __init__( directory_concurrency: int = DIRECTORY_CONCURRENCY, course_concurrency: int = COURSE_CONCURRENCY, audit_sample_size: Optional[int] = None, + read_max_concurrency: Optional[int] = None, + read_max_attempts: Optional[int] = None, + read_backoff_base_seconds: Optional[float] = None, + read_backoff_cap_seconds: Optional[float] = None, + read_sleep: Optional[Callable[[float], Awaitable[None]]] = None, + read_jitter: Optional[Callable[[], float]] = None, ) -> None: self.store = store self.connector = connector @@ -164,6 +285,67 @@ def __init__( else audit_sample_size ) self.audit_sample_size = max(0, min(int(configured_audit), 25)) + configured_read_concurrency = ( + getattr( + connector, + "oneroster_read_max_concurrency", + ONEROSTER_READ_MAX_CONCURRENCY, + ) + if read_max_concurrency is None + else read_max_concurrency + ) + configured_read_attempts = ( + getattr( + connector, + "oneroster_read_max_attempts", + ONEROSTER_READ_MAX_ATTEMPTS, + ) + if read_max_attempts is None + else read_max_attempts + ) + configured_backoff_base = ( + getattr( + connector, + "oneroster_read_backoff_base_seconds", + ONEROSTER_READ_BACKOFF_BASE_SECONDS, + ) + if read_backoff_base_seconds is None + else read_backoff_base_seconds + ) + configured_backoff_cap = ( + getattr( + connector, + "oneroster_read_backoff_cap_seconds", + ONEROSTER_READ_BACKOFF_CAP_SECONDS, + ) + if read_backoff_cap_seconds is None + else read_backoff_cap_seconds + ) + self.read_max_concurrency = max( + 1, + min(int(configured_read_concurrency), ONEROSTER_READ_MAX_CONCURRENCY), + ) + self.read_max_attempts = max(1, min(int(configured_read_attempts), 6)) + self.read_backoff_base_seconds = max( + 0.0, + min(float(configured_backoff_base), 60.0), + ) + self.read_backoff_cap_seconds = max( + self.read_backoff_base_seconds, + min(float(configured_backoff_cap), 120.0), + ) + self._read_sleep = read_sleep or getattr( + connector, + "oneroster_read_sleep", + asyncio.sleep, + ) + self._read_jitter = read_jitter or getattr( + connector, + "oneroster_read_jitter", + random.random, + ) + self._read_performance = _ReadPerformance() + self._checkpointed_scopes: set[tuple[str, str]] = set() async def plan( self, @@ -174,6 +356,8 @@ async def plan( required_metadata_aliases: Sequence[str] = (), ) -> LivePlanningResult: planning_started = time.perf_counter() + self._read_performance = _ReadPerformance() + self._checkpointed_scopes = set() snapshot = self.store.refresh_schedule_scope( import_id, today=district_roster_date(now), @@ -224,30 +408,34 @@ async def plan( or state.metadata_dirty or state.recovery_required ) - async def timed_directory_snapshot() -> tuple[Optional[dict[str, Any]], float]: started = time.perf_counter() value = await self._read_directory_snapshot() return value, time.perf_counter() - started - async def timed_classroom_snapshot( - aliases: Sequence[str], - ) -> tuple[Optional[dict[str, Any]], float]: + async def timed_bootstrap_snapshot() -> tuple[Optional[dict[str, Any]], float]: started = time.perf_counter() - value = await self._read_managed_course_snapshot(aliases) + value = await self._read_managed_course_snapshot( + bootstrap_metadata_aliases + ) return value, time.perf_counter() - started - if bootstrap_metadata_aliases: + bootstrap_read_concurrently = bool( + 0 < len(bootstrap_metadata_aliases) <= ONEROSTER_MANAGED_ALIAS_CHUNK_CAP + ) + if bootstrap_read_concurrently: directory_result, bootstrap_result = await asyncio.gather( timed_directory_snapshot(), - timed_classroom_snapshot(bootstrap_metadata_aliases), + timed_bootstrap_snapshot(), ) - bootstrap_courses, classroom_snapshot_seconds = bootstrap_result + directory_snapshot, directory_snapshot_seconds = directory_result + bootstrap_courses, bootstrap_snapshot_seconds = bootstrap_result else: - directory_result = await timed_directory_snapshot() + directory_snapshot, directory_snapshot_seconds = ( + await timed_directory_snapshot() + ) bootstrap_courses = {} - classroom_snapshot_seconds = 0.0 - directory_snapshot, directory_snapshot_seconds = directory_result + bootstrap_snapshot_seconds = 0.0 resolved, resolution_issues = await self._resolve_directory( desired, directory_snapshot, @@ -321,64 +509,106 @@ async def timed_classroom_snapshot( if (intent := read_plan.get(alias.casefold())) is not None and intent.metadata ) - try: - remaining_metadata_aliases = tuple( - alias - for alias in metadata_aliases - if alias.casefold() - not in {item.casefold() for item in bootstrap_metadata_aliases} - ) - managed_courses = bootstrap_courses - if remaining_metadata_aliases and managed_courses is not None: - additional_courses, additional_seconds = ( - await timed_classroom_snapshot(remaining_metadata_aliases) - ) - classroom_snapshot_seconds += additional_seconds - if additional_courses is None: - managed_courses = None - else: - managed_courses = {**managed_courses, **additional_courses} - roster_started = time.perf_counter() - live_courses = await self._read_live_courses( - eligible, - managed_courses, + eligible_by_alias = {course.alias.casefold(): course for course in eligible} + aliases_by_key = {alias.casefold(): alias for alias in relevant_aliases} + known_course_ids: dict[str, str] = {} + if bootstrap_read_concurrently and bootstrap_courses is not None: + await self._persist_metadata_chunk( + import_id, + bootstrap_metadata_aliases, + bootstrap_courses, + eligible_by_alias, directory_snapshot, - read_plan=read_plan, - managed_states=managed_states, - cached_members=cached_members, + required_metadata_keys, ) - live_courses, live_resolution_issues = await self._resolve_live_participants( - live_courses, - resolved, - directory_snapshot, + known_course_ids.update( + (_text(detail, "id"), key) + for key, detail in bootstrap_courses.items() + if _text(detail, "id") ) - issues.extend(live_resolution_issues) - roster_snapshot_seconds = time.perf_counter() - roster_started - self._persist_live_verification( + + async def checkpoint_metadata( + chunk: tuple[str, ...], + indexed: Mapping[str, Any], + ) -> Mapping[str, Any]: + for key, detail in indexed.items(): + course_id = _text(detail, "id") + previous_key = known_course_ids.get(course_id) if course_id else None + if previous_key is not None and previous_key != key: + self.store.mark_managed_course_dirty( + tuple( + ManagedCourseDirty(aliases_by_key[item], metadata=True) + for item in (previous_key, key) + ), + error_code="OR-ALIAS-COLLISION", + recovery_required=True, + ) + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one managed alias resolves to the same live Classroom course.", + ) + await self._persist_metadata_chunk( import_id, - eligible, - live_courses, - read_plan, + chunk, + indexed, + eligible_by_alias, + directory_snapshot, + required_metadata_keys, ) - except BaseException: - dirty = tuple( - ManagedCourseDirty( - alias=alias, - metadata=intent.metadata, - teachers=intent.teachers, - students=intent.students, - ) - for alias, intent in ( - (course.alias, read_plan.get(course.alias.casefold())) - for course in eligible - ) - if intent is not None + known_course_ids.update( + (_text(detail, "id"), key) + for key, detail in indexed.items() + if _text(detail, "id") ) - self.store.mark_managed_course_dirty( - dirty, - error_code="OR-LIVE-READ-INCOMPLETE", + return indexed + + bootstrap_keys = { + alias.casefold() for alias in bootstrap_metadata_aliases + } if bootstrap_read_concurrently else set() + remaining_metadata_aliases = tuple( + alias + for alias in metadata_aliases + if alias.casefold() not in bootstrap_keys + ) + managed_courses = bootstrap_courses + classroom_snapshot_seconds = bootstrap_snapshot_seconds + if remaining_metadata_aliases and managed_courses is not None: + classroom_started = time.perf_counter() + additional_courses = await self._read_managed_course_snapshot( + remaining_metadata_aliases, + on_chunk=checkpoint_metadata, ) - raise + classroom_snapshot_seconds += time.perf_counter() - classroom_started + if additional_courses is None: + managed_courses = None + else: + managed_courses = {**managed_courses, **additional_courses} + roster_started = time.perf_counter() + live_courses = await self._read_live_courses( + import_id, + eligible, + managed_courses, + directory_snapshot, + read_plan=read_plan, + managed_states=managed_states, + cached_members=cached_members, + ) + live_courses, live_resolution_issues = await self._resolve_live_participants( + live_courses, + resolved, + directory_snapshot, + ) + issues.extend(live_resolution_issues) + roster_snapshot_seconds = time.perf_counter() - roster_started + # Optional legacy connectors without exact bulk capabilities checkpoint + # their bounded fallback reads here. Normal GAM bulk scopes were already + # committed chunk-by-chunk and are skipped by this final compatibility pass. + self._persist_live_verification( + import_id, + eligible, + live_courses, + read_plan, + ) archive_actions, archive_basis, archive_issues = await self._archive_actions( import_id, @@ -469,6 +699,50 @@ async def timed_classroom_snapshot( for course in eligible ), audit_courses_requested=len(audit_aliases), + metadata_chunk_count=self._read_performance.metadata_chunk_count, + teacher_roster_chunk_count=( + self._read_performance.teacher_roster_chunk_count + ), + student_roster_chunk_count=( + self._read_performance.student_roster_chunk_count + ), + completed_chunk_count=self._read_performance.completed_chunk_count, + retried_chunk_count=self._read_performance.retried_chunk_count, + failed_chunk_count=self._read_performance.failed_chunk_count, + rate_limit_count=self._read_performance.rate_limit_count, + timeout_retry_count=self._read_performance.timeout_retry_count, + incomplete_coverage_count=( + self._read_performance.incomplete_coverage_count + ), + latency_regression_count=( + self._read_performance.latency_regression_count + ), + largest_metadata_chunk=self._read_performance.largest_metadata_chunk, + largest_roster_chunk=self._read_performance.largest_roster_chunk, + maximum_observed_read_concurrency=( + self._read_performance.maximum_observed_read_concurrency + ), + final_recommended_read_concurrency=( + self._read_performance.final_recommendation() + ), + metadata_final_read_concurrency=( + self._read_performance.metadata_final_read_concurrency + ), + teacher_final_read_concurrency=( + self._read_performance.teacher_final_read_concurrency + ), + student_final_read_concurrency=( + self._read_performance.student_final_read_concurrency + ), + metadata_chunk_worker_levels=tuple( + self._read_performance.metadata_chunk_worker_levels + ), + teacher_roster_chunk_worker_levels=tuple( + self._read_performance.teacher_roster_chunk_worker_levels + ), + student_roster_chunk_worker_levels=tuple( + self._read_performance.student_roster_chunk_worker_levels + ), ), desired_course_hashes=desired_states, ) @@ -490,27 +764,367 @@ async def _read_directory_snapshot(self) -> Optional[dict[str, Any]]: async def _read_managed_course_snapshot( self, aliases: Sequence[str], + *, + on_chunk: Optional[ + Callable[ + [tuple[str, ...], Mapping[str, Any]], + Awaitable[Mapping[str, Any]], + ] + ] = None, ) -> Optional[dict[str, Any]]: - if not aliases: + selected_by_key: dict[str, str] = {} + for raw_alias in aliases: + alias = str(raw_alias or "").strip() + if alias.startswith("d:"): + alias = alias[2:] + key = _managed_alias_key(alias) + if not key: + raise OneRosterError( + "OR-ALIAS-INVALID", + "A managed Classroom alias must use the exact Section_ form.", + ) + selected_by_key.setdefault(key, alias) + selected = tuple(sorted(selected_by_key.values(), key=str.casefold)) + if not selected: return {} bulk = getattr(self.connector, "list_oneroster_managed_courses", None) if not callable(bulk): return None + chunks = tuple( + tuple(selected[offset : offset + ONEROSTER_MANAGED_ALIAS_CHUNK_CAP]) + for offset in range(0, len(selected), ONEROSTER_MANAGED_ALIAS_CHUNK_CAP) + ) + merged: dict[str, Any] = {} + course_ids: dict[str, str] = {} + + async def read_chunk(chunk: tuple[str, ...]) -> Mapping[str, Any]: + raw = await bulk(chunk) + return _index_managed_courses(raw, requested_aliases=chunk) + + async def accept_chunk( + chunk: tuple[str, ...], + indexed: Mapping[str, Any], + ) -> None: + for alias_key, detail in indexed.items(): + course_id = _text(detail, "id") + previous_alias = course_ids.get(course_id) if course_id else None + if previous_alias is not None and previous_alias != alias_key: + affected = tuple( + ManagedCourseDirty(alias, metadata=True) + for alias in ( + selected_by_key[previous_alias], + selected_by_key[alias_key], + ) + ) + self.store.mark_managed_course_dirty( + affected, + error_code="OR-ALIAS-COLLISION", + recovery_required=True, + ) + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one managed alias resolves to the same live Classroom course.", + ) + if course_id: + course_ids[course_id] = alias_key + if on_chunk is not None: + await on_chunk(chunk, indexed) + merged.update(indexed) + + async def fail_chunk(chunk: tuple[str, ...], error_code: str) -> None: + self.store.mark_managed_course_dirty( + tuple(ManagedCourseDirty(alias, metadata=True) for alias in chunk), + error_code=error_code, + ) + + await self._coordinate_read_chunks( + chunks, + category="metadata", + read_chunk=read_chunk, + accept_chunk=accept_chunk, + fail_chunk=fail_chunk, + ) + return merged + + async def _coordinate_read_chunks( + self, + chunks: Sequence[tuple[str, ...]], + *, + category: str, + read_chunk: Callable[[tuple[str, ...]], Awaitable[Any]], + accept_chunk: Callable[[tuple[str, ...], Any], Awaitable[None]], + fail_chunk: Callable[[tuple[str, ...], str], Awaitable[None]], + ) -> None: + tuner = _ReadTuner(self.read_max_concurrency) + offset = 0 try: - raw = await bulk(aliases) - return _index_managed_courses(raw) - except GAMError: - # The exact-set bulk command is an optimization. Fall back to - # bounded per-alias reads, which distinguish absent new courses - # from genuine authentication and permission failures. - return None - except OneRosterError: - raise - except Exception as exc: - raise OneRosterError( + while offset < len(chunks): + worker_level = tuner.level + wave = tuple(chunks[offset : offset + worker_level]) + for chunk in wave: + self._read_performance.scheduled( + category, + len(chunk), + worker_level, + ) + outcomes = await asyncio.gather( + *( + self._attempt_bounded_read(chunk, read_chunk, tuner) + for chunk in wave + ), + return_exceptions=True, + ) + first_failure: Optional[tuple[OneRosterError, BaseException]] = None + for chunk, outcome in zip(wave, outcomes): + if isinstance(outcome, asyncio.CancelledError): + raise outcome + failure: Optional[BaseException] = ( + outcome if isinstance(outcome, BaseException) else None + ) + completed: Optional[_CompletedReadChunk] = ( + outcome if isinstance(outcome, _CompletedReadChunk) else None + ) + if completed is not None: + try: + await accept_chunk(chunk, completed.value) + except BaseException as exc: + if isinstance(exc, asyncio.CancelledError): + raise + failure = exc + if isinstance(exc, _IncompleteReadCoverage): + self._read_performance.incomplete_coverage_count += 1 + tuner.penalize() + if failure is not None: + self._read_performance.failed_chunk_count += 1 + safe_error = self._safe_read_error(failure, category) + await fail_chunk(chunk, safe_error.code) + if first_failure is None: + first_failure = (safe_error, failure) + continue + assert completed is not None + self._read_performance.completed_chunk_count += 1 + if completed.clean and tuner.completed_cleanly( + completed.duration_seconds + ): + self._read_performance.latency_regression_count += 1 + if first_failure is not None: + safe_error, cause = first_failure + if safe_error is cause: + raise safe_error + raise safe_error from cause + offset += len(wave) + finally: + self._read_performance.set_final_level(category, tuner.level) + + async def _attempt_bounded_read( + self, + chunk: tuple[str, ...], + read_chunk: Callable[[tuple[str, ...]], Awaitable[Any]], + tuner: _ReadTuner, + ) -> _CompletedReadChunk: + self._read_performance.active_chunks += 1 + self._read_performance.maximum_observed_read_concurrency = max( + self._read_performance.maximum_observed_read_concurrency, + self._read_performance.active_chunks, + ) + retried = False + try: + for attempt in range(self.read_max_attempts): + started = time.perf_counter() + try: + value = await read_chunk(chunk) + except _IncompleteReadCoverage: + self._read_performance.incomplete_coverage_count += 1 + tuner.penalize() + raise + except GAMError as exc: + retryable = exc.kind in { + GAMErrorKind.RATE_LIMITED, + GAMErrorKind.TIMEOUT, + } + if not retryable: + raise + tuner.penalize() + if exc.kind is GAMErrorKind.RATE_LIMITED: + self._read_performance.rate_limit_count += 1 + if attempt + 1 >= self.read_max_attempts: + raise + if exc.kind is GAMErrorKind.TIMEOUT: + self._read_performance.timeout_retry_count += 1 + if not retried: + retried = True + self._read_performance.retried_chunk_count += 1 + await self._read_sleep(self._read_backoff_delay(attempt)) + continue + return _CompletedReadChunk( + value=value, + duration_seconds=time.perf_counter() - started, + clean=not retried, + ) + raise RuntimeError("bounded OneRoster read exhausted unexpectedly") + finally: + self._read_performance.active_chunks -= 1 + + def _read_backoff_delay(self, retry_index: int) -> float: + jitter = max(0.0, min(float(self._read_jitter()), 1.0)) + exponential = self.read_backoff_base_seconds * (2 ** max(0, retry_index)) + return min( + self.read_backoff_cap_seconds, + exponential * (1.0 + (ONEROSTER_READ_BACKOFF_JITTER_RATIO * jitter)), + ) + + @staticmethod + def _safe_read_error(error: BaseException, category: str) -> OneRosterError: + if isinstance(error, OneRosterError): + return error + if isinstance(error, GAMError): + if error.kind is GAMErrorKind.RATE_LIMITED: + return OneRosterError( + "OR-ONEROSTER-READ-RATE-LIMITED", + "Google continued to rate-limit a bounded OneRoster read; completed chunks were preserved.", + ) + if error.kind is GAMErrorKind.TIMEOUT: + return OneRosterError( + "OR-ONEROSTER-READ-TIMEOUT", + "A bounded OneRoster read timed out after finite retries; completed chunks were preserved.", + ) + if error.kind in { + GAMErrorKind.AUTH_EXPIRED, + GAMErrorKind.NOT_AUTHENTICATED, + }: + return OneRosterError( + "OR-ONEROSTER-READ-AUTH", + "OneRoster live reads require renewed Google authentication.", + ) + if error.kind in { + GAMErrorKind.PERMISSION_DENIED, + GAMErrorKind.SCOPE_MISSING, + }: + return OneRosterError( + "OR-ONEROSTER-READ-PERMISSION", + "The authorized account cannot safely complete OneRoster live reads.", + ) + if category == "metadata": + return OneRosterError( "OR-CLASSROOM-READ", "Live Classroom course resolution failed; no import plan was created.", + ) + return OneRosterError( + "OR-CLASSROOM-ROSTER-READ", + "Live Classroom rosters could not be read safely.", + ) + + async def _persist_metadata_chunk( + self, + import_id: str, + chunk: Sequence[str], + indexed: Mapping[str, Any], + courses_by_alias: Mapping[str, _DesiredCourse], + directory_snapshot: Optional[Mapping[str, Any]], + manifest_guard_aliases: set[str], + ) -> None: + states = self.store.get_managed_course_states(chunk) + aliases_by_key = { + _managed_alias_key(alias): str(alias).removeprefix("d:") + for alias in chunk + } + verifications: list[ManagedCourseVerification] = [] + verified_keys: list[str] = [] + for key, detail in indexed.items(): + alias = aliases_by_key.get(key, "") + state = states.get(key) + if state is None and key not in courses_by_alias: + # A pre-registry accepted alias can still supply fresh archive + # evidence, but there is no durable Branch 1 row to checkpoint. + continue + if not alias or state is None: + raise _IncompleteReadCoverage( + "A returned managed course has no registered exact alias." + ) + course_id = _text(detail, "id") + if state.course_id and state.course_id != course_id: + self.store.mark_managed_course_dirty( + (ManagedCourseDirty(alias, metadata=True),), + error_code="OR-MANAGED-COURSE-ID-DRIFT", + recovery_required=True, + ) + if key in manifest_guard_aliases: + # Executor/gate revalidation must retain the fresh live basis + # so its manifest comparison can persist an exact drift report. + continue + raise OneRosterError( + "OR-MANAGED-COURSE-ID-DRIFT", + "The exact managed alias resolved to a different Classroom course.", + ) + owner_email = await self._normalize_live_identity( + _text(detail, "owner_email", "ownerEmail"), + _text(detail, "owner_id", "ownerId"), + directory_snapshot, + ) + course_state = _text(detail, "course_state", "courseState").upper() + if not ( + course_id + and _has_exact_alias(detail, alias) + and owner_email + and course_state + ): + raise _IncompleteReadCoverage( + "A managed-course metadata chunk returned incomplete exact coverage." + ) + desired_course = courses_by_alias.get(key) + verifications.append( + ManagedCourseVerification( + alias=alias, + course_id=course_id, + source_class_id=( + desired_course.class_id + if desired_course is not None + else state.source_class_id + ), + last_seen_import_id=import_id, + metadata_hash=metadata_hash( + alias, + _text(detail, "name"), + owner_email, + _text(detail, "room"), + _text(detail, "section"), + course_state, + ), + course_state=course_state, + ) + ) + verified_keys.append(key) + self.store.record_managed_course_verifications(tuple(verifications)) + self._checkpointed_scopes.update( + (key, "metadata") for key in verified_keys + ) + + async def _normalize_live_identity( + self, + email: str, + user_id: str, + directory_snapshot: Optional[Mapping[str, Any]], + ) -> str: + email_key = _directory_identifier(email) + id_key = _directory_id_key(user_id) + if directory_snapshot is not None: + return _snapshot_primary( + directory_snapshot.get(email_key) + or directory_snapshot.get(id_key) + ) + reference = email_key or str(user_id or "").strip() + if not reference: + return "" + try: + user = await self.connector.get_user(reference) + except Exception as exc: + if _is_not_found(exc): + return "" + raise OneRosterError( + "OR-DIRECTORY-READ", + "Live Directory resolution failed; no import plan was created.", ) from exc + return _snapshot_primary(user) async def _resolve_directory( self, @@ -554,6 +1168,7 @@ async def resolve(email: str) -> tuple[str, str, str]: async def _read_live_courses( self, + import_id: str, courses: Sequence[_DesiredCourse], managed_courses: Optional[Mapping[str, Any]] = None, directory_snapshot: Optional[Mapping[str, Any]] = None, @@ -613,7 +1228,13 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: return course.alias.casefold(), detail details = dict(await asyncio.gather(*(read(course) for course in courses))) - rosters = await self._read_rosters(details, intents) + rosters = await self._read_rosters( + import_id, + courses, + details, + intents, + directory_snapshot, + ) result: dict[str, Optional[_LiveCourse]] = {} for course in courses: key = course.alias.casefold() @@ -623,7 +1244,10 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: continue course_id = _text(detail, "id") cached_teachers, cached_students = retained_members.get(key, ((), ())) - read_teachers, read_students = rosters.get(course_id, (None, None)) + read_teachers, read_students, unresolved_members = rosters.get( + course_id, + (None, None, ()), + ) teachers = ( tuple(read_teachers) if read_teachers is not None @@ -652,6 +1276,7 @@ async def read(course: _DesiredCourse) -> tuple[str, Optional[Any]]: teachers, students, owner_email, + unresolved_members=tuple(unresolved_members), metadata_loaded=bool( (intent := intents.get(key)) and intent.metadata ), @@ -710,7 +1335,8 @@ async def resolve(email: str) -> tuple[str, str]: continue unresolved = tuple( sorted( - { + set(live.unresolved_members) + | { email for email in (*live.teachers, *live.students) if not mapping.get(email) @@ -743,70 +1369,190 @@ async def resolve(email: str) -> tuple[str, str]: async def _read_rosters( self, + import_id: str, + courses: Sequence[_DesiredCourse], details: Mapping[str, Optional[Any]], read_plan: Mapping[str, _CourseReadIntent], + directory_snapshot: Optional[Mapping[str, Any]], ) -> dict[ str, - tuple[Optional[tuple[str, ...]], Optional[tuple[str, ...]]], + tuple[ + Optional[tuple[str, ...]], + Optional[tuple[str, ...]], + tuple[str, ...], + ], ]: - requested: dict[str, tuple[bool, bool]] = {} + courses_by_alias = {course.alias.casefold(): course for course in courses} + manifest_guard_aliases = { + alias + for alias, intent in read_plan.items() + if "execution_identity_guard" in intent.reasons + } + requested: dict[ + str, + tuple[str, _DesiredCourse, Any, bool, bool], + ] = {} for alias, detail in details.items(): intent = read_plan.get(alias) course_id = _text(detail, "id") if detail is not None else "" if not intent or not course_id or not (intent.teachers or intent.students): continue - requested[course_id] = (intent.teachers, intent.students) + course = courses_by_alias.get(alias) + if course is None: + continue + previous = requested.get(course_id) + if previous is not None and previous[0] != alias: + self.store.mark_managed_course_dirty( + ( + ManagedCourseDirty( + previous[1].alias, + teachers=previous[3], + students=previous[4], + ), + ManagedCourseDirty( + course.alias, + teachers=intent.teachers, + students=intent.students, + ), + ), + error_code="OR-ALIAS-COLLISION", + recovery_required=True, + ) + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one managed alias resolves to the same live Classroom course.", + ) + requested[course_id] = ( + alias, + course, + detail, + intent.teachers, + intent.students, + ) if not requested: return {} - both_ids = sorted( - course_id - for course_id, (teachers, students) in requested.items() - if teachers and students - ) teacher_ids = sorted( course_id - for course_id, (teachers, students) in requested.items() - if teachers and not students + for course_id, (_alias, _course, _detail, teachers, _students) in requested.items() + if teachers ) student_ids = sorted( course_id - for course_id, (teachers, students) in requested.items() - if students and not teachers + for course_id, (_alias, _course, _detail, _teachers, students) in requested.items() + if students ) result: dict[ str, - tuple[Optional[tuple[str, ...]], Optional[tuple[str, ...]]], + tuple[ + Optional[tuple[str, ...]], + Optional[tuple[str, ...]], + tuple[str, ...], + ], ] = {} bulk = getattr(self.connector, "list_course_participants_many", None) if callable(bulk): for course_ids, role in ( - (both_ids, "all"), (teacher_ids, "teachers"), (student_ids, "students"), ): if not course_ids: continue - try: - participants = await bulk(course_ids, role) + chunks = tuple( + tuple(course_ids[offset : offset + ONEROSTER_ROSTER_CHUNK_CAP]) + for offset in range(0, len(course_ids), ONEROSTER_ROSTER_CHUNK_CAP) + ) + + async def read_chunk( + chunk: tuple[str, ...], + *, + selected_role: str = role, + ) -> Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]]: + participants = await bulk(chunk, selected_role) + requested_ids = frozenset(chunk) if ( not isinstance(participants, CourseRosterSnapshot) - or not participants.covers(course_ids) + or participants.seen_course_ids != requested_ids + or not set(participants.rosters) <= requested_ids ): - raise ValueError( - "The Classroom roster snapshot did not prove complete " - "coverage of the requested courses." + raise _IncompleteReadCoverage( + "The Classroom roster snapshot did not prove exact complete coverage." ) - except Exception as exc: - raise OneRosterError( - "OR-CLASSROOM-ROSTER-READ", - "Live Classroom rosters could not be read safely.", - ) from exc - for course_id in course_ids: - teachers, students = participants.for_course(course_id) - result[course_id] = ( - tuple(sorted(teachers)) if role in {"all", "teachers"} else None, - tuple(sorted(students)) if role in {"all", "students"} else None, + normalized: dict[ + str, + tuple[tuple[str, ...], tuple[str, ...]], + ] = {} + for course_id in chunk: + teachers, students = participants.for_course(course_id) + selected_members = ( + teachers if selected_role == "teachers" else students + ) + unexpected_members = ( + students if selected_role == "teachers" else teachers + ) + if unexpected_members: + raise _IncompleteReadCoverage( + "A role-specific Classroom roster read returned another role." + ) + normalized[course_id] = await self._normalize_roster_members( + selected_members, + directory_snapshot, + ) + return normalized + + async def accept_chunk( + chunk: tuple[str, ...], + normalized: Mapping[ + str, + tuple[tuple[str, ...], tuple[str, ...]], + ], + *, + selected_role: str = role, + ) -> None: + self._persist_roster_chunk( + import_id, + selected_role, + chunk, + normalized, + requested, + manifest_guard_aliases, ) + for course_id in chunk: + members, unresolved = normalized[course_id] + old_teachers, old_students, old_unresolved = result.get( + course_id, + (None, None, ()), + ) + result[course_id] = ( + members if selected_role == "teachers" else old_teachers, + members if selected_role == "students" else old_students, + tuple(sorted({*old_unresolved, *unresolved})), + ) + + async def fail_chunk( + chunk: tuple[str, ...], + error_code: str, + *, + selected_role: str = role, + ) -> None: + self.store.mark_managed_course_dirty( + tuple( + ManagedCourseDirty( + requested[course_id][1].alias, + teachers=selected_role == "teachers", + students=selected_role == "students", + ) + for course_id in chunk + ), + error_code=error_code, + ) + + await self._coordinate_read_chunks( + chunks, + category=role, + read_chunk=read_chunk, + accept_chunk=accept_chunk, + fail_chunk=fail_chunk, + ) return result semaphore = asyncio.Semaphore(self.course_concurrency) @@ -819,6 +1565,7 @@ async def read( str, Optional[tuple[str, ...]], Optional[tuple[str, ...]], + tuple[str, ...], ]: async with semaphore: try: @@ -835,26 +1582,168 @@ async def read( "OR-CLASSROOM-ROSTER-READ", "Live Classroom rosters could not be read safely.", ) from exc + normalized_teachers: Optional[tuple[str, ...]] = None + normalized_students: Optional[tuple[str, ...]] = None + unresolved: set[str] = set() + if teachers_needed: + normalized_teachers, missing = await self._normalize_roster_members( + tuple( + _participant_email(item) + for item in teachers + if _participant_email(item) + ), + directory_snapshot, + ) + unresolved.update(missing) + self._persist_roster_chunk( + import_id, + "teachers", + (course_id,), + {course_id: (normalized_teachers, missing)}, + requested, + manifest_guard_aliases, + ) + if students_needed: + normalized_students, missing = await self._normalize_roster_members( + tuple( + _participant_email(item) + for item in students + if _participant_email(item) + ), + directory_snapshot, + ) + unresolved.update(missing) + self._persist_roster_chunk( + import_id, + "students", + (course_id,), + {course_id: (normalized_students, missing)}, + requested, + manifest_guard_aliases, + ) return ( course_id, - tuple(sorted({_participant_email(item) for item in teachers if _participant_email(item)})) - if teachers_needed - else None, - tuple(sorted({_participant_email(item) for item in students if _participant_email(item)})) - if students_needed - else None, + normalized_teachers, + normalized_students, + tuple(sorted(unresolved)), ) return { - course_id: (teachers, students) - for course_id, teachers, students in await asyncio.gather( + course_id: (teachers, students, unresolved) + for course_id, teachers, students, unresolved in await asyncio.gather( *( read(course_id, teachers, students) - for course_id, (teachers, students) in requested.items() + for course_id, ( + _alias, + _course, + _detail, + teachers, + students, + ) in requested.items() ) ) } + async def _normalize_roster_members( + self, + members: Iterable[str], + directory_snapshot: Optional[Mapping[str, Any]], + ) -> tuple[tuple[str, ...], tuple[str, ...]]: + normalized: set[str] = set() + unresolved: set[str] = set() + for raw_email in sorted( + {_directory_identifier(email) for email in members} - {""} + ): + primary = await self._normalize_live_identity( + raw_email, + "", + directory_snapshot, + ) + if primary: + normalized.add(primary) + else: + unresolved.add(raw_email) + return tuple(sorted(normalized)), tuple(sorted(unresolved)) + + def _persist_roster_chunk( + self, + import_id: str, + role: str, + chunk: Sequence[str], + normalized: Mapping[str, tuple[tuple[str, ...], tuple[str, ...]]], + requested: Mapping[ + str, + tuple[str, _DesiredCourse, Any, bool, bool], + ], + manifest_guard_aliases: set[str], + ) -> None: + aliases = tuple(requested[course_id][1].alias for course_id in chunk) + states = self.store.get_managed_course_states(aliases) + verifications: list[ManagedCourseVerification] = [] + dirty: list[ManagedCourseDirty] = [] + verified_keys: list[str] = [] + for course_id in chunk: + alias_key, course, detail, _teachers, _students = requested[course_id] + state = states.get(alias_key) + if ( + state is None + or not _has_exact_alias(detail, course.alias) + or state.course_id not in {"", course_id} + ): + if state is not None and state.course_id not in {"", course_id}: + self.store.mark_managed_course_dirty( + ( + ManagedCourseDirty( + course.alias, + metadata=True, + teachers=role == "teachers", + students=role == "students", + ), + ), + error_code="OR-MANAGED-COURSE-ID-DRIFT", + recovery_required=True, + ) + if alias_key in manifest_guard_aliases: + # Do not adopt the rebound ID. The executor still needs + # this live roster basis to produce OR-MANIFEST-DRIFT. + continue + raise OneRosterError( + "OR-MANAGED-COURSE-ID-DRIFT", + "The exact managed alias resolved to a different Classroom course.", + ) + raise _IncompleteReadCoverage( + "A roster chunk did not retain exact managed-course identity." + ) + members, unresolved = normalized[course_id] + if unresolved: + dirty.append( + ManagedCourseDirty( + course.alias, + teachers=role == "teachers", + students=role == "students", + ) + ) + continue + verifications.append( + ManagedCourseVerification( + alias=course.alias, + course_id=course_id, + source_class_id=course.class_id, + last_seen_import_id=import_id, + teacher_hash=teacher_hash(members) if role == "teachers" else None, + student_hash=student_hash(members) if role == "students" else None, + teacher_members=members if role == "teachers" else None, + student_members=members if role == "students" else None, + ) + ) + verified_keys.append(alias_key) + self.store.record_managed_course_verifications(tuple(verifications)) + self.store.mark_managed_course_dirty( + tuple(dirty), + error_code="OR-LIVE-PARTICIPANT-UNRESOLVED", + ) + self._checkpointed_scopes.update((key, role) for key in verified_keys) + def _persist_live_verification( self, import_id: str, @@ -874,22 +1763,33 @@ def _persist_live_verification( live = live_courses.get(key) if intent is None or live is None: continue + metadata_requested = bool( + intent.metadata and (key, "metadata") not in self._checkpointed_scopes + ) + teachers_requested = bool( + intent.teachers and (key, "teachers") not in self._checkpointed_scopes + ) + students_requested = bool( + intent.students and (key, "students") not in self._checkpointed_scopes + ) + if not (metadata_requested or teachers_requested or students_requested): + continue course_id = _text(live.detail, "id") known = known_states.get(key) if known is not None and known.course_id and course_id != known.course_id: identity_dirty.append( ManagedCourseDirty( course.alias, - metadata=True, - teachers=intent.teachers, - students=intent.students, + metadata=metadata_requested, + teachers=teachers_requested, + students=students_requested, ) ) continue exact_alias = _has_exact_alias(live.detail, course.alias) state = _text(live.detail, "course_state", "courseState").upper() metadata_covered = bool( - intent.metadata + metadata_requested and course_id and exact_alias and live.owner_email @@ -898,13 +1798,17 @@ def _persist_live_verification( roster_covered = bool( course_id and exact_alias and not live.unresolved_members ) - if intent.metadata and not metadata_covered: + if metadata_requested and not metadata_covered: dirty.append(ManagedCourseDirty(course.alias, metadata=True)) - if intent.teachers and not roster_covered: + if teachers_requested and not roster_covered: dirty.append(ManagedCourseDirty(course.alias, teachers=True)) - if intent.students and not roster_covered: + if students_requested and not roster_covered: dirty.append(ManagedCourseDirty(course.alias, students=True)) - if not (metadata_covered or (intent.teachers and roster_covered) or (intent.students and roster_covered)): + if not ( + metadata_covered + or (teachers_requested and roster_covered) + or (students_requested and roster_covered) + ): continue verified.append( ManagedCourseVerification( @@ -926,22 +1830,22 @@ def _persist_live_verification( ), teacher_hash=( teacher_hash(live.teachers) - if intent.teachers and roster_covered + if teachers_requested and roster_covered else None ), student_hash=( student_hash(live.students) - if intent.students and roster_covered + if students_requested and roster_covered else None ), teacher_members=( tuple(live.teachers) - if intent.teachers and roster_covered + if teachers_requested and roster_covered else None ), student_members=( tuple(live.students) - if intent.students and roster_covered + if students_requested and roster_covered else None ), course_state=state if metadata_covered else None, @@ -1341,7 +2245,11 @@ def _resolve_snapshot_user( ) -def _index_managed_courses(raw: Any) -> dict[str, Any]: +def _index_managed_courses( + raw: Any, + *, + requested_aliases: Sequence[str], +) -> dict[str, Any]: if ( isinstance(raw, (str, bytes, Mapping)) or not isinstance(raw, Sequence) @@ -1350,30 +2258,46 @@ def _index_managed_courses(raw: Any) -> dict[str, Any]: "OR-CLASSROOM-READ", "The live Classroom snapshot had an invalid shape; no import plan was created.", ) + requested = { + _managed_alias_key(alias): str(alias).removeprefix("d:") + for alias in requested_aliases + if _managed_alias_key(alias) + } + if len(requested) != len(requested_aliases): + raise OneRosterError( + "OR-ALIAS-COLLISION", + "The managed-course metadata chunk did not contain unique exact aliases.", + ) indexed: dict[str, Any] = {} course_ids: dict[str, str] = {} for detail in raw: - aliases = { + matches = { key for alias in _aliases(detail) if (key := _managed_alias_key(alias)) + and key in requested } - for key in aliases: - if key in indexed: + if len(matches) != 1: + raise OneRosterError( + "OR-ALIAS-COLLISION", + "A live Classroom course did not map to exactly one requested managed alias.", + ) + key = next(iter(matches)) + if key in indexed: + raise OneRosterError( + "OR-ALIAS-COLLISION", + "More than one live Classroom course claims the same managed alias.", + ) + indexed[key] = detail + course_id = _text(detail, "id") + if course_id: + existing_alias = course_ids.get(course_id) + if existing_alias is not None and existing_alias != key: raise OneRosterError( "OR-ALIAS-COLLISION", - "More than one live Classroom course claims the same managed alias.", + "More than one managed alias resolves to the same live Classroom course.", ) - indexed[key] = detail - course_id = _text(detail, "id") - if course_id: - existing_alias = course_ids.get(course_id) - if existing_alias is not None and existing_alias != key: - raise OneRosterError( - "OR-ALIAS-COLLISION", - "More than one managed alias resolves to the same live Classroom course.", - ) - course_ids[course_id] = key + course_ids[course_id] = key return indexed diff --git a/gamgui/core/connectors/gam_connector.py b/gamgui/core/connectors/gam_connector.py index b4b34a6..1cea9ce 100644 --- a/gamgui/core/connectors/gam_connector.py +++ b/gamgui/core/connectors/gam_connector.py @@ -64,6 +64,8 @@ from .person import ConnectorAccount, Person MAX_ONEROSTER_BULK_ROSTER_MEMBERS = 500_000 +ONEROSTER_MANAGED_ALIAS_CHUNK_CAP = 200 +ONEROSTER_ROSTER_CHUNK_CAP = 100 def _parse_signature(text: str) -> str: @@ -851,6 +853,10 @@ async def list_oneroster_managed_courses( selected.append(alias) if not selected: return [] + if len(selected) > ONEROSTER_MANAGED_ALIAS_CHUNK_CAP: + raise ValueError( + "managed Classroom alias request exceeds the OneRoster chunk cap" + ) selector_path = _write_private_selector( selected, @@ -1041,6 +1047,10 @@ async def list_course_participants_many( raise ValueError("invalid Classroom roster role") if not selected: return CourseRosterSnapshot.empty() + if len(selected) > ONEROSTER_ROSTER_CHUNK_CAP: + raise ValueError( + "Classroom roster request exceeds the OneRoster chunk cap" + ) if any( len(course_id) > 512 or any(character in course_id for character in "\r\n\x00") @@ -1071,7 +1081,7 @@ async def list_course_participants_many( normalized_role, ), timeout=timeout, - serialize=True, + serialize=False, ) as result: participants = await _parse_private_spool_off_loop( _read_oneroster_participants_spool, diff --git a/tests/test_oneroster_bulk_connector.py b/tests/test_oneroster_bulk_connector.py index c824d02..fd2044f 100644 --- a/tests/test_oneroster_bulk_connector.py +++ b/tests/test_oneroster_bulk_connector.py @@ -305,12 +305,12 @@ async def test_sparse_alias_results_are_mapped_without_recursive_gam_calls( ) -async def test_fifty_thousand_missing_aliases_use_one_private_selector_process( +async def test_one_bounded_missing_alias_chunk_uses_one_private_selector_process( tmp_path: Path, ): runner = AliasLookupRunner(tmp_path, {}) connector = GAMConnector(runner, "example.org") # type: ignore[arg-type] - aliases = [f"Section_{number}" for number in range(50_000)] + aliases = [f"Section_{number}" for number in range(200)] courses = await connector.list_oneroster_managed_courses(aliases) @@ -318,8 +318,8 @@ async def test_fifty_thousand_missing_aliases_use_one_private_selector_process( assert len(runner.calls) == 1 assert len(runner.selector_values) == 1 assert runner.selector_values[0][0] == "d:Section_0" - assert runner.selector_values[0][-1] == "d:Section_49999" - assert len(runner.selector_values[0]) == 50_000 + assert runner.selector_values[0][-1] == "d:Section_199" + assert len(runner.selector_values[0]) == 200 assert all(not path.exists() for path in runner.selector_paths) diff --git a/tests/test_oneroster_bulk_planner.py b/tests/test_oneroster_bulk_planner.py index 3062c63..0793ffb 100644 --- a/tests/test_oneroster_bulk_planner.py +++ b/tests/test_oneroster_bulk_planner.py @@ -52,26 +52,28 @@ async def list_oneroster_managed_courses(self, aliases): async def list_course_participants_many(self, course_ids, role="all"): self.calls["rosters"] += 1 - assert role == "all" + assert role in {"teachers", "students"} requested = set(course_ids) participants = [] for course in self.courses: if course.id not in requested: continue - participants.extend( - [ + if role == "teachers": + participants.append( CourseParticipant( course_id=course.id, email="teacher@example.org", role="teachers", - ), + ) + ) + else: + participants.append( CourseParticipant( course_id=course.id, email="student@example.org", role="students", - ), - ] - ) + ) + ) return CourseRosterSnapshot.from_participants( participants, requested, @@ -98,9 +100,9 @@ def __init__(self, detail_error_kind: GAMErrorKind) -> None: async def list_oneroster_managed_courses(self, aliases): self.calls["courses"] += 1 raise GAMError( - kind=GAMErrorKind.NOT_FOUND, + kind=self.detail_error_kind, exit_code=1, - stderr="Requested Classroom course alias was not found.", + stderr="Bounded Classroom lookup failed.", ) async def get_course(self, *_args, **_kwargs): @@ -155,7 +157,7 @@ def _managed_course( @pytest.mark.asyncio -async def test_directory_and_classroom_snapshots_begin_concurrently(tmp_path: Path): +async def test_directory_and_first_bounded_classroom_chunk_begin_concurrently(tmp_path: Path): service, import_id = _ready_service(tmp_path) connector = ConcurrentSnapshotConnector() @@ -185,34 +187,35 @@ async def test_planner_reuses_each_bulk_snapshot_and_never_reads_details( { "directory": 1, "courses": 1, - "rosters": 1, + "rosters": 2, } ) @pytest.mark.asyncio -async def test_bulk_alias_not_found_falls_back_to_bounded_detail_reads( +async def test_bulk_alias_failure_does_not_fan_out_to_detail_reads( tmp_path: Path, ): service, import_id = _ready_service(tmp_path) connector = BulkLookupFailureConnector(GAMErrorKind.NOT_FOUND) - plan = await service.build_live_plan( - connector, - import_id, - limited_import=True, - ) + with pytest.raises(OneRosterError) as failure: + await service.build_live_plan( + connector, + import_id, + limited_import=True, + ) - assert service.scope_readiness().ready - assert any(action.kind == "course_create" for action in plan.actions) + assert failure.value.code == "OR-CLASSROOM-READ" + assert not service.scope_readiness().ready assert connector.calls["directory"] == 1 assert connector.calls["courses"] == 1 - assert connector.calls["get_course"] >= 1 + assert connector.calls["get_course"] == 0 assert connector.calls["rosters"] == 0 @pytest.mark.asyncio -async def test_bulk_alias_fallback_still_fails_closed_on_permission_error( +async def test_bulk_alias_permission_failure_does_not_fan_out( tmp_path: Path, ): service, import_id = _ready_service(tmp_path) @@ -225,10 +228,10 @@ async def test_bulk_alias_fallback_still_fails_closed_on_permission_error( limited_import=True, ) - assert failure.value.code == "OR-CLASSROOM-READ" + assert failure.value.code == "OR-ONEROSTER-READ-PERMISSION" assert not service.scope_readiness().ready assert connector.calls["courses"] == 1 - assert connector.calls["get_course"] >= 1 + assert connector.calls["get_course"] == 0 @pytest.mark.asyncio @@ -396,7 +399,7 @@ async def test_limited_plan_does_not_reactivate_existing_archived_course( { "directory": 1, "courses": 1, - "rosters": 1, + "rosters": 2, } ) diff --git a/tests/test_oneroster_delta.py b/tests/test_oneroster_delta.py index 5d67396..b362e98 100644 --- a/tests/test_oneroster_delta.py +++ b/tests/test_oneroster_delta.py @@ -185,7 +185,10 @@ async def test_first_baseline_then_identical_plan_skips_live_course_reads( assert first.actions == () assert connector.metadata_calls == [("Section_101",)] - assert connector.roster_calls == [(('1000',), "all")] + assert connector.roster_calls == [ + (("1000",), "teachers"), + (("1000",), "students"), + ] state = service.store.get_managed_course_states(["Section_101"])["section_101"] assert state.course_id == "1000" assert state.desired_metadata_hash == state.verified_metadata_hash diff --git a/tests/test_oneroster_read_chunks.py b/tests/test_oneroster_read_chunks.py new file mode 100644 index 0000000..4b04afe --- /dev/null +++ b/tests/test_oneroster_read_chunks.py @@ -0,0 +1,556 @@ +from __future__ import annotations + +import asyncio +import os +import tempfile +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from gamgui.components.oneroster import OneRosterService, ThresholdProfile +from gamgui.components.oneroster.models import ManagedCourseDirty, OneRosterError +from gamgui.components.oneroster.planner import OneRosterPlanner +from gamgui.components.oneroster.store import OneRosterStore +from gamgui.core.classroom.models import CourseDetail, CourseRosterSnapshot +from gamgui.core.connectors.gam_connector import ( + GAMConnector, + ONEROSTER_MANAGED_ALIAS_CHUNK_CAP, + ONEROSTER_ROSTER_CHUNK_CAP, +) +from gamgui.core.gam.errors import GAMError, GAMErrorKind +from gamgui.core.gam.models import GAMUser +from tests.test_oneroster_helpers import csv_text, valid_files, zip_bytes + + +def _alias(index: int) -> str: + return f"Section_{100_000 + index:06d}" + + +def _course_id(index: int) -> str: + return f"course-{index:06d}" + + +def _detail(index: int, *, course_id: str | None = None) -> CourseDetail: + return CourseDetail( + id=course_id or _course_id(index), + name=f"Algebra I – P{index:04d} (2026-27)", + section=f"P{index:04d}", + room=f"R{index:04d}", + owner_id="teacher-id", + owner_email="teacher@example.org", + course_state="ACTIVE", + aliases=(f"d:{_alias(index)}",), + ) + + +def _files_for_courses(count: int) -> dict[str, str]: + files = valid_files() + classes = [] + enrollments = [] + for index in range(1, count + 1): + class_id = str(100_000 + index) + classes.append( + { + "sourcedId": class_id, + "status": "active", + "title": f"Algebra Section {index:04d}", + "classCode": f"P{index:04d}", + "location": f"R{index:04d}", + "courseSourcedId": "course-1", + "terms": "term-1", + "schoolSourcedId": "school-1", + "grades": "9", + } + ) + enrollments.extend( + ( + { + "sourcedId": f"teacher-{index:06d}", + "status": "active", + "classSourcedId": class_id, + "schoolSourcedId": "school-1", + "userSourcedId": "teacher-1", + "role": "teacher", + "primary": "true", + "beginDate": "", + "endDate": "", + }, + { + "sourcedId": f"student-{index:06d}", + "status": "active", + "classSourcedId": class_id, + "schoolSourcedId": "school-1", + "userSourcedId": "student-1", + "role": "student", + "primary": "false", + "beginDate": "", + "endDate": "", + }, + ) + ) + files["classes.csv"] = csv_text( + ( + "sourcedId", + "status", + "title", + "classCode", + "location", + "courseSourcedId", + "terms", + "schoolSourcedId", + "grades", + ), + classes, + ) + files["enrollments.csv"] = csv_text( + ( + "sourcedId", + "status", + "classSourcedId", + "schoolSourcedId", + "userSourcedId", + "role", + "primary", + "beginDate", + "endDate", + ), + enrollments, + ) + return files + + +class ChunkConnector: + oneroster_audit_sample_size = 0 + oneroster_read_max_concurrency = 4 + oneroster_read_max_attempts = 3 + oneroster_read_backoff_base_seconds = 1.0 + oneroster_read_backoff_cap_seconds = 8.0 + + def __init__(self, count: int) -> None: + self.users = { + "teacher@example.org": GAMUser( + "teacher@example.org", + user_id="teacher-id", + ), + "student@example.org": GAMUser( + "student@example.org", + user_id="student-id", + ), + } + self.courses = {_alias(index): _detail(index) for index in range(1, count + 1)} + self.teachers = { + _course_id(index): {"teacher@example.org"} + for index in range(1, count + 1) + } + self.students = { + _course_id(index): {"student@example.org"} + for index in range(1, count + 1) + } + self.metadata_calls: list[tuple[str, ...]] = [] + self.roster_calls: list[tuple[tuple[str, ...], str]] = [] + self.get_course_calls = 0 + self.metadata_failures: dict[tuple[str, ...], list[BaseException]] = {} + self.roster_failures: dict[ + tuple[tuple[str, ...], str], + list[BaseException | CourseRosterSnapshot], + ] = {} + self.sleep_delays: list[float] = [] + self.active_reads = 0 + self.maximum_active_reads = 0 + self.unrelated_courses: dict[str, CourseDetail] = {} + self.oneroster_read_sleep = self._record_sleep + self.oneroster_read_jitter = lambda: 0.0 + + async def _record_sleep(self, delay: float) -> None: + self.sleep_delays.append(delay) + await asyncio.sleep(0) + + async def _begin_read(self) -> None: + self.active_reads += 1 + self.maximum_active_reads = max(self.maximum_active_reads, self.active_reads) + await asyncio.sleep(0) + + def _finish_read(self) -> None: + self.active_reads -= 1 + + def reset_reads(self) -> None: + self.metadata_calls.clear() + self.roster_calls.clear() + self.get_course_calls = 0 + self.sleep_delays.clear() + self.active_reads = 0 + self.maximum_active_reads = 0 + + async def list_oneroster_directory(self): + return dict(self.users) + + async def get_user(self, email: str, **_kwargs): + return self.users[email.casefold()] + + async def get_course(self, course_ref: str, **_kwargs): + self.get_course_calls += 1 + return self.courses[course_ref.removeprefix("d:")] + + async def list_oneroster_managed_courses(self, aliases): + requested = tuple(str(alias) for alias in aliases) + self.metadata_calls.append(requested) + await self._begin_read() + try: + failures = self.metadata_failures.get(requested) + if failures: + raise failures.pop(0) + return [ + self.courses[alias.removeprefix("d:")] + for alias in requested + if alias.removeprefix("d:") in self.courses + ] + finally: + self._finish_read() + + async def list_course_participants_many(self, course_ids, role="all"): + requested = tuple(str(course_id) for course_id in course_ids) + selected_role = str(role) + key = (requested, selected_role) + self.roster_calls.append(key) + await self._begin_read() + try: + failures = self.roster_failures.get(key) + if failures: + outcome = failures.pop(0) + if isinstance(outcome, BaseException): + raise outcome + return outcome + return CourseRosterSnapshot( + { + course_id: ( + frozenset(self.teachers.get(course_id, set())) + if selected_role == "teachers" + else frozenset(), + frozenset(self.students.get(course_id, set())) + if selected_role == "students" + else frozenset(), + ) + for course_id in requested + }, + frozenset(requested), + ) + finally: + self._finish_read() + + async def list_course_participants(self, course_id: str, role: str): + source = self.teachers if role == "teachers" else self.students + return tuple( + SimpleNamespace(email=email) + for email in sorted(source.get(course_id, set())) + ) + + +def _ready_service( + tmp_path: Path, + count: int, +) -> tuple[OneRosterService, str, ChunkConnector]: + service = OneRosterService("example.org", tmp_path / "component") + snapshot = service.upload(zip_bytes(_files_for_courses(count))) + service.save_threshold_profile(ThresholdProfile(configured=True)) + return service, snapshot.id, ChunkConnector(count) + + +def _dirty_students(service: OneRosterService, count: int, code: str = "TEST-DIRTY") -> None: + service.store.mark_managed_course_dirty( + tuple( + ManagedCourseDirty(_alias(index), students=True) + for index in range(1, count + 1) + ), + error_code=code, + ) + + +class EmptySpoolRunner: + def __init__(self, root: Path) -> None: + self.base_dir = root + self.timeout = 120.0 + self.calls = 0 + + @asynccontextmanager + async def run_authenticated_to_file(self, *_args, **_kwargs): + self.calls += 1 + fd, raw_path = tempfile.mkstemp(dir=str(self.base_dir)) + os.close(fd) + path = Path(raw_path) + try: + yield SimpleNamespace(path=path) + finally: + path.unlink(missing_ok=True) + + +@pytest.mark.asyncio +async def test_connector_caps_normalize_duplicates_and_keep_empty_reads_safe( + tmp_path: Path, +): + runner = EmptySpoolRunner(tmp_path) + connector = GAMConnector(runner, "example.org") # type: ignore[arg-type] + + assert await connector.list_oneroster_managed_courses([]) == [] + assert (await connector.list_course_participants_many([], "students")).covers([]) + assert await connector.list_oneroster_managed_courses(["Section_1"] * 1_001) == [] + assert runner.calls == 1 + + with pytest.raises(ValueError, match="alias request exceeds"): + await connector.list_oneroster_managed_courses( + [f"Section_{index}" for index in range(ONEROSTER_MANAGED_ALIAS_CHUNK_CAP + 1)] + ) + with pytest.raises(ValueError, match="roster request exceeds"): + await connector.list_course_participants_many( + [str(index) for index in range(ONEROSTER_ROSTER_CHUNK_CAP + 1)], + "students", + ) + assert runner.calls == 1 + + +@pytest.mark.asyncio +async def test_1001_aliases_use_deterministic_bounded_adaptive_metadata_chunks( + tmp_path: Path, +): + connector = ChunkConnector(1_001) + planner = OneRosterPlanner( + OneRosterStore("example.org", tmp_path / "component"), + connector, + ) + aliases = [_alias(index) for index in range(1_001, 0, -1)] + + indexed = await planner._read_managed_course_snapshot(aliases) + + expected = tuple(sorted(set(aliases), key=str.casefold)) + assert indexed is not None and len(indexed) == 1_001 + assert tuple(alias for call in connector.metadata_calls for alias in call) == expected + assert [len(call) for call in connector.metadata_calls] == [200, 200, 200, 200, 200, 1] + assert max(map(len, connector.metadata_calls)) == ONEROSTER_MANAGED_ALIAS_CHUNK_CAP + assert planner._read_performance.metadata_chunk_count == 6 + assert planner._read_performance.largest_metadata_chunk == 200 + assert planner._read_performance.metadata_chunk_worker_levels == [1, 1, 2, 2, 3, 3] + assert planner._read_performance.maximum_observed_read_concurrency == 2 + assert connector.maximum_active_reads == 2 + + +@pytest.mark.asyncio +async def test_501_student_candidates_use_only_bounded_student_chunks(tmp_path: Path): + service, import_id, connector = _ready_service(tmp_path, 501) + await service.build_live_plan(connector, import_id) + _dirty_students(service, 501) + connector.reset_reads() + + plan = await service.build_live_plan(connector, import_id) + + assert connector.metadata_calls == [] + assert {role for _ids, role in connector.roster_calls} == {"students"} + assert [len(ids) for ids, _role in connector.roster_calls] == [100, 100, 100, 100, 100, 1] + assert max(len(ids) for ids, _role in connector.roster_calls) == ONEROSTER_ROSTER_CHUNK_CAP + assert plan.performance.student_roster_chunk_count == 6 + assert plan.performance.teacher_roster_chunk_count == 0 + assert plan.performance.largest_roster_chunk == 100 + + +@pytest.mark.asyncio +async def test_teacher_and_student_candidates_are_chunked_independently(tmp_path: Path): + service, import_id, connector = _ready_service(tmp_path, 3) + await service.build_live_plan(connector, import_id) + service.store.mark_managed_course_dirty( + ( + ManagedCourseDirty(_alias(1), teachers=True), + ManagedCourseDirty(_alias(2), students=True), + ManagedCourseDirty(_alias(3), teachers=True, students=True), + ), + error_code="TEST-DIRTY", + ) + connector.reset_reads() + + await service.build_live_plan(connector, import_id) + + assert connector.roster_calls == [ + ((_course_id(1), _course_id(3)), "teachers"), + ((_course_id(2), _course_id(3)), "students"), + ] + assert _course_id(2) not in connector.roster_calls[0][0] + + +@pytest.mark.asyncio +async def test_roster_failure_preserves_first_chunk_and_retry_skips_it(tmp_path: Path): + service, import_id, connector = _ready_service(tmp_path, 201) + await service.build_live_plan(connector, import_id) + _dirty_students(service, 201) + connector.reset_reads() + ordered_ids = tuple(_course_id(index) for index in range(1, 202)) + failed_chunk = ordered_ids[100:200] + rate_error = lambda: GAMError( + GAMErrorKind.RATE_LIMITED, + exit_code=1, + stderr="429", + ) + connector.roster_failures[(failed_chunk, "students")] = [ + rate_error(), + rate_error(), + rate_error(), + ] + + with pytest.raises(OneRosterError) as failure: + await service.build_live_plan(connector, import_id) + + assert failure.value.code == "OR-ONEROSTER-READ-RATE-LIMITED" + assert connector.roster_calls == [ + (ordered_ids[:100], "students"), + (failed_chunk, "students"), + (failed_chunk, "students"), + (failed_chunk, "students"), + ] + assert connector.sleep_delays == [1.0, 2.0] + states = service.store.get_managed_course_states( + (_alias(1), _alias(101), _alias(201)) + ) + assert not states[_alias(1).casefold()].student_roster_dirty + assert states[_alias(101).casefold()].student_roster_dirty + assert states[_alias(101).casefold()].last_error_code == ( + "OR-ONEROSTER-READ-RATE-LIMITED" + ) + assert states[_alias(201).casefold()].last_error_code == "TEST-DIRTY" + assert not states[_alias(101).casefold()].metadata_dirty + assert not states[_alias(101).casefold()].teacher_roster_dirty + + connector.roster_failures.clear() + connector.reset_reads() + await service.build_live_plan(connector, import_id) + + retried_ids = tuple( + course_id + for ids, role in connector.roster_calls + if role == "students" + for course_id in ids + ) + assert not set(ordered_ids[:100]) & set(retried_ids) + assert retried_ids == ordered_ids[100:] + assert [len(ids) for ids, _role in connector.roster_calls] == [100, 1] + + +@pytest.mark.asyncio +async def test_rate_limited_metadata_retries_same_chunk_without_alias_fanout( + tmp_path: Path, +): + connector = ChunkConnector(1_001) + planner = OneRosterPlanner( + OneRosterStore("example.org", tmp_path / "component"), + connector, + ) + aliases = tuple(_alias(index) for index in range(1, 1_002)) + ordered = tuple(sorted(aliases, key=str.casefold)) + throttled = ordered[400:600] + connector.metadata_failures[throttled] = [ + GAMError(GAMErrorKind.RATE_LIMITED, exit_code=1, stderr="429") + ] + + await planner._read_managed_course_snapshot(aliases) + + assert connector.metadata_calls.count(throttled) == 2 + assert connector.get_course_calls == 0 + assert connector.sleep_delays == [1.0] + assert planner._read_performance.rate_limit_count == 1 + assert planner._read_performance.retried_chunk_count == 1 + assert planner._read_performance.metadata_chunk_worker_levels == [1, 1, 2, 2, 1, 2] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("kind", "code"), + [ + (GAMErrorKind.AUTH_EXPIRED, "OR-ONEROSTER-READ-AUTH"), + (GAMErrorKind.PERMISSION_DENIED, "OR-ONEROSTER-READ-PERMISSION"), + ], +) +async def test_auth_and_permission_failures_never_fan_out_to_alias_reads( + tmp_path: Path, + kind: GAMErrorKind, + code: str, +): + connector = ChunkConnector(1) + planner = OneRosterPlanner( + OneRosterStore("example.org", tmp_path / kind.value), + connector, + ) + chunk = (_alias(1),) + connector.metadata_failures[chunk] = [ + GAMError(kind, exit_code=1, stderr=kind.value) + ] + + with pytest.raises(OneRosterError) as failure: + await planner._read_managed_course_snapshot(chunk) + + assert failure.value.code == code + assert connector.metadata_calls == [chunk] + assert connector.get_course_calls == 0 + + +@pytest.mark.asyncio +async def test_incomplete_roster_coverage_is_dirty_and_not_verified(tmp_path: Path): + service, import_id, connector = _ready_service(tmp_path, 1) + await service.build_live_plan(connector, import_id) + before = service.store.get_managed_course_states((_alias(1),))[ + _alias(1).casefold() + ] + _dirty_students(service, 1) + connector.reset_reads() + key = ((_course_id(1),), "students") + connector.roster_failures[key] = [CourseRosterSnapshot.empty()] + + with pytest.raises(OneRosterError) as failure: + await service.build_live_plan(connector, import_id) + + after = service.store.get_managed_course_states((_alias(1),))[ + _alias(1).casefold() + ] + assert failure.value.code == "OR-CLASSROOM-ROSTER-READ" + assert after.student_roster_dirty + assert after.verified_student_hash == before.verified_student_hash + assert after.last_student_verified_at == before.last_student_verified_at + assert not after.metadata_dirty + assert not after.teacher_roster_dirty + + +@pytest.mark.asyncio +async def test_duplicate_course_id_mapping_across_metadata_chunks_fails_closed( + tmp_path: Path, +): + connector = ChunkConnector(201) + connector.courses[_alias(201)] = _detail(201, course_id=_course_id(1)) + planner = OneRosterPlanner( + OneRosterStore("example.org", tmp_path / "component"), + connector, + ) + + with pytest.raises(OneRosterError) as failure: + await planner._read_managed_course_snapshot( + tuple(_alias(index) for index in range(1, 202)) + ) + + assert failure.value.code == "OR-ALIAS-COLLISION" + assert connector.get_course_calls == 0 + + +@pytest.mark.asyncio +async def test_ten_thousand_unrelated_courses_add_zero_managed_reads(tmp_path: Path): + service, import_id, connector = _ready_service(tmp_path, 1) + await service.build_live_plan(connector, import_id) + connector.unrelated_courses = { + f"unrelated-{index}": CourseDetail( + id=f"unrelated-{index}", + name="Unmanaged", + aliases=(f"d:Other_{index}",), + ) + for index in range(10_000) + } + connector.reset_reads() + + await service.build_live_plan(connector, import_id) + + assert len(connector.unrelated_courses) == 10_000 + assert connector.metadata_calls == [] + assert connector.roster_calls == [] From 08adb8297f022ef4cdc1c2f2d6176a61681eea11 Mon Sep 17 00:00:00 2001 From: Sykez <276981287+Sykezzz@users.noreply.github.com> Date: Mon, 24 Aug 2026 03:48:40 -0500 Subject: [PATCH 3/3] fix(toolchain): complete Windows GAM 7.47.06 pin (cherry picked from commit af01c2a4c8fe75274e861d158e5a4fe4d44ac20f) --- .github/workflows/windows-prerelease.yml | 6 +- scripts/build_windows_release.ps1 | 12 +- scripts/build_windows_setup.ps1 | 11 +- scripts/bump_gam.py | 74 ++++++++--- scripts/fetch_gam_windows.ps1 | 72 +++++++++-- scripts/gam_checksums.txt | 1 + scripts/windows_setup.iss | 5 +- tests/test_build_profiles.py | 22 +++- tests/test_bump_gam.py | 157 ++++++++++++++++++----- tests/test_command_contract.py | 46 ++++++- 10 files changed, 330 insertions(+), 76 deletions(-) diff --git a/.github/workflows/windows-prerelease.yml b/.github/workflows/windows-prerelease.yml index ddc9797..2a9013c 100644 --- a/.github/workflows/windows-prerelease.yml +++ b/.github/workflows/windows-prerelease.yml @@ -125,7 +125,11 @@ jobs: if ($actual -ne $expected -or $actual -ne $manifest.setup.sha256) { throw "Setup checksum receipt mismatch." } if ((Get-Item -LiteralPath $setup).Length -ge 2GB) { throw "Setup exceeds GitHub's asset limit." } if (@($manifest.profiles.profile | Sort-Object) -join ',' -ne 'classroom-oneroster,core') { throw "Both profiles were not embedded." } - if ($manifest.gam_version -ne '7.47.02') { throw "Bundled GAM pin mismatch." } + $gamVersionOutput = & .venv\Scripts\python.exe -c "from gamgui.core.gam.commands import EXPECTED_GAM_VERSION; print(EXPECTED_GAM_VERSION)" + $gamVersionExitCode = $LASTEXITCODE + $expectedGamVersion = ($gamVersionOutput | Out-String).Trim() + if ($gamVersionExitCode -ne 0 -or $expectedGamVersion -notmatch '^\d+\.\d+\.\d+$') { throw "Repository GAM source pin is invalid." } + if ([string]$manifest.gam_version -cne $expectedGamVersion) { throw "Bundled GAM pin mismatch." } - name: Exercise fail-closed setup, both profiles, signatures, and uninstall shell: pwsh run: | diff --git a/scripts/build_windows_release.ps1 b/scripts/build_windows_release.ps1 index 8d8bc82..0e1faee 100644 --- a/scripts/build_windows_release.ps1 +++ b/scripts/build_windows_release.ps1 @@ -16,6 +16,12 @@ $repoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path Set-Location $repoRoot $python = Join-Path $repoRoot ".venv\Scripts\python.exe" if (-not (Test-Path -LiteralPath $python)) { throw "Run uv sync with the dev, desktop, and build extras first." } +$gamVersionOutput = & $python -c "from gamgui.core.gam.commands import EXPECTED_GAM_VERSION; print(EXPECTED_GAM_VERSION)" +$gamVersionExitCode = $LASTEXITCODE +$expectedGamVersion = ($gamVersionOutput | Out-String).Trim() +if ($gamVersionExitCode -ne 0 -or $expectedGamVersion -notmatch '^\d+\.\d+\.\d+$') { + throw "The repository GAM source pin is invalid." +} $packagedPaths = @( "main.py", "gamgui", "gamgui.spec", "gamgui-updater.spec", "pyproject.toml", "uv.lock", @@ -44,7 +50,7 @@ if ($Bootstrap) { } } -& (Join-Path $PSScriptRoot "fetch_gam_windows.ps1") +& (Join-Path $PSScriptRoot "fetch_gam_windows.ps1") -Tag "v$expectedGamVersion" if ($LASTEXITCODE) { throw "Pinned GAM vendoring failed." } & $python -c "import PyInstaller, webview, keyring" if ($LASTEXITCODE) { throw "Locked build dependencies are unavailable." } @@ -88,8 +94,8 @@ if ( [string]$embeddedProfile.artifact.toolchain_manifest_digest -ne $toolchainDigest -or [string]$embeddedProfile.artifact.signer_thumbprint -ne $CertificateSha256.ToLowerInvariant() ) { throw "The embedded Windows artifact identity did not match the exact build inputs." } -$embeddedGamVersion = (Get-Content -Raw -LiteralPath (Join-Path $bundle "_internal\resources\gam7\VERSION")).Trim() -if ($embeddedGamVersion -notmatch '7\.47\.02') { throw "The embedded Windows GAM version did not match the tested pin." } +$embeddedGamTag = (Get-Content -Raw -LiteralPath (Join-Path $bundle "_internal\resources\gam7\VERSION")).Trim() +if ($embeddedGamTag -cne "v$expectedGamVersion") { throw "The embedded Windows GAM version did not match the repository source pin." } & $python -c "import struct,sys; p=open(sys.argv[1],'rb'); p.seek(0x3c); p.seek(struct.unpack('