From e705dc6d45d99130c240296ad6806a18e6c4132c Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 21 Sep 2026 12:08:00 +0100 Subject: [PATCH] feat(observability): expose microbatch lifecycle bounds --- docs/execution-observability.md | 25 +- docs/sinks.md | 3 +- scripts/fensu_policy/constants.py | 1 + .../rules/execution_observability.py | 3 +- .../_helpers/materializations/microbatch.py | 232 ++++++++++--- src/sqlbuild/runtime/event_exporting/types.py | 1 + .../observability/_helpers/validation.py | 25 ++ .../classes/microbatch_lifecycle.py | 125 +++++++ .../runtime/observability/constants.py | 56 ++- src/sqlbuild/runtime/observability/models.py | 32 ++ .../executor/run/microbatch/_test_types.py | 14 + .../executor/run/microbatch/helpers.py | 24 ++ .../test_microbatch_observability.py | 324 ++++++++++++++++++ .../runtime/event_exporting/test_policy.py | 5 +- .../runtime/observability/_test_types.py | 8 + .../sqlbuild/runtime/observability/helpers.py | 27 ++ .../runtime/observability/test_catalog.py | 41 +++ .../test_microbatch_lifecycle.py | 208 +++++++++++ 18 files changed, 1099 insertions(+), 55 deletions(-) create mode 100644 src/sqlbuild/runtime/observability/classes/microbatch_lifecycle.py create mode 100644 tests/integration/src/sqlbuild/executor/run/microbatch/test_microbatch_observability.py create mode 100644 tests/unit/src/sqlbuild/runtime/observability/test_microbatch_lifecycle.py diff --git a/docs/execution-observability.md b/docs/execution-observability.md index 41e7d5e095..97cd6217d7 100644 --- a/docs/execution-observability.md +++ b/docs/execution-observability.md @@ -7,7 +7,7 @@ record that answers the question instead of treating every JSON or log file as i | Record | Purpose | Authority and retention | | --- | --- | --- | -| `LifecycleEvent` | Immutable facts about invocation, run, resource-attempt, operation, retry, and statement lifecycles | Canonical execution evidence. An explicitly attached `LifecycleEventLogStorage` is authoritative once the fact is durable. SQLite and PostgreSQL retain facts until an operator removes the database or applies an external retention policy. | +| `LifecycleEvent` | Immutable facts about invocation, run, resource-attempt, microbatch, operation, retry, and statement lifecycles | Canonical execution evidence. An explicitly attached `LifecycleEventLogStorage` is authoritative once the fact is durable. SQLite and PostgreSQL retain facts until an operator removes the database or applies an external retention policy. | | `DiagnosticLog` | Structured framework and user diagnostics | Explanatory only. It cannot establish that work started, completed, failed, skipped, or retried. Local compute-log retention applies. | | `stdout.log` and `stderr.log` | Exact process output for one invocation | Troubleshooting data, not execution truth. Local compute-log retention applies. | | `CommandOutputRecord` | Normalized, ANSI-free, line-oriented stdout/stderr chunks for an explicitly configured remote sink | Potentially sensitive, bounded, and potentially lossy troubleshooting data. It is not execution evidence and destination retention is project-owned. | @@ -52,8 +52,10 @@ Schema version 1 has these fields: `event_id` identifies immutable content. Re-appending canonically equal content with the same ID is an idempotent no-op; different content under that ID is an integrity error. `producer` and `producer_version` identify the emitting implementation, while `schema_version` identifies the -wire contract. Known schema-v1 events reject unknown envelope and payload fields. Readers preserve -unknown event names or newer schema versions as opaque envelopes rather than inventing semantics. +wire contract. Known schema-v1 events reject unknown envelope and payload fields. Schema version 2 +adds `invocation_sequence` and `external_context`, and catalogs bounded microbatch lifecycle facts. +Readers preserve unknown event names or newer schema versions as opaque envelopes rather than +inventing semantics. History storage retains those opaque envelopes and exposes them to queries, but run projection, native progress, integration-result output, and lifecycle-event sinks consume only known `LifecycleEvent` values and ignore opaque facts. @@ -61,7 +63,9 @@ native progress, integration-result output, and lifecycle-event sinks consume on An invocation ID exists for every fact. A run groups executable graph work. A stable `resource_id` identifies the logical model, source, check, or other resource, while a new `resource_attempt_id` identifies each attempt. `operation_id` identifies one blocking non-SQL -operation and `statement_id` one SQL statement. IDs are correlation values, not timestamps. +operation and `statement_id` one SQL statement. Microbatch facts correlate through the containing +resource attempt and identify their interval through bounded payload fields. IDs are correlation +values, not timestamps. Publication is synchronous and preserves each producer's call order. Concurrent producers have no global event-time ordering guarantee. Durable `storage_order` is the total order assigned by one @@ -70,6 +74,19 @@ Completed, failed, and `resource_attempt_skipped` facts are terminal. `retry_sch first-class fact between attempts. A start without terminal evidence remains unknown or presumed lost: no storage, projection, integration, or sink may fabricate success or failure. +## Microbatch lifecycle + +Each active microbatch emits `microbatch_started` before staging begins. The payload records the +inclusive `cursor_start`, exclusive `cursor_end_exclusive`, complete planned range, one-based +`batch_index`, `batch_count`, configured and effective batch sizes, strategy, run type, and whether +the interval is ordinary or recovery work. `microbatch_completed` repeats that context with +duration and optional affected rows; `microbatch_failed` repeats it with bounded error identity. + +The start fact is sufficient to identify the active interval when a process is cancelled or lost +before it can publish a terminal. Concurrent intervals share the containing resource-attempt +identity and retain distinct bounds and indexes; their publication order is not a completion-order +guarantee. + ## Integration consumers `SQLBUILD_INTEGRATION_RESULT_PATH` (or SQLBuild's integration-owned hidden `--event-output` route) diff --git a/docs/sinks.md b/docs/sinks.md index f659b70fca..c0a7e844ab 100644 --- a/docs/sinks.md +++ b/docs/sinks.md @@ -144,11 +144,12 @@ measurement/sample/threshold summary, bounded evidence metadata and rendered aud | `invocation_started` | `invocation` | `debug` | 0 | | `run_started` | `run` | `debug` | 0 | | `resource_attempt_started` | `resource` | `debug` | 0 | +| `microbatch_started` | `microbatch` | `debug` | 0 | | `operation_started` | `operation` | `debug` | 0 | | `statement_started`, `statement_submitted` | `statement` | `debug` | 0 | | `retry_scheduled` | `retry` | `warning` | 1 | | `audit_completed` | `audit` | `info` | 2 | -| `invocation_completed`, `run_completed`, `resource_attempt_completed`, `resource_attempt_skipped`, `operation_completed`, `statement_completed` | Corresponding kind | `info` | 2 | +| `invocation_completed`, `run_completed`, `resource_attempt_completed`, `resource_attempt_skipped`, `microbatch_completed`, `operation_completed`, `statement_completed` | Corresponding kind | `info` | 2 | | Any `*_failed` event | Corresponding kind | `error` | 3 | Diagnostics and command output never enter lifecycle-event queues. diff --git a/scripts/fensu_policy/constants.py b/scripts/fensu_policy/constants.py index efbd32f012..264aefef52 100644 --- a/scripts/fensu_policy/constants.py +++ b/scripts/fensu_policy/constants.py @@ -251,6 +251,7 @@ "src/sqlbuild/cli/commands/main/entrypoint/_dispatch_with_observability.py", "src/sqlbuild/executor/pipeline/main/run.py", "src/sqlbuild/observability.py", + "src/sqlbuild/runtime/observability/classes/microbatch_lifecycle.py", "src/sqlbuild/runtime/observability/classes/operation_lifecycle.py", "src/sqlbuild/runtime/observability/classes/resource_attempt_lifecycle.py", "src/sqlbuild/runtime/observability/classes/run_lifecycle.py", diff --git a/scripts/fensu_policy/rules/execution_observability.py b/scripts/fensu_policy/rules/execution_observability.py index 67e040665c..84ee4fd737 100644 --- a/scripts/fensu_policy/rules/execution_observability.py +++ b/scripts/fensu_policy/rules/execution_observability.py @@ -112,7 +112,8 @@ def raw_driver_execution_boundary(*, module: object, ctx: RuleContext) -> list[F slug="event-construction-ownership", message="canonical lifecycle records and catalogs must be created by observability owners", remediation=( - "Use OperationLifecycle, ResourceAttemptLifecycle, StatementLifecycle, or dispatcher APIs." + "Use MicrobatchLifecycle, OperationLifecycle, ResourceAttemptLifecycle, " + "StatementLifecycle, or dispatcher APIs." ), ) def event_construction_ownership(*, module: object, ctx: RuleContext) -> list[Fault]: diff --git a/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py b/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py index b69b697cc1..f5e2e88e0d 100644 --- a/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py +++ b/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py @@ -147,6 +147,9 @@ ReplayRequirementState, UnaccountedPartitionPolicy, ) +from sqlbuild.runtime.observability.classes.microbatch_lifecycle import ( + MicrobatchLifecycle, +) from sqlbuild.runtime.observability.classes.operation_lifecycle import ( OperationAttributes, OperationLifecycle, @@ -154,6 +157,7 @@ from sqlbuild.runtime.observability.main.canonicalize_operation_adapter import ( canonicalize_operation_adapter, ) +from sqlbuild.runtime.observability.models import MicrobatchLifecycleContext from sqlbuild.spec.contracts.types import MicrobatchLimitAction, TableType _DEFAULT_ON_SCHEMA_CHANGE: OnSchemaChange = OnSchemaChange.APPEND_NEW_COLUMNS @@ -225,6 +229,24 @@ class _PreparedMicrobatchExecution: history: _MicrobatchHistoryContext +@dataclass(frozen=True) +class _MicrobatchLifecyclePlan: + """Stable planned context repeated on each active microbatch fact.""" + + batch_count: int + cursor_start: str + cursor_end_exclusive: str + + +@dataclass(frozen=True) +class _MicrobatchBatchPreparation: + """Prepared batch state or the failure that prevents target DML.""" + + state: MicrobatchLifecycleState + schema_checked: bool + failure: ModelExecutionResult | None = None + + class _SerialMicrobatchEventStore: """Reject accidental durable-state use from the serial execution path.""" @@ -494,6 +516,10 @@ def execute_microbatch_entry( state=state, history_context=history_context, on_progress=on_progress, + lifecycle_plan=_microbatch_lifecycle_plan( + batch_plan=batch_plan, + cursor_type=context.entry.cursor_type or "", + ), ) state = batch_outcome.state if context.microbatch_event_store is not None: @@ -994,7 +1020,13 @@ def _execute_microbatch_batches( state: MicrobatchLifecycleState, history_context: _MicrobatchHistoryContext, on_progress: Callable[[str], None] | None = None, + lifecycle_plan: _MicrobatchLifecyclePlan | None = None, ) -> MicrobatchPhaseOutcome: + plan: _MicrobatchLifecyclePlan = lifecycle_plan or _MicrobatchLifecyclePlan( + batch_count=len(batches), + cursor_start=render(value=batches[0].start if batches else history_context.run_start), + cursor_end_exclusive=render(value=batches[-1].end if batches else history_context.run_end), + ) if context.microbatch_batch_runner is not None and not is_full_refresh and len(batches) > 1: return _execute_microbatch_batches_concurrently( context=context, @@ -1004,15 +1036,22 @@ def _execute_microbatch_batches( state=state, history_context=history_context, on_progress=on_progress, + lifecycle_plan=plan, ) schema_checked: bool = False completed_batches: int = 0 applied_intervals: list[tuple[str, str]] = [] total_rows: int = 0 row_count_known: bool = False - total_batches: int = len(batches) batch: BatchWindow for batch in batches: + lifecycle: MicrobatchLifecycle = _microbatch_lifecycle( + context=context, + batch=batch, + history=history_context, + plan=plan, + ) + lifecycle.start() lease_failure: ModelExecutionResult | None = _microbatch_lease_failure( context=context, state=state, @@ -1020,6 +1059,7 @@ def _execute_microbatch_batches( boundary="batch staging", ) if lease_failure is not None: + lifecycle.failed(error_code=lease_failure.error_code) return MicrobatchPhaseOutcome( state=state, failure=lease_failure, @@ -1032,53 +1072,26 @@ def _execute_microbatch_batches( window_text: str = f"{render(value=batch.start)}..{render(value=batch.end)}" display_window: str = _format_batch_window_for_display(batch=batch, entry=context.entry) if on_progress is not None: - on_progress(f"batch {completed_batches + 1}/{total_batches} {display_window}") - stage_failure: ModelExecutionResult | None = _stage_microbatch_delta( - context=context, - batch=batch, - window_text=window_text, - targets=batch_targets, - state=state, - ) - if stage_failure is not None: - return MicrobatchPhaseOutcome( - state=state, failure=stage_failure, completed_batches=completed_batches - ) - schema_outcome: MicrobatchSchemaPhaseOutcome = _apply_microbatch_schema_change( - context=context, - is_full_refresh=is_full_refresh, - schema_checked=schema_checked, - window_text=window_text, - targets=batch_targets, - state=state, - ) - state = schema_outcome.state - if schema_outcome.failure is not None: - return MicrobatchPhaseOutcome( - state=state, failure=schema_outcome.failure, completed_batches=completed_batches - ) - schema_checked = schema_outcome.schema_checked - type_failure: ModelExecutionResult | None = _enforce_microbatch_types( + on_progress(f"batch {batch.index + 1}/{plan.batch_count} {display_window}") + preparation: _MicrobatchBatchPreparation = _prepare_microbatch_batch( context=context, declared_columns=declared_columns, + is_full_refresh=is_full_refresh, batch=batch, + schema_checked=schema_checked, window_text=window_text, targets=batch_targets, state=state, ) - if type_failure is not None: + state = preparation.state + if preparation.failure is not None: + lifecycle.failed(error_code=preparation.failure.error_code) return MicrobatchPhaseOutcome( - state=state, failure=type_failure, completed_batches=completed_batches + state=state, + failure=preparation.failure, + completed_batches=completed_batches, ) - audit_outcome: MicrobatchPhaseOutcome = _run_microbatch_delta_audits( - context=context, - batch=batch, - targets=batch_targets, - state=state, - ) - state = audit_outcome.state - if audit_outcome.failure is not None: - return replace(audit_outcome, completed_batches=completed_batches) + schema_checked = preparation.schema_checked lease_failure = _microbatch_lease_failure( context=context, state=state, @@ -1086,6 +1099,7 @@ def _execute_microbatch_batches( boundary="target DML", ) if lease_failure is not None: + lifecycle.failed(error_code=lease_failure.error_code) return MicrobatchPhaseOutcome( state=state, failure=lease_failure, @@ -1101,6 +1115,7 @@ def _execute_microbatch_batches( state=state, ) if isinstance(dml_result, ModelExecutionResult): + lifecycle.failed(error_code=dml_result.error_code) return MicrobatchPhaseOutcome( state=state, failure=dml_result, @@ -1121,6 +1136,7 @@ def _execute_microbatch_batches( boundary="completion publication", ) if lease_failure is not None: + lifecycle.failed(error_code=lease_failure.error_code) return MicrobatchPhaseOutcome( state=state, failure=lease_failure, @@ -1145,6 +1161,7 @@ def _execute_microbatch_batches( ) ) if completion_failure is not None: + lifecycle.failed(error_code=completion_failure.error_code) return MicrobatchPhaseOutcome( state=state, failure=completion_failure, @@ -1153,18 +1170,23 @@ def _execute_microbatch_batches( total_rows=total_rows, row_count_known=row_count_known ), ) - _complete_microbatch_batch( - context=context, - window_text=window_text, - targets=batch_targets, - state=state, - ) + try: + _complete_microbatch_batch( + context=context, + window_text=window_text, + targets=batch_targets, + state=state, + ) + except BaseException as error: + lifecycle.failed(error=error) + raise + lifecycle.completed(affected_rows=dml_result if isinstance(dml_result, int) else None) completed_batches += 1 applied_intervals.append((render(value=batch.start), render(value=batch.end))) if on_progress is not None: batch_elapsed: float = time.monotonic() - batch_start_time on_progress( - f"batch {completed_batches}/{total_batches} {display_window} {batch_elapsed:.1f}s" + f"batch {batch.index + 1}/{plan.batch_count} {display_window} {batch_elapsed:.1f}s" ) return MicrobatchPhaseOutcome( state=state, @@ -1177,6 +1199,119 @@ def _execute_microbatch_batches( ) +def _microbatch_lifecycle( + *, + context: ModelMaterializationContext, + batch: BatchWindow, + history: _MicrobatchHistoryContext, + plan: _MicrobatchLifecyclePlan, +) -> MicrobatchLifecycle: + start: str = render(value=batch.start) + end: str = render(value=batch.end) + return MicrobatchLifecycle( + context=MicrobatchLifecycleContext( + cursor_start=start, + cursor_end_exclusive=end, + planned_cursor_start=plan.cursor_start, + planned_cursor_end_exclusive=plan.cursor_end_exclusive, + batch_index=batch.index + 1, + batch_count=plan.batch_count, + configured_batch_size=context.entry.batch_size or history.batch_size, + effective_batch_size=history.batch_size, + microbatch_strategy=context.entry.microbatch_strategy or "watermark", + microbatch_run_type=history.run_type.value, + batch_kind=("recovery" if (start, end) in history.recovery_intervals else "ordinary"), + ) + ) + + +def _microbatch_lifecycle_plan( + *, batch_plan: _MicrobatchPlan, cursor_type: str +) -> _MicrobatchLifecyclePlan: + selected_intervals: tuple[MicrobatchInterval, ...] = merge_intervals( + intervals=tuple( + MicrobatchInterval(start=render(value=batch.start), end=render(value=batch.end)) + for batch in batch_plan.batches + ), + cursor_type=cursor_type, + ) + if selected_intervals: + return _MicrobatchLifecyclePlan( + batch_count=len(batch_plan.batches), + cursor_start=selected_intervals[0].start, + cursor_end_exclusive=selected_intervals[-1].end, + ) + resolved_range: CursorBounds | None = batch_plan.resolved_range + if resolved_range is None: + raise ExecutorInputError("microbatch lifecycle requires resolved cursor bounds") + return _MicrobatchLifecyclePlan( + batch_count=len(batch_plan.batches), + cursor_start=render(value=_concrete_bound(bound=resolved_range.start)), + cursor_end_exclusive=render(value=_concrete_bound(bound=resolved_range.end)), + ) + + +def _prepare_microbatch_batch( + *, + context: ModelMaterializationContext, + declared_columns: tuple[ColumnInfo, ...], + is_full_refresh: bool, + batch: BatchWindow, + schema_checked: bool, + window_text: str, + targets: MicrobatchTargets, + state: MicrobatchLifecycleState, +) -> _MicrobatchBatchPreparation: + stage_failure: ModelExecutionResult | None = _stage_microbatch_delta( + context=context, + batch=batch, + window_text=window_text, + targets=targets, + state=state, + ) + if stage_failure is not None: + return _MicrobatchBatchPreparation(state, schema_checked, stage_failure) + schema_outcome: MicrobatchSchemaPhaseOutcome = _apply_microbatch_schema_change( + context=context, + is_full_refresh=is_full_refresh, + schema_checked=schema_checked, + window_text=window_text, + targets=targets, + state=state, + ) + if schema_outcome.failure is not None: + return _MicrobatchBatchPreparation( + schema_outcome.state, + schema_outcome.schema_checked, + schema_outcome.failure, + ) + type_failure: ModelExecutionResult | None = _enforce_microbatch_types( + context=context, + declared_columns=declared_columns, + batch=batch, + window_text=window_text, + targets=targets, + state=schema_outcome.state, + ) + if type_failure is not None: + return _MicrobatchBatchPreparation( + schema_outcome.state, + schema_outcome.schema_checked, + type_failure, + ) + audit_outcome: MicrobatchPhaseOutcome = _run_microbatch_delta_audits( + context=context, + batch=batch, + targets=targets, + state=schema_outcome.state, + ) + return _MicrobatchBatchPreparation( + audit_outcome.state, + schema_outcome.schema_checked, + audit_outcome.failure, + ) + + def _microbatch_lease_failure( *, context: ModelMaterializationContext, @@ -1211,6 +1346,7 @@ def _execute_microbatch_batches_concurrently( state: MicrobatchLifecycleState, history_context: _MicrobatchHistoryContext, on_progress: Callable[[str], None] | None, + lifecycle_plan: _MicrobatchLifecyclePlan, ) -> MicrobatchPhaseOutcome: recovery_batches: tuple[BatchWindow, ...] = tuple( batch @@ -1239,6 +1375,7 @@ def _execute_microbatch_batches_concurrently( aggregate_state=state, history_context=history_context, on_progress=on_progress, + lifecycle_plan=lifecycle_plan, ) state = phase_outcome.state completed_batches += phase_outcome.completed_batches @@ -1275,6 +1412,7 @@ def _execute_concurrent_microbatch_phase( aggregate_state: MicrobatchLifecycleState, history_context: _MicrobatchHistoryContext, on_progress: Callable[[str], None] | None, + lifecycle_plan: _MicrobatchLifecyclePlan, ) -> MicrobatchPhaseOutcome: first: BatchWindow = batches[0] serial_context: ModelMaterializationContext = replace(context, microbatch_batch_runner=None) @@ -1287,6 +1425,7 @@ def _execute_concurrent_microbatch_phase( state=aggregate_state, history_context=history_context, on_progress=on_progress, + lifecycle_plan=lifecycle_plan, ) if first_outcome.failure is not None or len(batches) == 1: return first_outcome @@ -1317,6 +1456,7 @@ def execute(*, batch: BatchWindow, connection: Any) -> MicrobatchPhaseOutcome: state=worker_state, history_context=replace(history_context, store=worker_store), on_progress=on_progress, + lifecycle_plan=lifecycle_plan, ) runner: MicrobatchBatchRunner | None = context.microbatch_batch_runner diff --git a/src/sqlbuild/runtime/event_exporting/types.py b/src/sqlbuild/runtime/event_exporting/types.py index c1796d14ce..3a4046999e 100644 --- a/src/sqlbuild/runtime/event_exporting/types.py +++ b/src/sqlbuild/runtime/event_exporting/types.py @@ -9,6 +9,7 @@ class LifecycleEventKind(StrEnum): INVOCATION = "invocation" RUN = "run" RESOURCE = "resource" + MICROBATCH = "microbatch" OPERATION = "operation" STATEMENT = "statement" RETRY = "retry" diff --git a/src/sqlbuild/runtime/observability/_helpers/validation.py b/src/sqlbuild/runtime/observability/_helpers/validation.py index 61e144010d..6e752841a3 100644 --- a/src/sqlbuild/runtime/observability/_helpers/validation.py +++ b/src/sqlbuild/runtime/observability/_helpers/validation.py @@ -23,6 +23,10 @@ LIFECYCLE_EVENT_CATALOGS, MAX_METADATA_BYTES, METADATA_FIELD, + MICROBATCH_EVENT_PREFIX, + MICROBATCH_KINDS, + MICROBATCH_RUN_TYPES, + MICROBATCH_STRATEGIES, NONNEGATIVE_INTEGER_PAYLOAD_FIELDS, OPERATION_ADAPTERS, OPERATION_EVENT_PREFIX, @@ -295,6 +299,8 @@ def validate_known_lifecycle_event(*, event: LifecycleEvent) -> None: ) if event.event_type.startswith(OPERATION_EVENT_PREFIX): _validate_operation_payload(payload=event.payload) + if event.event_type.startswith(MICROBATCH_EVENT_PREFIX): + _validate_microbatch_payload(payload=event.payload) if event.event_type == RESOURCE_ATTEMPT_SKIPPED_EVENT: if event.payload.get("skip_code") not in RESOURCE_SKIP_CODES: raise ObservabilityValidationError("skip_code must be a catalogued value") @@ -323,6 +329,25 @@ def validate_known_lifecycle_event(*, event: LifecycleEvent) -> None: _validate_run_payload(event=event) +def _validate_microbatch_payload(*, payload: Mapping[str, JSONValue]) -> None: + batch_index: JSONValue | None = payload.get("batch_index") + batch_count: JSONValue | None = payload.get("batch_count") + if isinstance(batch_index, bool) or not isinstance(batch_index, int) or batch_index < 1: + raise ObservabilityValidationError("batch_index must be a positive integer") + if isinstance(batch_count, bool) or not isinstance(batch_count, int) or batch_count < 1: + raise ObservabilityValidationError("batch_count must be a positive integer") + if batch_index > batch_count: + raise ObservabilityValidationError("batch_index must not exceed batch_count") + for field_name, allowed in ( + ("microbatch_strategy", MICROBATCH_STRATEGIES), + ("microbatch_run_type", MICROBATCH_RUN_TYPES), + ("batch_kind", MICROBATCH_KINDS), + ): + value: JSONValue | None = payload.get(field_name) + if not isinstance(value, str) or value not in allowed: + raise ObservabilityValidationError(f"{field_name} must be a catalogued value") + + def _validate_run_payload(*, event: LifecycleEvent) -> None: payload: Mapping[str, JSONValue] = event.payload if event.event_type == RUN_STARTED_EVENT and CONFIGURED_CONCURRENCY_FIELD in payload: diff --git a/src/sqlbuild/runtime/observability/classes/microbatch_lifecycle.py b/src/sqlbuild/runtime/observability/classes/microbatch_lifecycle.py new file mode 100644 index 0000000000..3a472e52e6 --- /dev/null +++ b/src/sqlbuild/runtime/observability/classes/microbatch_lifecycle.py @@ -0,0 +1,125 @@ +"""Lifecycle facts for one active microbatch interval.""" + +from __future__ import annotations + +import re +import time +from collections.abc import Mapping +from types import TracebackType + +from sqlbuild.runtime.observability._helpers.dispatcher import current_event_dispatcher +from sqlbuild.runtime.observability._helpers.factory import create_lifecycle_event +from sqlbuild.runtime.observability._helpers.identity import current_execution_identity +from sqlbuild.runtime.observability.classes.event_dispatcher import EventDispatcher +from sqlbuild.runtime.observability.exceptions import ObservabilityValidationError +from sqlbuild.runtime.observability.models import ExecutionIdentity, MicrobatchLifecycleContext +from sqlbuild.runtime.observability.types import JSONValue + +_ERROR_TOKEN_PATTERN: re.Pattern[str] = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") + + +class MicrobatchLifecycle: + """Publish the active bounds and terminal outcome for one microbatch.""" + + def __init__( + self, + *, + context: MicrobatchLifecycleContext, + ) -> None: + self._payload: dict[str, JSONValue] = context.payload() + self._dispatcher: EventDispatcher | None = None + self._started_monotonic: float | None = None + self._terminal: bool = False + self._entered: bool = False + + def __enter__(self) -> MicrobatchLifecycle: + if self._entered: + raise ObservabilityValidationError( + "microbatch lifecycle cannot be entered more than once" + ) + self._entered = True + identity: ExecutionIdentity | None = current_execution_identity() + dispatcher: EventDispatcher | None = current_event_dispatcher() + if ( + identity is None + or identity.run_id is None + or identity.resource_id is None + or identity.resource_attempt_id is None + or dispatcher is None + ): + return self + self._dispatcher = dispatcher + self._started_monotonic = time.monotonic() + self._publish(event_type="microbatch_started", payload=self._payload) + return self + + def start(self) -> None: + """Publish the start fact without requiring context-manager ownership.""" + + self.__enter__() + + def completed(self, *, affected_rows: int | None = None) -> None: + payload: dict[str, JSONValue] = self._terminal_payload() + if affected_rows is not None: + payload["affected_rows"] = affected_rows + self._publish_terminal(event_type="microbatch_completed", payload=payload) + + def failed( + self, + *, + error: BaseException | None = None, + error_code: str | None = None, + ) -> None: + payload: dict[str, JSONValue] = self._terminal_payload() + payload["error_type"] = _safe_error_token( + value="ExecutionFailed" if error is None else type(error).__name__, + fallback="ExecutionFailed", + ) + safe_error_code: str | None = _safe_optional_error_code(error_code) + if safe_error_code is not None: + payload["error_code"] = safe_error_code + self._publish_terminal(event_type="microbatch_failed", payload=payload) + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + del exc_type, traceback + if not self._terminal: + if exc_value is None: + self.completed() + else: + self.failed(error=exc_value) + + def _terminal_payload(self) -> dict[str, JSONValue]: + payload: dict[str, JSONValue] = dict(self._payload) + started: float = self._started_monotonic if self._started_monotonic is not None else 0.0 + payload["duration_ms"] = max(0.0, (time.monotonic() - started) * 1000.0) + return payload + + def _publish(self, *, event_type: str, payload: Mapping[str, JSONValue]) -> None: + if self._dispatcher is None: + return + self._dispatcher.publish_lifecycle( + create_lifecycle_event(event_type=event_type, payload=payload) + ) + + def _publish_terminal(self, *, event_type: str, payload: Mapping[str, JSONValue]) -> None: + if self._terminal: + return + self._terminal = True + self._publish(event_type=event_type, payload=payload) + + +def _safe_error_token(*, value: object, fallback: str) -> str: + if not isinstance(value, str) or _ERROR_TOKEN_PATTERN.fullmatch(value) is None: + return fallback + return value + + +def _safe_optional_error_code(value: str | None) -> str | None: + if value is None or _ERROR_TOKEN_PATTERN.fullmatch(value) is None: + return None + return value diff --git a/src/sqlbuild/runtime/observability/constants.py b/src/sqlbuild/runtime/observability/constants.py index 271db54ab9..ea1ddc81e9 100644 --- a/src/sqlbuild/runtime/observability/constants.py +++ b/src/sqlbuild/runtime/observability/constants.py @@ -31,6 +31,25 @@ RESOURCE_TERMINALS: frozenset[str] = frozenset( {"resource_attempt_completed", "resource_attempt_failed", RESOURCE_ATTEMPT_SKIPPED_EVENT} ) +MICROBATCH_EVENT_PREFIX: str = "microbatch_" +MICROBATCH_FIELDS: frozenset[str] = frozenset( + { + "cursor_start", + "cursor_end_exclusive", + "planned_cursor_start", + "planned_cursor_end_exclusive", + "batch_index", + "batch_count", + "configured_batch_size", + "effective_batch_size", + "microbatch_strategy", + "microbatch_run_type", + "batch_kind", + } +) +MICROBATCH_STRATEGIES: frozenset[str] = frozenset({"rolling_window", "watermark"}) +MICROBATCH_RUN_TYPES: frozenset[str] = frozenset({"normal", "backfill", "replay_on_change"}) +MICROBATCH_KINDS: frozenset[str] = frozenset({"ordinary", "recovery"}) HOOK_PHASES: frozenset[str] = frozenset({"post_hooks", "pre_hooks"}) HOOK_TYPES: frozenset[str] = frozenset({"python", "sql"}) OPERATION_FIELDS: frozenset[str] = frozenset( @@ -253,6 +272,15 @@ "executed_sql", "execution_error", "result_id", + "cursor_start", + "cursor_end_exclusive", + "planned_cursor_start", + "planned_cursor_end_exclusive", + "configured_batch_size", + "effective_batch_size", + "microbatch_strategy", + "microbatch_run_type", + "batch_kind", } ) NONNEGATIVE_INTEGER_PAYLOAD_FIELDS: frozenset[str] = frozenset( @@ -280,6 +308,8 @@ "sample_count", "minimum_samples", "evidence_count", + "batch_index", + "batch_count", } ) FINITE_NUMBER_PAYLOAD_FIELDS: frozenset[str] = frozenset({"measured_value"}) @@ -523,7 +553,29 @@ ), } ) +LIFECYCLE_EVENT_CATALOG_V2: Mapping[str, LifecycleEventDefinition] = MappingProxyType( + { + **LIFECYCLE_EVENT_CATALOG_V1, + "microbatch_started": LifecycleEventDefinition.create( + required_correlations=frozenset({"run_id", "resource_id", "resource_attempt_id"}), + required_payload=MICROBATCH_FIELDS, + allowed=MICROBATCH_FIELDS, + ), + "microbatch_completed": LifecycleEventDefinition.create( + required_correlations=frozenset({"run_id", "resource_id", "resource_attempt_id"}), + required_payload=MICROBATCH_FIELDS, + allowed=MICROBATCH_FIELDS | DURATION_FIELDS | frozenset({"affected_rows"}), + terminal=True, + ), + "microbatch_failed": LifecycleEventDefinition.create( + required_correlations=frozenset({"run_id", "resource_id", "resource_attempt_id"}), + required_payload=MICROBATCH_FIELDS | frozenset({"error_type"}), + allowed=MICROBATCH_FIELDS | DURATION_FIELDS | ERROR_FIELDS, + terminal=True, + ), + } +) LIFECYCLE_EVENT_CATALOGS: Mapping[int, Mapping[str, LifecycleEventDefinition]] = MappingProxyType( - {1: LIFECYCLE_EVENT_CATALOG_V1, 2: LIFECYCLE_EVENT_CATALOG_V1} + {1: LIFECYCLE_EVENT_CATALOG_V1, 2: LIFECYCLE_EVENT_CATALOG_V2} ) -LIFECYCLE_EVENT_CATALOG: Mapping[str, LifecycleEventDefinition] = LIFECYCLE_EVENT_CATALOG_V1 +LIFECYCLE_EVENT_CATALOG: Mapping[str, LifecycleEventDefinition] = LIFECYCLE_EVENT_CATALOG_V2 diff --git a/src/sqlbuild/runtime/observability/models.py b/src/sqlbuild/runtime/observability/models.py index 552007b5a8..0e937361d4 100644 --- a/src/sqlbuild/runtime/observability/models.py +++ b/src/sqlbuild/runtime/observability/models.py @@ -22,6 +22,38 @@ class OperationAttributes: scope: str | None = None +@dataclass(frozen=True) +class MicrobatchLifecycleContext: + """Bounded execution context repeated across one microbatch lifecycle.""" + + cursor_start: str + cursor_end_exclusive: str + planned_cursor_start: str + planned_cursor_end_exclusive: str + batch_index: int + batch_count: int + configured_batch_size: str + effective_batch_size: str + microbatch_strategy: str + microbatch_run_type: str + batch_kind: str + + def payload(self) -> dict[str, JSONValue]: + return { + "cursor_start": self.cursor_start, + "cursor_end_exclusive": self.cursor_end_exclusive, + "planned_cursor_start": self.planned_cursor_start, + "planned_cursor_end_exclusive": self.planned_cursor_end_exclusive, + "batch_index": self.batch_index, + "batch_count": self.batch_count, + "configured_batch_size": self.configured_batch_size, + "effective_batch_size": self.effective_batch_size, + "microbatch_strategy": self.microbatch_strategy, + "microbatch_run_type": self.microbatch_run_type, + "batch_kind": self.batch_kind, + } + + @dataclass(frozen=True) class ExecutionIdentity: """Immutable correlation identity for one point in an execution hierarchy.""" diff --git a/tests/integration/src/sqlbuild/executor/run/microbatch/_test_types.py b/tests/integration/src/sqlbuild/executor/run/microbatch/_test_types.py index f1309465e3..5135161793 100644 --- a/tests/integration/src/sqlbuild/executor/run/microbatch/_test_types.py +++ b/tests/integration/src/sqlbuild/executor/run/microbatch/_test_types.py @@ -100,6 +100,13 @@ class MicrobatchSuccessTestCase: microbatch_limit_action: MicrobatchLimitAction | None = None expected_microbatch_limit_count: int | None = None expected_microbatch_limit_warning: bool = False + expected_lifecycle_event_types: tuple[str, ...] = field(default_factory=tuple) + expected_lifecycle_batch_indexes: tuple[int, ...] = field(default_factory=tuple) + expected_lifecycle_cursor_starts: tuple[str, ...] = field(default_factory=tuple) + expected_lifecycle_error_type: str | None = None + expected_lifecycle_batch_count: int | None = None + expected_lifecycle_planned_cursor_start: str | None = None + expected_lifecycle_planned_cursor_end: str | None = None @dataclass(frozen=True) @@ -148,3 +155,10 @@ class MicrobatchFailureTestCase: microbatch_limit_action: MicrobatchLimitAction | None = None expected_microbatch_limit_count: int | None = None expected_microbatch_limit_warning: bool = False + expected_lifecycle_event_types: tuple[str, ...] = field(default_factory=tuple) + expected_lifecycle_batch_indexes: tuple[int, ...] = field(default_factory=tuple) + expected_lifecycle_cursor_starts: tuple[str, ...] = field(default_factory=tuple) + expected_lifecycle_error_type: str | None = None + expected_lifecycle_batch_count: int | None = None + expected_lifecycle_planned_cursor_start: str | None = None + expected_lifecycle_planned_cursor_end: str | None = None diff --git a/tests/integration/src/sqlbuild/executor/run/microbatch/helpers.py b/tests/integration/src/sqlbuild/executor/run/microbatch/helpers.py index 3f5e8e2fbb..599bce595a 100644 --- a/tests/integration/src/sqlbuild/executor/run/microbatch/helpers.py +++ b/tests/integration/src/sqlbuild/executor/run/microbatch/helpers.py @@ -47,6 +47,7 @@ RuntimeCursorSpec, ) from sqlbuild.executor.scheduling.types import ExecutionStatus +from sqlbuild.observability import EventDispatcher, LifecycleEvent from tests.integration.src.sqlbuild.executor.run.microbatch._test_types import ( MicrobatchFailureTestCase, MicrobatchSuccessTestCase, @@ -59,6 +60,29 @@ } +def capture_lifecycle_events() -> tuple[EventDispatcher, list[LifecycleEvent]]: + events: list[LifecycleEvent] = [] + dispatcher: EventDispatcher = EventDispatcher() + dispatcher.subscribe_lifecycle(subscriber=events.append, accepts_opaque=False) + return dispatcher, events + + +def lifecycle_events_with_prefix( + *, events: list[LifecycleEvent], prefix: str +) -> tuple[LifecycleEvent, ...]: + return tuple(filter(lambda event: event.event_type.startswith(prefix), events)) + + +def lifecycle_events_by_type( + *, events: tuple[LifecycleEvent, ...], event_type: str +) -> tuple[LifecycleEvent, ...]: + return tuple(filter(lambda event: event.event_type == event_type, events)) + + +def fail_microbatch_cleanup(**_: object) -> None: + raise RuntimeError("cleanup failed") + + def resolve_nonempty_terminal_bounds( *, adapter: DuckDbAdapter, connection: Any, mode: str ) -> CursorBounds | None: diff --git a/tests/integration/src/sqlbuild/executor/run/microbatch/test_microbatch_observability.py b/tests/integration/src/sqlbuild/executor/run/microbatch/test_microbatch_observability.py new file mode 100644 index 0000000000..afc61aa22c --- /dev/null +++ b/tests/integration/src/sqlbuild/executor/run/microbatch/test_microbatch_observability.py @@ -0,0 +1,324 @@ +"""Integration coverage for microbatch lifecycle observability.""" + +from __future__ import annotations + +from functools import partial +from typing import Any + +import pytest + +from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter +from sqlbuild.executor.run._helpers.materializations import microbatch as microbatch_module +from sqlbuild.executor.run.models import BatchWindow +from sqlbuild.executor.run.types import ExecutionPhase +from sqlbuild.observability import ( + LifecycleEvent, + dispatcher_scope, + invocation_scope, + resource_attempt_scope, + run_scope, +) +from sqlbuild.spec.contracts.types import MicrobatchLimitAction +from tests.integration.src.sqlbuild.executor.run.microbatch._test_types import ( + MicrobatchFailureTestCase, + MicrobatchSuccessTestCase, +) +from tests.integration.src.sqlbuild.executor.run.microbatch.helpers import ( + capture_lifecycle_events, + fail_microbatch_cleanup, + lifecycle_events_by_type, + lifecycle_events_with_prefix, + reconcile_microbatch_batches, + run_failure_test, + run_success_test, +) + +_SOURCE_SQL: str = "CREATE TABLE main.raw_orders (id INTEGER, ordered_at TIMESTAMP)" +_SOURCE_DATA: str = ( + "INSERT INTO main.raw_orders VALUES " + "(1, '2026-01-01 00:30:00'), " + "(2, '2026-01-01 01:30:00'), " + "(3, '2026-01-01 02:30:00')" +) +_MODEL_SQL: str = ( + "SELECT id, ordered_at FROM main.raw_orders " + "WHERE ordered_at >= '__SQB_CURSOR_START__' " + "AND ordered_at < '__SQB_CURSOR_END__'" +) + + +@pytest.mark.parametrize( + "test_case", + ( + MicrobatchSuccessTestCase( + description="three observable hourly batches", + setup_sql=( + _SOURCE_SQL, + _SOURCE_DATA, + "CREATE TABLE main.daily_orders (id INTEGER, ordered_at TIMESTAMP)", + ), + model_sql=_MODEL_SQL, + target_schema="main", + target_name="daily_orders", + incremental_strategy="delete_insert", + cursor_column="ordered_at", + cursor_type="timestamp", + batch_size="1h", + microbatch_start="2026-01-01T00:00:00", + microbatch_end="2026-01-01T03:00:00", + expected_row_count=3, + expected_lifecycle_event_types=( + "microbatch_started", + "microbatch_completed", + "microbatch_started", + "microbatch_completed", + "microbatch_started", + "microbatch_completed", + ), + expected_lifecycle_batch_indexes=(1, 2, 3), + expected_lifecycle_cursor_starts=( + "2026-01-01T00:00:00", + "2026-01-01T01:00:00", + "2026-01-01T02:00:00", + ), + expected_lifecycle_batch_count=3, + expected_lifecycle_planned_cursor_end="2026-01-01T03:00:00", + ), + ), + ids=lambda case: case.description, +) +def test_given_three_batch_model_when_executing_then_each_active_interval_is_observable( + test_case: MicrobatchSuccessTestCase, + adapter: DuckDbAdapter, + connection: Any, +) -> None: + dispatcher, events = capture_lifecycle_events() + + with ( + invocation_scope("invocation-1"), + run_scope("test_run"), + resource_attempt_scope(resource_id="model:daily_orders", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + ): + run_success_test(test_case=test_case, adapter=adapter, connection=connection) + + lifecycle_events: tuple[LifecycleEvent, ...] = lifecycle_events_with_prefix( + events=events, prefix="microbatch_" + ) + assert ( + tuple(event.event_type for event in lifecycle_events) + == test_case.expected_lifecycle_event_types + ) + started: tuple[LifecycleEvent, ...] = lifecycle_events[::2] + assert ( + tuple(event.payload["batch_index"] for event in started) + == test_case.expected_lifecycle_batch_indexes + ) + assert all( + event.payload["batch_count"] == test_case.expected_lifecycle_batch_count + for event in started + ) + assert all(event.payload["configured_batch_size"] == "1h" for event in started) + assert all(event.payload["effective_batch_size"] == "1h" for event in started) + assert ( + tuple(event.payload["cursor_start"] for event in started) + == test_case.expected_lifecycle_cursor_starts + ) + assert all( + event.payload["planned_cursor_end_exclusive"] + == test_case.expected_lifecycle_planned_cursor_end + for event in started + ) + + +@pytest.mark.parametrize( + "test_case", + ( + MicrobatchFailureTestCase( + description="observable staging failure", + setup_sql=( + _SOURCE_SQL, + _SOURCE_DATA, + "CREATE TABLE main.daily_orders (id INTEGER, ordered_at TIMESTAMP)", + ), + model_sql=_MODEL_SQL.replace( + "SELECT id, ordered_at", "SELECT missing_column, ordered_at" + ), + target_schema="main", + target_name="daily_orders", + incremental_strategy="delete_insert", + cursor_column="ordered_at", + cursor_type="timestamp", + batch_size="1h", + microbatch_start="2026-01-01T00:00:00", + microbatch_end="2026-01-01T03:00:00", + expected_failed_phase=ExecutionPhase.STAGING, + expected_lifecycle_event_types=("microbatch_started", "microbatch_failed"), + expected_lifecycle_batch_indexes=(1,), + expected_lifecycle_batch_count=3, + ), + ), + ids=lambda case: case.description, +) +def test_given_staging_failure_when_executing_then_active_interval_has_failed_terminal( + test_case: MicrobatchFailureTestCase, + adapter: DuckDbAdapter, + connection: Any, +) -> None: + dispatcher, events = capture_lifecycle_events() + + with ( + invocation_scope("invocation-1"), + run_scope("test_run"), + resource_attempt_scope(resource_id="model:daily_orders", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + ): + run_failure_test(test_case=test_case, adapter=adapter, connection=connection) + + lifecycle_events: tuple[LifecycleEvent, ...] = lifecycle_events_with_prefix( + events=events, prefix="microbatch_" + ) + assert ( + tuple(event.event_type for event in lifecycle_events) + == test_case.expected_lifecycle_event_types + ) + assert ( + lifecycle_events[0].payload["cursor_start"] == lifecycle_events[1].payload["cursor_start"] + ) + assert ( + lifecycle_events[1].payload["batch_index"] == test_case.expected_lifecycle_batch_indexes[0] + ) + assert lifecycle_events[1].payload["batch_count"] == test_case.expected_lifecycle_batch_count + + +@pytest.mark.parametrize( + "test_case", + ( + MicrobatchSuccessTestCase( + description="observable cleanup exception", + setup_sql=( + _SOURCE_SQL, + _SOURCE_DATA, + "CREATE TABLE main.daily_orders (id INTEGER, ordered_at TIMESTAMP)", + ), + model_sql=_MODEL_SQL, + target_schema="main", + target_name="daily_orders", + incremental_strategy="delete_insert", + cursor_column="ordered_at", + cursor_type="timestamp", + batch_size="1h", + microbatch_start="2026-01-01T00:00:00", + microbatch_end="2026-01-01T03:00:00", + expected_row_count=3, + expected_lifecycle_event_types=("microbatch_started", "microbatch_failed"), + expected_lifecycle_error_type="RuntimeError", + ), + ), + ids=lambda case: case.description, +) +def test_given_cleanup_exception_when_executing_then_failed_terminal_precedes_propagation( + test_case: MicrobatchSuccessTestCase, + adapter: DuckDbAdapter, + connection: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + dispatcher, events = capture_lifecycle_events() + monkeypatch.setattr(microbatch_module, "_complete_microbatch_batch", fail_microbatch_cleanup) + with ( + invocation_scope("invocation-1"), + run_scope("test_run"), + resource_attempt_scope(resource_id="model:daily_orders", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + pytest.raises(RuntimeError, match="cleanup failed"), + ): + run_success_test(test_case=test_case, adapter=adapter, connection=connection) + + lifecycle_events: tuple[LifecycleEvent, ...] = lifecycle_events_with_prefix( + events=events, prefix="microbatch_" + ) + assert ( + tuple(event.event_type for event in lifecycle_events) + == test_case.expected_lifecycle_event_types + ) + assert lifecycle_events[1].payload["error_type"] == test_case.expected_lifecycle_error_type + + +@pytest.mark.parametrize( + "test_case", + ( + MicrobatchSuccessTestCase( + description="out of order reconciliation envelope", + setup_sql=( + _SOURCE_SQL, + _SOURCE_DATA, + "CREATE TABLE main.daily_orders (id INTEGER, ordered_at TIMESTAMP)", + ), + model_sql=_MODEL_SQL, + target_schema="main", + target_name="daily_orders", + incremental_strategy="delete_insert", + cursor_column="ordered_at", + cursor_type="timestamp", + batch_size="1h", + microbatch_start="2026-01-01T00:00:00", + microbatch_end="2026-01-01T03:00:00", + expected_row_count=2, + microbatch_limit=2, + microbatch_limit_action=MicrobatchLimitAction.CAP_FROM_START, + expected_lifecycle_batch_indexes=(1, 2), + expected_lifecycle_cursor_starts=( + "2026-01-01T02:00:00", + "2026-01-01T00:00:00", + ), + expected_lifecycle_planned_cursor_start="2026-01-01T00:00:00", + expected_lifecycle_planned_cursor_end="2026-01-01T03:00:00", + ), + ), + ids=lambda case: case.description, +) +def test_given_out_of_order_reconciliation_when_executing_then_planned_bounds_use_cursor_envelope( + test_case: MicrobatchSuccessTestCase, + adapter: DuckDbAdapter, + connection: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + dispatcher, events = capture_lifecycle_events() + reconciled_batches: tuple[BatchWindow, ...] = ( + BatchWindow(start="2026-01-01T02:00:00", end="2026-01-01T03:00:00", index=0), + BatchWindow(start="2026-01-01T00:00:00", end="2026-01-01T01:00:00", index=1), + BatchWindow(start="2026-01-01T01:00:00", end="2026-01-01T02:00:00", index=2), + ) + monkeypatch.setattr( + microbatch_module, + "_run_microbatch_reconciliation", + partial(reconcile_microbatch_batches, reconciled_batches=reconciled_batches), + ) + + with ( + invocation_scope("invocation-1"), + run_scope("test_run"), + resource_attempt_scope(resource_id="model:daily_orders", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + ): + run_success_test(test_case=test_case, adapter=adapter, connection=connection) + + lifecycle_events: tuple[LifecycleEvent, ...] = lifecycle_events_with_prefix( + events=events, prefix="microbatch_" + ) + started: tuple[LifecycleEvent, ...] = lifecycle_events_by_type( + events=lifecycle_events, event_type="microbatch_started" + ) + assert tuple(event.payload["batch_index"] for event in started) == ( + test_case.expected_lifecycle_batch_indexes + ) + assert ( + tuple(event.payload["cursor_start"] for event in started) + == test_case.expected_lifecycle_cursor_starts + ) + assert all( + event.payload["planned_cursor_start"] == test_case.expected_lifecycle_planned_cursor_start + and event.payload["planned_cursor_end_exclusive"] + == test_case.expected_lifecycle_planned_cursor_end + for event in started + ) diff --git a/tests/unit/src/sqlbuild/runtime/event_exporting/test_policy.py b/tests/unit/src/sqlbuild/runtime/event_exporting/test_policy.py index b45fcee1b0..dec74484d6 100644 --- a/tests/unit/src/sqlbuild/runtime/event_exporting/test_policy.py +++ b/tests/unit/src/sqlbuild/runtime/event_exporting/test_policy.py @@ -32,11 +32,14 @@ def test_given_canonical_catalog_when_building_export_policy_then_every_type_is_ assert policy["statement_submitted"].severity == "debug" assert policy["audit_completed"].kind == LifecycleEventKind.AUDIT assert policy["audit_completed"].severity == "info" + assert policy["microbatch_started"].kind == LifecycleEventKind.MICROBATCH + assert policy["microbatch_started"].severity == "debug" + assert policy["microbatch_completed"].severity == "info" @pytest.mark.parametrize( "test_case", - (EventExportPolicyTestCase("typed kind vocabulary is exhaustive", 7),), + (EventExportPolicyTestCase("typed kind vocabulary is exhaustive", 8),), ids=lambda case: case.description, ) def test_given_lifecycle_catalog_when_deriving_export_dimensions_then_kind_enum_is_exhaustive( diff --git a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py index 9463a71998..d272b95ae8 100644 --- a/tests/unit/src/sqlbuild/runtime/observability/_test_types.py +++ b/tests/unit/src/sqlbuild/runtime/observability/_test_types.py @@ -75,6 +75,14 @@ class CatalogVersionCase: expected_unsupported_version: int +@dataclass(frozen=True) +class CatalogEventPresenceCase: + description: str + schema_version: int + event_type: str + expected_present: bool + + @dataclass(frozen=True) class IdempotencyCase: description: str diff --git a/tests/unit/src/sqlbuild/runtime/observability/helpers.py b/tests/unit/src/sqlbuild/runtime/observability/helpers.py index cea7b07aba..0a9495eed4 100644 --- a/tests/unit/src/sqlbuild/runtime/observability/helpers.py +++ b/tests/unit/src/sqlbuild/runtime/observability/helpers.py @@ -12,10 +12,12 @@ from sqlbuild.runtime.observability._helpers.factory import create_lifecycle_event from sqlbuild.runtime.observability._helpers.identity import invocation_scope from sqlbuild.runtime.observability.classes.event_dispatcher import EventDispatcher +from sqlbuild.runtime.observability.classes.microbatch_lifecycle import MicrobatchLifecycle from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle from sqlbuild.runtime.observability.models import ( DiagnosticLog, LifecycleEvent, + MicrobatchLifecycleContext, OpaqueLifecycleEvent, ) from sqlbuild.runtime.observability.types import JSONValue @@ -23,6 +25,31 @@ OCCURRED_AT: datetime = datetime(2026, 8, 31, 12, 34, 56, 123456, tzinfo=UTC) +def capture_lifecycle_events() -> tuple[EventDispatcher, list[LifecycleEvent]]: + events: list[LifecycleEvent] = [] + dispatcher: EventDispatcher = EventDispatcher() + dispatcher.subscribe_lifecycle(subscriber=events.append, accepts_opaque=False) + return dispatcher, events + + +def microbatch_lifecycle() -> MicrobatchLifecycle: + return MicrobatchLifecycle( + context=MicrobatchLifecycleContext( + cursor_start="2026-01-01T00:00:00", + cursor_end_exclusive="2026-01-02T00:00:00", + planned_cursor_start="2026-01-01T00:00:00", + planned_cursor_end_exclusive="2026-01-04T00:00:00", + batch_index=1, + batch_count=3, + configured_batch_size="1d", + effective_batch_size="1d", + microbatch_strategy="watermark", + microbatch_run_type="normal", + batch_kind="ordinary", + ) + ) + + async def delayed_statement(*, release: asyncio.Event, query_id: str) -> None: await release.wait() with StatementLifecycle(adapter="async", sql="SELECT delayed", intent="execute") as lifecycle: diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_catalog.py b/tests/unit/src/sqlbuild/runtime/observability/test_catalog.py index e3c95d68cf..52e65f7816 100644 --- a/tests/unit/src/sqlbuild/runtime/observability/test_catalog.py +++ b/tests/unit/src/sqlbuild/runtime/observability/test_catalog.py @@ -19,6 +19,7 @@ from sqlbuild.runtime.observability.models import LifecycleEvent from sqlbuild.runtime.observability.types import JSONValue from tests.unit.src.sqlbuild.runtime.observability._test_types import ( + CatalogEventPresenceCase, CatalogVersionCase, IdempotencyCase, LifecycleErrorCase, @@ -376,6 +377,20 @@ def test_given_only_started_and_submitted_facts_when_inspecting_then_no_terminal expected_terminal=True, expected_unsupported_version=3, ), + CatalogVersionCase( + description="microbatch completion is terminal in v2", + schema_version=2, + event_type="microbatch_completed", + expected_terminal=True, + expected_unsupported_version=3, + ), + CatalogVersionCase( + description="microbatch failure is terminal in v2", + schema_version=2, + event_type="microbatch_failed", + expected_terminal=True, + expected_unsupported_version=3, + ), ], ids=lambda case: case.description, ) @@ -389,6 +404,32 @@ def test_given_catalog_when_looking_up_events_then_authority_is_version_dimensio assert test_case.expected_unsupported_version not in LIFECYCLE_EVENT_CATALOGS +@pytest.mark.parametrize( + "test_case", + ( + CatalogEventPresenceCase( + description="v1 excludes microbatch events", + schema_version=1, + event_type="microbatch_started", + expected_present=False, + ), + CatalogEventPresenceCase( + description="v2 includes microbatch events", + schema_version=2, + event_type="microbatch_started", + expected_present=True, + ), + ), + ids=lambda case: case.description, +) +def test_given_microbatch_events_when_inspecting_versioned_catalog_then_v1_remains_unchanged( + test_case: CatalogEventPresenceCase, +) -> None: + assert ( + test_case.event_type in LIFECYCLE_EVENT_CATALOGS[test_case.schema_version] + ) is test_case.expected_present + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/runtime/observability/test_microbatch_lifecycle.py b/tests/unit/src/sqlbuild/runtime/observability/test_microbatch_lifecycle.py new file mode 100644 index 0000000000..a697411ff4 --- /dev/null +++ b/tests/unit/src/sqlbuild/runtime/observability/test_microbatch_lifecycle.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +from collections import Counter +from concurrent.futures import Future, ThreadPoolExecutor +from contextvars import copy_context +from typing import cast + +import pytest + +from sqlbuild.observability import ( + dispatcher_scope, + invocation_scope, + resource_attempt_scope, + run_scope, +) +from sqlbuild.runtime.observability.classes.microbatch_lifecycle import ( + MicrobatchLifecycle, +) +from sqlbuild.runtime.observability.exceptions import ObservabilityValidationError +from sqlbuild.runtime.observability.main.is_terminal_event import is_terminal_event +from sqlbuild.runtime.observability.models import MicrobatchLifecycleContext +from tests.unit.src.sqlbuild.runtime.observability._test_types import ( + LifecycleErrorCase, + OperationLifecycleCase, +) +from tests.unit.src.sqlbuild.runtime.observability.helpers import ( + capture_lifecycle_events, + microbatch_lifecycle, +) + + +@pytest.mark.parametrize( + "test_case", + ( + OperationLifecycleCase( + description="completed interval bounds", + expected_event_types=("microbatch_started", "microbatch_completed"), + ), + ), + ids=lambda case: case.description, +) +def test_given_active_resource_when_microbatch_completes_then_bounds_and_progress_are_published( + test_case: OperationLifecycleCase, +) -> None: + dispatcher, events = capture_lifecycle_events() + + with ( + invocation_scope("invocation-1"), + run_scope("run-1"), + resource_attempt_scope(resource_id="model:daily_orders", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + ): + with microbatch_lifecycle() as lifecycle: + lifecycle.completed(affected_rows=12) + + assert tuple(event.event_type for event in events) == test_case.expected_event_types + assert events[0].payload == { + "cursor_start": "2026-01-01T00:00:00", + "cursor_end_exclusive": "2026-01-02T00:00:00", + "planned_cursor_start": "2026-01-01T00:00:00", + "planned_cursor_end_exclusive": "2026-01-04T00:00:00", + "batch_index": 1, + "batch_count": 3, + "configured_batch_size": "1d", + "effective_batch_size": "1d", + "microbatch_strategy": "watermark", + "microbatch_run_type": "normal", + "batch_kind": "ordinary", + } + assert events[1].payload["affected_rows"] == 12 + assert cast(float, events[1].payload["duration_ms"]) >= 0 + assert not is_terminal_event(events[0]) + assert is_terminal_event(events[1]) + + +@pytest.mark.parametrize( + "test_case", + ( + OperationLifecycleCase( + description="interrupted interval bounds", + expected_event_types=("microbatch_started", "microbatch_failed"), + ), + ), + ids=lambda case: case.description, +) +def test_given_interrupted_microbatch_when_scope_exits_then_failed_fact_preserves_bounds( + test_case: OperationLifecycleCase, +) -> None: + dispatcher, events = capture_lifecycle_events() + + with ( + invocation_scope("invocation-1"), + run_scope("run-1"), + resource_attempt_scope(resource_id="model:daily_orders", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + pytest.raises(KeyboardInterrupt), + ): + with microbatch_lifecycle(): + raise KeyboardInterrupt + + assert tuple(event.event_type for event in events) == test_case.expected_event_types + assert events[1].payload["error_type"] == "KeyboardInterrupt" + assert events[1].payload["cursor_start"] == events[0].payload["cursor_start"] + assert events[1].payload["cursor_end_exclusive"] == events[0].payload["cursor_end_exclusive"] + + +@pytest.mark.parametrize( + "test_case", + ( + OperationLifecycleCase( + description="concurrent interval identity", + expected_event_types=("microbatch_started", "microbatch_completed"), + ), + ), + ids=lambda case: case.description, +) +def test_given_concurrent_microbatches_when_publishing_then_each_interval_keeps_resource_identity( + test_case: OperationLifecycleCase, +) -> None: + dispatcher, events = capture_lifecycle_events() + + def execute(batch_index: int) -> None: + with MicrobatchLifecycle( + context=MicrobatchLifecycleContext( + cursor_start=f"{batch_index - 1}", + cursor_end_exclusive=f"{batch_index}", + planned_cursor_start="0", + planned_cursor_end_exclusive="3", + batch_index=batch_index, + batch_count=3, + configured_batch_size="1", + effective_batch_size="1", + microbatch_strategy="watermark", + microbatch_run_type="normal", + batch_kind="ordinary", + ) + ): + pass + + with ( + invocation_scope("invocation-1"), + run_scope("run-1"), + resource_attempt_scope(resource_id="model:order_index", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + ThreadPoolExecutor(max_workers=3) as pool, + ): + futures: list[Future[None]] = [ + cast(Future[None], pool.submit(copy_context().run, execute, batch_index)) + for batch_index in range(1, 4) + ] + for future in futures: + future.result() + + assert len(events) == 6 + assert {event.payload["batch_index"] for event in events} == {1, 2, 3} + assert all(event.resource_id == "model:order_index" for event in events) + assert all(event.resource_attempt_id == "attempt-1" for event in events) + assert Counter((event.payload["batch_index"], event.event_type) for event in events) == Counter( + { + (1, test_case.expected_event_types[0]): 1, + (1, test_case.expected_event_types[1]): 1, + (2, test_case.expected_event_types[0]): 1, + (2, test_case.expected_event_types[1]): 1, + (3, test_case.expected_event_types[0]): 1, + (3, test_case.expected_event_types[1]): 1, + } + ) + + +@pytest.mark.parametrize( + "test_case", + ( + LifecycleErrorCase( + description="batch index beyond plan", + event_type="microbatch_started", + payload={}, + expected_error="batch_index must not exceed", + ), + ), + ids=lambda case: case.description, +) +def test_given_batch_index_beyond_plan_when_starting_then_payload_is_rejected( + test_case: LifecycleErrorCase, +) -> None: + dispatcher, _ = capture_lifecycle_events() + + with ( + invocation_scope("invocation-1"), + run_scope("run-1"), + resource_attempt_scope(resource_id="model:order_index", resource_attempt_id="attempt-1"), + dispatcher_scope(dispatcher), + pytest.raises(ObservabilityValidationError, match=test_case.expected_error), + ): + MicrobatchLifecycle( + context=MicrobatchLifecycleContext( + cursor_start="3", + cursor_end_exclusive="4", + planned_cursor_start="0", + planned_cursor_end_exclusive="3", + batch_index=4, + batch_count=3, + configured_batch_size="1", + effective_batch_size="1", + microbatch_strategy="watermark", + microbatch_run_type="normal", + batch_kind="ordinary", + ) + ).start()