Skip to content

Recompute a ranking from a refresh message (Step 4 of #386) - #496

Merged
zipdoki merged 3 commits into
mainfrom
feat/topk-sweep
Aug 27, 2026
Merged

zipdoki merged 3 commits into
mainfrom
feat/topk-sweep

Conversation

@zipdoki

@zipdoki zipdoki commented Aug 19, 2026

Copy link
Copy Markdown
Contributor

Summary

The previous PR schedules a refresh for the instant a bucket leaves the sliding window, but nothing consumes it. A windowed ranking goes stale the moment its oldest bucket ages out, and stays stale. This consumes the refresh.

POST /aggregations/v1/sweep takes refresh messages and recomputes the ranking row each one names:

{
  "items": [
    {
      "type": "TOPK",
      "item": {
        "database": "graph", "table": "likes", "topk": "daily_likes",
        "source": "...", "target": "...", "direction": "OUT",
        "entity": "...", "topkDimensionValue": "...", "dimensionValues": "",
        "ranges": "", "properties": {}, "refreshAt": 1755500000000
      }
    }
  ]
}
{ "items": [ { "database": "graph", "table": "likes", "topk": "daily_likes", "entity": "...", "status": "OK", "error": null } ] }

SweepItem resolves item polymorphically off type through Jackson's EXTERNAL_PROPERTY, so a second aggregation kind adds a SweepItemPayload subtype and a JsonSubTypes entry rather than a second endpoint. AggregationHandler gains sweep, and AggregationService dispatches by AggregationType instead of fanning out to
every handler the way aggregate does — a refresh message already knows which kind it belongs to.

TopkSweepItem carries exactly what refreshKey encodes, plus refreshAt, so the consumer can rebuild the ranking without reading the queue row back.

Sweep is a write path and is rejected under read-only mode, alongside POST /aggregations/v1/aggregate.

Stacked on #494 — that needs to land first.

Part of #386.

Test plan

  • ./gradlew :engine:test --tests '*TopkAggregationHandlerTest*' — recompute from a refresh message, including the rank row rewrite
  • ./gradlew :engine:test --tests '*AggregationServiceTest*' — dispatch to the handler named by AggregationType
  • ./gradlew :server:test --tests '*MetadataAggControllerE2ETest*' — POST /aggregations/v1/sweep over HTTP, per-item status and error
  • ./gradlew :server:test --tests '*ReadOnlyRequestFilterTest*' — sweep is blocked in read-only mode
  • ./gradlew spotlessKotlinCheck build — formatting and full build

AI Assistance

  • This PR was written largely with AI assistance.
    • Tool / model: Claude Code (Opus 5)

@zipdoki
zipdoki requested a review from em3s as a code owner August 19, 2026 04:49
@zipdoki zipdoki self-assigned this Aug 19, 2026
@dosubot dosubot Bot added size:L This PR changes 100-499 lines, ignoring generated files. enhancement New feature or request labels Aug 19, 2026
@dosubot dosubot Bot added size:XXL This PR changes 1000+ lines, ignoring generated files. and removed size:L This PR changes 100-499 lines, ignoring generated files. labels Aug 19, 2026
@dosubot dosubot Bot added size:L This PR changes 100-499 lines, ignoring generated files. and removed size:XXL This PR changes 1000+ lines, ignoring generated files. labels Aug 20, 2026
Base automatically changed from feat/topk-aggregate to main August 27, 2026 09:34
@dosubot dosubot Bot added size:XL This PR changes 500-999 lines, ignoring generated files. and removed size:L This PR changes 100-499 lines, ignoring generated files. labels Aug 27, 2026
zipdoki and others added 3 commits August 27, 2026 18:40
The sweep design moved from an _expire table scanned by expiredAt to a
refresh queue, and these three payloads never got wired to anything.
AggregationExpireItemRequest has no references at all, and
AggregationExpireResult is only read by AggregationsExpireResponse.from,
which nothing calls.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`AggregationSweepResult` names no item it came from, so a caller pairs a result to its request by position — the same contract `aggregate` has. `sweep` merged handler output as it completed, so a three-item batch came back fully reversed when the first item was the slowest. Same fix as #494 made on the aggregate path, and concurrency is unchanged: `flatMapSequential` subscribes eagerly under the same default `maxConcurrency`.
@dosubot dosubot Bot added size:L This PR changes 100-499 lines, ignoring generated files. and removed size:XL This PR changes 500-999 lines, ignoring generated files. labels Aug 27, 2026
@zipdoki

zipdoki commented Aug 27, 2026

Copy link
Copy Markdown
Contributor Author

Optimistic Merge

@zipdoki
zipdoki merged commit 80369c4 into main Aug 27, 2026
3 checks passed
@zipdoki
zipdoki deleted the feat/topk-sweep branch August 27, 2026 10:04
zipdoki added a commit that referenced this pull request Aug 31, 2026
Steps 3 and 4 landed on main as squash merges (#494, #496) carrying review
changes this branch never saw: `AggregationConstants.Topk.REFRESH_TABLE` is now
`REFRESH_QUEUE`, the aggregate and sweep dispatchers emit in request order via
`flatMapSequential`, and both test classes gained cases for that ordering and for
a window that never slides. Took main's side wherever the two overlapped, since
this branch's copies are the pre-review versions of the same work. Kept the
`AggregationQueryService` bean in `GraphConfiguration`, which is Step 5 and
exists only here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

enhancement New feature or request size:L This PR changes 100-499 lines, ignoring generated files.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant