Conversation
New table + DAL for a per-source watermark. Nothing calls them yet; wiring follows in a separate PR. Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>
Reads/commits the per-source watermark around the run. lf-criticality-score keeps its full-fetch behavior unchanged. Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>
Contributor
There was a problem hiding this comment.
Copilot review overview
🟢 Approval recommended
Watermarks advance only after complete successful processing, and the persistence semantics have focused coverage.
Review effort: Balanced
Findings: None
What changed in this PR
Adds persistent discovery watermarks for incremental insights-discussions processing while preserving full-fetch behavior for LF Criticality Score.
Changes:
- Adds watermark persistence and forward-only DAL operations.
- Propagates
sincethrough Temporal activities and source descriptors. - Commits watermarks only after successful, non-truncated processing.
| File | Description |
|---|---|
services/libs/data-access-layer/src/index.ts |
Exports discovery DAL APIs. |
services/libs/data-access-layer/src/discovery/types.ts |
Defines persisted source state. |
services/libs/data-access-layer/src/discovery/sourceState.ts |
Implements watermark reads and upserts. |
services/libs/data-access-layer/src/discovery/sourceState.test.ts |
Tests watermark persistence semantics. |
services/libs/data-access-layer/src/discovery/index.ts |
Exports discovery modules. |
services/apps/automatic_projects_discovery_worker/src/workflows/discoverProjects.ts |
Coordinates watermark lifecycle. |
services/apps/automatic_projects_discovery_worker/src/sources/types.ts |
Generalizes source filtering with since. |
services/apps/automatic_projects_discovery_worker/src/sources/lf-criticality-score/source.ts |
Maps since to scoredAfter. |
services/apps/automatic_projects_discovery_worker/src/sources/insights-discussions/source.ts |
Filters unchanged discussions. |
services/apps/automatic_projects_discovery_worker/src/activities/activities.ts |
Adds watermark activities and truncation tracking. |
backend/src/database/migrations/V1789994431__discovery-source-state.sql |
Creates source watermark storage. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
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
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
Second of two PRs for CM-1455 (watermark for discovery sources). Depends on the PR1
migration/DAL being merged first (
discoverySourceStatetable + DAL functions).This PR wires the watermark into the workflow, but only for
insights-discussions.lf-criticality-score's interface is generalized (scoredAfter→since) but itstill does a full fetch every run — no behavior change for that source in this PR.
Changes
sources/types.ts:IDiscoverySource.listAvailableDatasetstakes{ since? }instead of nothing;
IDatasetDescriptorgains an optionalsince, stamped bylistAvailableDatasetsso it survives intofetchDatasetStreamvia workflowhistory (they're separate Temporal activities).
sources/insights-discussions/source.ts:fetchAllDiscussionRepoUrlsnow skipsdiscussions with
updatedAtolder thansince, client-side, after walking everypage (no early-stop — pagination isn't ordered by
updatedAt, so an early-stopwould risk permanently missing edited discussions). Adds a
skippedUnchangedcounter to the aggregate log.
sources/lf-criticality-score/source.ts: renamedscoredAfterparam tosinceto match the generic interface; maps
since→scoredAfterinternally. Behaviorunchanged — the workflow doesn't pass it a watermark in this PR.
activities/activities.ts:listDatasetsnow forwardssinceto the source (previously called with noarguments — dead code).
readSourceWatermark(sourceName): reads the persisted watermark, subtractsa 24h overlap, and captures
now()— all inside the activity, since workflowcode can't do I/O or read the clock directly.
commitSourceWatermark(sourceName, watermark, force).IProcessDatasetResultgainstruncated: boolean, set at all three early-exitpaths driven by
DISCOVERY_NEW_PROJECTS_LIMIT(biased towardtrue: a falsenegative would advance the watermark past unprocessed data).
workflows/discoverProjects.ts: forinsights-discussionsonly, reads thewatermark before listing datasets, wraps
listDatasetsin its own try/catch (asource-list failure no longer kills the whole run), and commits the new watermark
only if the source's entire run succeeded and wasn't truncated.
mode: 'full'skips the watermark read and forces the commit (bypasses the forward-only
GREATESTviaforce: true).Why a 24h overlap
Up to 90 minutes can elapse between
listDatasetsandprocessDataset(
startToCloseTimeout), plus retries, plus ordinary clock skew. 24h absorbs thatgap with a large margin. Re-processing already-seen discussions is harmless (the
catalog dedupes on
repoUrl), so a generous overlap costs nothing.Known first-run behavior
DISCOVERY_NEW_PROJECTS_LIMITdefaults to 20; the category currently has ~100candidates. The first several nightly runs will be truncated and won't commit a
watermark yet — expected, and the backlog drains in a few days once new candidates
per night drop below the limit.
Out of scope
Exact watermarking from data (
max(updatedAt)) — the committed value iscapturedAt, a timestamp taken inside the activity, not derived from the data.Deemed acceptable given the 24h overlap.
Type of change
JIRA ticket
https://linuxfoundation.atlassian.net/browse/CM-1455
Depends on: PR1 (
discoverySourceStatetable + DAL)