epic-2/story-3: Batch, Deduplicate, and Flush Tracking Events - #31
Closed
usmanabbas7 wants to merge 13 commits into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements conversion tracking, batching, and queue flushing for the Convert Python SDK. It introduces a thread-safe in-process TrackingQueue, a Tracker orchestrator, and runtime integration lifecycle hooks (explicit, periodic, and atexit flushes). The feedback focuses on several key improvements: preventing invalid tracking POST requests to /track/None when the SDK key is missing in offline mode, allowing the flush methods and background/shutdown hooks to accept and forward the correct ReleaseReason (such as TIMEOUT or ATEXIT), and validating that the configured data_store implements the DataStore protocol during configuration initialization to fail fast.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+191
to
+192
| try: | ||
| transport.send_tracking(payload, sdk_key=str(self._config.sdk_key)) |
There was a problem hiding this comment.
When the SDK is initialized with direct config (offline mode), self._config.sdk_key is None. Calling str(self._config.sdk_key) converts None to the string "None", which causes the transport to make invalid tracking POST requests to /track/None. We should check if sdk_key is present, and if not, restore the items to the queue and raise a TrackingDeliveryError.
Suggested change
| try: | |
| transport.send_tracking(payload, sdk_key=str(self._config.sdk_key)) | |
| sdk_key = self._config.sdk_key | |
| if not sdk_key: | |
| self._queue.restore(items) | |
| from convert_sdk.errors import TrackingDeliveryError | |
| raise TrackingDeliveryError("Tracking delivery failed: SDK key is missing.") | |
| try: | |
| transport.send_tracking(payload, sdk_key=sdk_key) |
Comment on lines
+167
to
+174
| def flush(self) -> None: | ||
| """Explicitly release the queue (``ReleaseReason.EXPLICIT``). | ||
|
|
||
| Drains, serializes via the Story 2.2 builder, and delivers through the | ||
| transport, clearing the queue on success. An empty queue is a safe | ||
| no-op (no transport call, no error). | ||
| """ | ||
| self._release(ReleaseReason.EXPLICIT) |
There was a problem hiding this comment.
The Tracker.flush method currently does not accept any arguments and always releases the queue with ReleaseReason.EXPLICIT. This means that ReleaseReason.TIMEOUT and ReleaseReason.ATEXIT are never actually used when periodic or shutdown flushes occur. We should allow flush to accept an optional reason parameter.
Suggested change
| def flush(self) -> None: | |
| """Explicitly release the queue (``ReleaseReason.EXPLICIT``). | |
| Drains, serializes via the Story 2.2 builder, and delivers through the | |
| transport, clearing the queue on success. An empty queue is a safe | |
| no-op (no transport call, no error). | |
| """ | |
| self._release(ReleaseReason.EXPLICIT) | |
| def flush(self, reason: ReleaseReason = ReleaseReason.EXPLICIT) -> None: | |
| """Explicitly release the queue (``ReleaseReason.EXPLICIT``). | |
| Drains, serializes via the Story 2.2 builder, and delivers through the | |
| transport, clearing the queue on success. An empty queue is a safe | |
| no-op (no transport call, no error). | |
| """ | |
| self._release(reason) |
Comment on lines
+108
to
+121
| def flush(self) -> None: | ||
| """Explicitly release the tracking queue and deliver queued events. | ||
|
|
||
| Drains the shared queue through the single release path, serializes the | ||
| batched per-visitor events via the Story 2.2 serializer, and delivers | ||
| through the configured transport, clearing the queue on success. A flush | ||
| on an empty queue is a safe no-op (no transport call, no error). This is | ||
| the canonical, deterministic control point for queue release (FR39); the | ||
| default lifecycle is explicit-flush-only. | ||
|
|
||
| Safe to call before :meth:`initialize` (no tracker yet) — it is a no-op. | ||
| """ | ||
| if self._tracker is not None: | ||
| self._tracker.flush() |
There was a problem hiding this comment.
To support forwarding the correct ReleaseReason (such as TIMEOUT or ATEXIT) from the periodic timer or interpreter shutdown hooks, Core.flush should accept an optional reason parameter and forward it to the underlying tracker.
Suggested change
| def flush(self) -> None: | |
| """Explicitly release the tracking queue and deliver queued events. | |
| Drains the shared queue through the single release path, serializes the | |
| batched per-visitor events via the Story 2.2 serializer, and delivers | |
| through the configured transport, clearing the queue on success. A flush | |
| on an empty queue is a safe no-op (no transport call, no error). This is | |
| the canonical, deterministic control point for queue release (FR39); the | |
| default lifecycle is explicit-flush-only. | |
| Safe to call before :meth:`initialize` (no tracker yet) — it is a no-op. | |
| """ | |
| if self._tracker is not None: | |
| self._tracker.flush() | |
| def flush(self, reason: Optional[Any] = None) -> None: | |
| """Explicitly release the tracking queue and deliver queued events. | |
| Drains the shared queue through the single release path, serializes the | |
| batched per-visitor events via the Story 2.2 serializer, and delivers | |
| through the configured transport, clearing the queue on success. A flush | |
| on an empty queue is a safe no-op (no transport call, no error). This is | |
| the canonical, deterministic control point for queue release (FR39); the | |
| default lifecycle is explicit-flush-only. | |
| Safe to call before :meth:`initialize` (no tracker yet) — it is a no-op. | |
| """ | |
| if self._tracker is not None: | |
| if reason is not None: | |
| self._tracker.flush(reason) | |
| else: | |
| self._tracker.flush() |
Comment on lines
+60
to
+81
| def setup_periodic_flush( | ||
| flushable: Flushable, | ||
| interval_ms: Optional[int], | ||
| ) -> Optional["PeriodicFlusher"]: | ||
| """Opt into periodic background flushes via a daemonic ``threading.Timer``. | ||
|
|
||
| When ``interval_ms`` is ``None`` (the default lifecycle), periodic flush is | ||
| disabled and this returns ``None`` — the queue is released only on explicit | ||
| :func:`flush` (or batch-size release). When a positive interval is given, | ||
| starts a daemonic, self-rescheduling timer that calls the flushable's | ||
| ``flush()`` every ``interval_ms`` milliseconds and returns a | ||
| :class:`PeriodicFlusher` the caller can :meth:`PeriodicFlusher.cancel`. | ||
|
|
||
| The timer thread is daemonic so it never blocks interpreter shutdown; if the | ||
| process exits before the timer fires the flush is silently skipped (the | ||
| documented, correct behavior for short-lived runtimes). | ||
| """ | ||
| if interval_ms is None: | ||
| return None | ||
| flusher = PeriodicFlusher(flushable.flush, interval_ms) | ||
| flusher.start() | ||
| return flusher |
There was a problem hiding this comment.
Update setup_periodic_flush to pass ReleaseReason.TIMEOUT when flushing, falling back to zero-arg flush() if the flushable does not support arguments.
def setup_periodic_flush(
flushable: Flushable,
interval_ms: Optional[int],
) -> Optional["PeriodicFlusher"]:
"""Opt into periodic background flushes via a daemonic ``threading.Timer``.
When ``interval_ms`` is ``None`` (the default lifecycle), periodic flush is
disabled and this returns ``None`` — the queue is released only on explicit
:func:`flush` (or batch-size release). When a positive interval is given,
starts a daemonic, self-rescheduling timer that calls the flushable's
``flush()`` every ``interval_ms`` milliseconds and returns a
:class:`PeriodicFlusher` the caller can :meth:`PeriodicFlusher.cancel`.
The timer thread is daemonic so it never blocks interpreter shutdown; if the
process exits before the timer fires the flush is silently skipped (the
documented, correct behavior for short-lived runtimes).
"""
if interval_ms is None:
return None
from convert_sdk.tracking.queue import ReleaseReason
def _periodic_flush() -> None:
try:
flushable.flush(ReleaseReason.TIMEOUT) # type: ignore[call-arg]
except TypeError:
flushable.flush()
flusher = PeriodicFlusher(_periodic_flush, interval_ms)
flusher.start()
return flusher
Comment on lines
+84
to
+102
| def register_atexit_flush(flushable: Flushable) -> Callable[[], None]: | ||
| """Register a best-effort final flush on interpreter shutdown (``atexit``). | ||
|
|
||
| Returns the registered callback so a caller can ``atexit.unregister`` it. | ||
| Best-effort only: ``atexit`` does NOT fire under ``SIGKILL`` or in some | ||
| serverless runtimes, and any exception during the final flush is swallowed | ||
| so shutdown never crashes (a never-flushed process must exit cleanly even if | ||
| it silently drops events — NFR18). | ||
| """ | ||
|
|
||
| def _final_flush() -> None: | ||
| try: | ||
| flushable.flush() | ||
| except Exception: # pragma: no cover - shutdown best-effort | ||
| # Never let a shutdown-time delivery failure crash the interpreter. | ||
| pass | ||
|
|
||
| atexit.register(_final_flush) | ||
| return _final_flush |
There was a problem hiding this comment.
Update register_atexit_flush to pass ReleaseReason.ATEXIT when flushing, falling back to zero-arg flush() if the flushable does not support arguments.
def register_atexit_flush(flushable: Flushable) -> Callable[[], None]:
"""Register a best-effort final flush on interpreter shutdown (``atexit``).
Returns the registered callback so a caller can ``atexit.unregister`` it.
Best-effort only: ``atexit`` does NOT fire under ``SIGKILL`` or in some
serverless runtimes, and any exception during the final flush is swallowed
so shutdown never crashes (a never-flushed process must exit cleanly even if
it silently drops events — NFR18).
"""
from convert_sdk.tracking.queue import ReleaseReason
def _final_flush() -> None:
try:
try:
flushable.flush(ReleaseReason.ATEXIT) # type: ignore[call-arg]
except TypeError:
flushable.flush()
except Exception: # pragma: no cover - shutdown best-effort
# Never let a shutdown-time delivery failure crash the interpreter.
pass
atexit.register(_final_flush)
return _final_flush
Comment on lines
+140
to
+147
| if self.auto_flush_interval_ms is not None and ( | ||
| not isinstance(self.auto_flush_interval_ms, int) | ||
| or self.auto_flush_interval_ms < 1 | ||
| ): | ||
| raise InvalidConfigError( | ||
| "SDKConfig 'auto_flush_interval_ms' must be a positive integer " | ||
| f"or None; got {self.auto_flush_interval_ms!r}" | ||
| ) |
There was a problem hiding this comment.
Validate that the configured data_store implements the DataStore protocol during initialization to fail fast with InvalidConfigError rather than failing with an attribute error later during tracking.
Suggested change
| if self.auto_flush_interval_ms is not None and ( | |
| not isinstance(self.auto_flush_interval_ms, int) | |
| or self.auto_flush_interval_ms < 1 | |
| ): | |
| raise InvalidConfigError( | |
| "SDKConfig 'auto_flush_interval_ms' must be a positive integer " | |
| f"or None; got {self.auto_flush_interval_ms!r}" | |
| ) | |
| if self.auto_flush_interval_ms is not None and ( | |
| not isinstance(self.auto_flush_interval_ms, int) | |
| or self.auto_flush_interval_ms < 1 | |
| ): | |
| raise InvalidConfigError( | |
| "SDKConfig 'auto_flush_interval_ms' must be a positive integer " | |
| f"or None; got {self.auto_flush_interval_ms!r}" | |
| ) | |
| if self.data_store is not None: | |
| from convert_sdk.ports.storage import DataStore | |
| if not isinstance(self.data_store, DataStore): | |
| raise InvalidConfigError( | |
| "SDKConfig 'data_store' must implement the DataStore protocol; " | |
| f"got {self.data_store!r}" | |
| ) |
This was referenced Jun 8, 2026
Beads: ai-driven-product-dev-rvx5 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
- SDKConfig: batch_size=10 (JS parity), auto_flush_interval_ms=None, data_store
- ports/storage.py: DataStore protocol + InMemoryDataStore + resolve_data_store
- ports/transport.py + httpx adapter: send_tracking POST /track/{sdkKey}
- errors.py: TrackingDeliveryError; redaction covers /track/ keys
Beads: ai-driven-product-dev-rvx5
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Beads: ai-driven-product-dev-23a9 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
tracking/queue.py: TrackingQueue (per-visitor grouping, threading.Lock, batch-size release signal, drain/restore) + ReleaseReason enum (F-031). Beads: ai-driven-product-dev-23a9 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Beads: ai-driven-product-dev-ahes Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
tracking/deduplication.py: evaluate_dedup + DedupDecision + goal_marker_key via DataStore boundary. F-006 (transaction-only re-send) + F-007 (uncond marker). Beads: ai-driven-product-dev-ahes Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Beads: ai-driven-product-dev-pyen Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
…REEN) - tracking/tracker.py: Tracker orchestrates dedup->enqueue->shared release; flush serializes via Story 2.2 payloads.py, delivers via Transport, clears queue; failed delivery restores queue (no retry). F-006 wiring. - tracking/flush.py: frozen flush()/setup_periodic_flush()/register_atexit_flush() + daemonic PeriodicFlusher (F-058/F-059); SIGTERM pattern as comment only. - ConversionStatus.DEDUPLICATED + ConversionResult.tracked/reason properties (PRD contract without changing Story 2.1 fields). - Context.track_conversion gains force_multiple, routes through shared tracker. - Core owns shared Tracker, adds flush(), wires opt-in periodic flusher, cancels on close. Beads: ai-driven-product-dev-pyen Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
- tests/test_flush_lifecycle.py(NEW): daemonic timer fires, thread.daemon True, errors swallowed, cancel stops, atexit best-effort. - docs/runtime-integration.md(NEW): per-runtime flush matrix (Lambda, Cloud Run, gunicorn, uvicorn, Celery, CLI) + SIGTERM pattern. - README.md: Conversion tracking + Runtime Integration sections; public API list. Beads: ai-driven-product-dev-ep61 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
…rix (GREEN)
- tests/integration/conftest.py(NEW): qs-06 frozen RESPX harness
(mock_config_endpoint, mock_tracking_endpoint, sdk_with_mock_transport,
in_memory_store); tests/fixtures/config/{minimal,full}_config.json(NEW).
- tests/integration/test_tracking_delivery.py(NEW): flush->serialize->POST,
queue clear, batch-size release, multi-visitor batch, failed-delivery retry.
- tests/integration/test_queue_lifecycle.py(NEW): qs-07 scenario matrix
(explicit, timer, atexit, no-flush, SIGTERM via subprocess) + NFR21 parity.
- fix tracker._build_batch_payload: merge events per visitor into one
visitors[] entry (JS VisitorsQueue grouping) — surfaced by TEST-1.
Beads: ai-driven-product-dev-b1jw
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
CLAUDE.md was an untracked working-tree file pulled in by git add -A during [PL-1]; it is not part of this feature and was never on the base branch. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Code review: tracker imported the private _build_goal_data from payloads.py. Replace with a public event_has_goal_data() predicate so the tracker decides the F-006 transaction-send branch without reaching into a serializer internal. No behavior change; full suite 340 green. Beads: ai-driven-product-dev-45m2 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
…mmit Keeps the pre-existing empty working-tree CLAUDE.md untracked, matching the base branch and sprint convention. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
usmanabbas7
force-pushed
the
epic-2/story-2-support-revenue-data-and-tracking-payload-construction
branch
from
June 14, 2026 16:58
021044d to
7cf6b2f
Compare
usmanabbas7
force-pushed
the
epic-2/story-3-batch-deduplicate-and-flush-tracking-events
branch
from
June 14, 2026 16:58
fb7d7f5 to
627d3ca
Compare
Collaborator
|
Superseded — all commits already in main (bc76b64). Closing without merge as part of post-sprint cleanup. |
abbaseya
deleted the
epic-2/story-3-batch-deduplicate-and-flush-tracking-events
branch
June 18, 2026 16:31
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds the tracking queue + dedup + flush/delivery layer on top of Story 2.2's serializer. Stops before lifecycle-event emission and delivery-outcome reporting (Story 2.4).
SDKConfiggainsauto_flush_interval_ms/batch_size/data_store; minimalports/storage.pyDataStoreprotocol + in-memory default (forward-compatible with Story 3.1);Transportport extended with the tracking-POST method + httpx adapter (/track/{sdkKey}, TLS-only)tracking/queue.py: thread-safe per-visitor queue, batch-size release, typedReleaseReasonenum (size/explicit/timeout/atexit) (F-031)tracking/deduplication.py:(visitor_id, goal_id)dedup via the DataStore boundary; truth-table exact; marker persisted unconditionally incl.force_multiple=Truecalls (F-007)tracking/tracker.pyorchestration seam;force_multipleactivated onContext.track_conversion; JS-parity nuance: forced repeat sends only the transaction (goalData) event, never re-fires the bare conversion (F-006);Core.flush()drains via the single shared release path through Story 2.2'spayloads.py(no second serializer); empty-queue flush is a no-optracking/flush.pywith qs-07 frozen names (flush(),setup_periodic_flush(),register_atexit_flush()); daemonicthreading.Timerwith explicittimer.daemon = Trueafter construction (F-058/F-059); SIGTERM pattern documented as comment only;docs/runtime-integration.mdper-runtime guide + README sectiontests/integration/conftest.py), delivery tests, qs-07 queue-lifecycle matrix (explicit/timer/atexit/no-flush/SIGTERM-pattern)Audit findings honored: F-006, F-007, F-031, F-058, F-059.
Tests: 268 → 340 (+72), full suite green (
uv run pytest).Traceability
sprint/2026-04-06-convert-python-sdk— stacked on epic-2/story-2: Support Revenue Data and Tracking Payload Construction #30 (story 2-2) → epic-2/story-1: Track Conversions from a Visitor Context #29 → epic-1/story-3: Create and Reuse Visitor Contexts (gap-fill) #28 → epic-1/story-6: Deliver Quickstart and First-Run Examples #27 → epic-1/story-4: Run Local Experience Evaluations #26 → epic-1/story-2: Support sdkKey and Direct-Config Initialization #25 → Epic 1 Story 1: Scaffold the publishable SDK foundation #24ai-driven-product-dev-45m2; tasks-rvx5,-23a9,-ahes,-pyen,-ep61,-b1jw— all closedai-driven-product-dev/work/2026-06-07-batch-deduplicate-and-flush-tracking-events/Notes for reviewer
batch_size=10default; Core-owned shared tracker; test-file locations. See conductor's readiness-assessment.md._build_goal_data; fixed via publicevent_has_goal_data()predicate).ConversionResultgainedDEDUPLICATEDstatus + derivedtracked/reasonproperties — Story 2.1 field contract unchanged.CLAUDE.mdaccidentally re-added by the review R1 commit.🤖 Generated with Claude Code