feat(server): stream paged task results - #3144
Open
contrueCT wants to merge 4 commits into
Open
Conversation
contrueCT
marked this pull request as ready for review
August 8, 2026 04:31
8 tasks
contrueCT
force-pushed
the
task/task-result-streaming-pagination-design
branch
from
August 25, 2026 09:12
300abc6 to
930ab69
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Adds an additive REST read path for streaming persisted task results (optionally as logical JSON pages) without materializing the full decompressed result string/object tree, while keeping the existing task-details endpoint behavior unchanged.
Changes:
- Introduces
GET/HEAD .../tasks/{id}/resultwith optionallimit/pagelogical pagination and signed continuation tokens. - Adds detached result snapshot + metadata SPI hooks on
TaskScheduler, implemented for both standard and distributed schedulers. - Adds streaming lifecycle handling (timeouts, permit limiting, post-commit connection termination) plus extensive unit/integration tests and new server options/defaults.
Reviewed changes
Copilot reviewed 38 out of 38 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| hugegraph-server/hugegraph-test/src/test/java/org/apache/hugegraph/unit/config/ServerOptionsTest.java | Asserts packaged timeout defaults align with request budget. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java | Registers new task-result streaming/pagination test classes. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskResultStreamerTest.java | Unit tests for streaming/paging behavior and limits. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskResultSnapshotTest.java | Unit tests for snapshot stream reopening + fingerprint stability. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/auth/HugeGraphAuthProxyTest.java | Verifies auth proxy delegates metadata-only HEAD path correctly. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/job/TaskResultPageTokenCodecTest.java | Tests page token encoding/decoding, tamper/expiry/rotation rules. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/task/TaskAndResultSchedulerTest.java | Tests distributed scheduler snapshot/metadata read paths. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java | Extends core task tests to cover result metadata SPI. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/TaskApiTest.java | API tests for new /result GET/HEAD, paging, token handling. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/job/TaskResultStreamingOutputTest.java | Unit tests for gzip finalization, deadlines, metrics, and abort behavior. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/job/TaskResultJerseyGrizzlyIntegrationTest.java | Jersey+Grizzly integration tests for gzip completeness and timeouts. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/job/TaskResultGrizzlyIntegrationTest.java | Raw Grizzly integration tests for slow-reader timeout/disconnect handling. |
| hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/job/TaskResultExceptionsTest.java | Verifies endpoint annotations, SPI default compatibility, and error envelope. |
| hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties | Adds new task-result streaming/pagination server configuration defaults. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/LZ4Util.java | Adds LZ4 InputStream decompression helper for streaming reads. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskScheduler.java | Adds default SPI methods for result snapshot + result metadata. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskResultStreamException.java | Adds dedicated exception type/reasons for streaming/paging failures. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskResultStreamer.java | Implements complete streaming + JSON token paging with budgets/deadlines. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskResultSnapshot.java | Defines detached compressed snapshot + fingerprint + streaming access. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskResultPageCursor.java | Cursor model for paging (root type, offset, page size, fingerprint). |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskResultMetadata.java | Defines metadata-only result presence/status DTO. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java | Implements snapshot + metadata SPI for distributed scheduler storage model. |
| hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java | Implements snapshot + metadata SPI for local task vertex results. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/RestServer.java | Initializes token codec + warns when using temporary per-JVM token secret. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java | Adds server options for paging limits, timeouts, active streams, token keys. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java | Adds auth-checked delegation for snapshot/metadata scheduler methods. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultUnavailableException.java | 409 error type for successful tasks with no persisted result. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultTimeoutException.java | 408 error type for request/preflight time budget exhaustion. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultStreamMetrics.java | Metrics for request/stream outcomes, durations, bytes, and failure reasons. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultStreamingOutput.java | StreamingOutput wrapper for deadlines, write timeouts, permit lifecycle, abort. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultPageTokenCodec.java | Implements HMAC-signed, versioned, rotation-aware page token encoding/decoding. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultNotReadyException.java | 409 error type for tasks not yet successful. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultNotReadableException.java | 409 error type for failed/cancelled task result reads. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultNotPageableException.java | 400 error type for scalar pagination attempts. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultException.java | Base WebApplicationException for task-result endpoint error mapping/metrics labels. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskResultChangedException.java | 409 error type when a token no longer matches the current snapshot/fingerprint. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java | Adds /tasks/{id}/result GET/HEAD, paging parameters, permit + lifecycle handling. |
| hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/InvalidTaskResultPageTokenException.java | 400 error type for malformed/expired/tampered/invalid page tokens. |
Suppressed comments (1)
hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java:311
getResult()always callsTaskScheduler.taskResultSnapshot(). If a deployment uses a custom scheduler that hasn’t implemented this new SPI method, the default throwsUnsupportedOperationException, which will currently become HTTP 500. Consider catching this here (after acquiring the permit) and returning a clearer client error (e.g., 501 Not Implemented), so the new endpoint fails gracefully for schedulers without streaming support.
HugeGraph hugeGraph = graph(manager, graphSpace, graph);
HugeConfig graphConfig = (HugeConfig) hugeGraph.configuration();
trace.backend(graphConfig.get(CoreOptions.BACKEND));
TaskScheduler scheduler = hugeGraph.taskScheduler();
TaskResultSnapshot snapshot = scheduler.taskResultSnapshot(
IdGenerator.of(id));
checkRequestDeadline(requestDeadlineNanos);
ensureReadable(snapshot);
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+230
to
+234
| TaskScheduler scheduler = graph(manager, graphSpace, graph) | ||
| .taskScheduler(); | ||
| TaskResultMetadata metadata = scheduler.taskResultMetadata( | ||
| IdGenerator.of(id)); | ||
| ensureReadable(metadata); |
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.
Purpose of the PR
Closes #3148.
This read-path feature builds on #3060 and is complementary to the physical
chunk-storage work tracked by #3071.
Large task results are currently retrieved through the task-details endpoint:
That endpoint materializes a
HugeTask, includes its result inasMap()bydefault, and therefore keeps the historical task-details contract at the cost
of fully materializing the decompressed result. This PR preserves that endpoint
unchanged and adds a dedicated result resource:
The new GET path streams the persisted LZ4 result and optionally returns
logical pages for top-level JSON arrays or objects. It avoids constructing a
complete decompressed
String, result object tree, orHugeTask.asMap()resultbefore writing the response.
Design overview
What this PR changes
This PR adds a dedicated result endpoint that streams LZ4 data and supports
logical JSON pagination. It removes complete decompressed result/object
materialization from the read path, while intentionally retaining the current
single compressed result blob.
Follow-up target (not part of this PR)
Physical chunks plus a manifest and cursor would address the remaining
single-blob allocation, deep-page rescans, and cross-node/restart cursor
stability. That storage-model change remains follow-up work in #3071.
Scope and non-goals
persisted formats, task schemas, or the existing task-details response.
byte[], because thecurrent backend APIs load a blob eagerly. The improvement removes complete
decompressed result/object materialization; it does not claim zero-copy
access to the persisted blob.
page reopens the snapshot and scans from the beginning to its offset. Scan
byte/time and maximum-offset limits bound this work. Physical chunking and
random access remain follow-up work in [Improve] Gremlin Task Large Result Chunking #3071.
require production-like profiling against representative result sizes and
backends.
Main Changes
1. Detached task-result read path
flowchart LR Client[REST client] --> Result["GET .../tasks/{id}/result"] Result --> Auth[HugeGraph authorization] Auth --> Snapshot["TaskScheduler.taskResultSnapshot(id)"] Snapshot --> Local["Standard scheduler<br/>task vertex P.RESULT"] Snapshot --> Distributed["Distributed scheduler<br/>HugeTaskResult vertex"] Local --> Detached[Detached compressed snapshot] Distributed --> Detached Detached --> LZ4[LZ4 input stream] LZ4 --> Mode{Request mode} Mode -->|complete| Raw[Raw JSON stream] Mode -->|limit / page| Parser[Jackson token parser] Parser --> Page[Logical page envelope] Raw --> Gzip[Jersey @Compress / gzip] Page --> Gzip Gzip --> ClientStandardTaskSchedulerreads the result from the local task vertex.TaskAndResultSchedulerreads task metadata and the separateHugeTaskResultvertex used by distributed scheduling. Both return a detachedTaskResultSnapshot, so the HTTP callback retains no backend iterator, vertex,transaction, or scheduler-thread context.
TaskScheduler.taskResultSnapshot()remains a default SPI method. The twobuilt-in schedulers override it; custom schedulers remain source/binary
compatible and report unsupported result streaming if they do not implement
the new capability.
2. Endpoint contract and HugeGraph pagination conventions
GET .../tasks/{id}/resultapplication/json;charset=UTF-8, gzip,Cache-Control: no-store.GET .../tasks/{id}/result?limit=Nlimitfollows existing HugeGraph naming and is bounded by server configuration.GET .../tasks/{id}/result?page=<token>pageis opaque and must be returned unchanged. It already carries the original limit.HEAD .../tasks/{id}/resultThe page envelope uses HugeGraph's existing
pagecontinuation field:{ "root_type": "array", "items": [1, 2], "page": "<opaque token or null>" }limitandpageare mutually exclusive.{ "key": ..., "value": ... }items so duplicate JSONobject keys are not collapsed by materialization.
"page": null; no empty trailing page is emitted.The explicit HEAD method uses
TaskResultMetadataandHugeElement.hasProperty()to determine result presence. It does not call
taskResultSnapshot(), copy thecompressed blob into a detached snapshot, open an LZ4 stream, run pagination
preflight, or consume one of the two default active-stream permits.
3. Page consistency and token protection
Before a paged response commits HTTP 200, the server preflights the persisted
JSON. It validates the root type, scans to the requested offset, enforces scan
budgets, determines whether the page is terminal, and probes page-token
encoding. The streaming pass then reopens the same detached snapshot for output.
When preflight reaches the terminal page, it advances the parser past the
top-level
END_ARRAY/END_OBJECTand requires end-of-input. Trailingwhitespace is accepted; a second root value or trailing garbage is rejected as
invalid persisted JSON before HTTP 200 is committed. The streaming pass repeats
the EOF check defensively.
Continuation tokens have the form
key-id.base64url(payload).base64url(hmac). HMAC-SHA256 with constant-time MACcomparison binds:
Malformed, expired, tampered, wrong-route, wrong-task, or oversized tokens are
rejected. If the persisted result or root type changes between page requests,
the server returns HTTP 409 instead of continuing the old offset against new
data. Current and previous key ids/secrets provide a bounded rotation window.
4. End-to-end response lifecycle and timeout ownership
The result stream owns one lifecycle boundary:
sequenceDiagram participant API as TaskAPI participant Out as TaskResultStreamingOutput participant Gzip as Jersey GZIPOutputStream participant Net as Grizzly connection API->>API: start request-level deadline API->>API: acquire permit and load snapshot API->>API: optional preflight (stage deadline capped by request deadline) API->>Out: write response entity Out->>Net: apply remaining write timeout Out->>Gzip: stream uncompressed result Out->>Gzip: finish gzip trailer and flush alt success Out->>Out: record success else post-commit failure or timeout Out->>Out: record post-commit failure Out->>Net: terminate connection explicitly end Out->>Net: restore prior timeout Out->>API: release permit in finallySuccess is recorded only after
GZIPOutputStream.finish()and the final flushsucceed. Gzip trailer/flush failures are therefore included in task-result
metrics, and the configured connection write timeout and active-stream permit
remain owned through final gzip output. A failure after response commit
explicitly terminates the Grizzly connection so a truncated HTTP 200 response
cannot end as a normal successful response.
The global
restserver.request_timeoutis the authoritative total deadline.Snapshot work, pagination preflight, decompression, gzip output, and blocked
writes share its remaining budget. Scan and stream stage limits are retained but
are capped by the request deadline. Both parser/decompression loops and output
boundaries check
Thread.currentThread().isInterrupted()so Grizzly transactiontimeout interrupts are classified consistently as task-result timeouts.
Packaged defaults now fit the global 30-second request budget:
restserver.task_result_page_size_maxrestserver.task_result_page_offset_maxrestserver.task_result_scan_uncompressed_bytes_maxrestserver.task_result_scan_time_maxrestserver.task_result_stream_time_maxrestserver.task_result_active_streams_maxrestserver.task_result_page_token_ttlrestserver.task_result_page_token_length_maxPre-commit request/preflight timeout returns HTTP 408. Once a response is
committed, its status cannot be replaced; the connection is terminated and the
timeout is recorded as a post-commit outcome.
5. Status and error behavior
409 Conflict409 Conflict; read the original error from task details409 Conflict400 Bad Request409 Conflict413 Request Entity Too Large408 Request Timeout503 Service UnavailablePre-commit task-result failures use HugeGraph's standard
exception,message,and
causeenvelope. Internal reason labels stay metrics-only. Metrics separatecomplete/page mode, active streams, success, timeout, disconnect, and
pre-/post-commit failure; logs include backend, task id, offset, byte counts,
stage, and error type without logging result content or complete tokens.
For multi-node deployments, every serving node must use the same Base64URL
token secret (at least 32 bytes) and key id. The generated per-JVM default is
convenient for a single process but cannot support continuation on another node
or after restart.
Review focus and rollout
scheduler implementations.
token bindings.
timeout restoration, and permit release under slow-reader behavior.
proxy that buffers the full response removes the client-visible streaming
benefit.
The endpoint is additive and requires no migration. Rollback is a code rollback;
persisted task results and the existing task-details endpoint remain compatible.
Clients must stop using
/tasks/{id}/resultbefore rolling back to a versionthat does not expose it.
Verifying these changes
mvn editorconfig:formatmvn clean compile '-Dmaven.javadoc.skip=true'upstream/masterat401e627c3.mvn test -pl hugegraph-server/hugegraph-test -am -rf :hugegraph-core -P unit-test '-Dtest=TaskResultStreamerTest,TaskResultStreamingOutputTest,TaskResultExceptionsTest,TaskResultJerseyGrizzlyIntegrationTest,ServerOptionsTest' -DfailIfNoTests=falseServerOptionsTesttests).mvn test -pl hugegraph-server/hugegraph-test -am -rf :hugegraph-core -P core-test,memory '-Dtest=TaskCoreTest#testTaskWithoutResult,TaskAndResultSchedulerTest' -DfailIfNoTests=falsegit diff --check upstream/master...HEADThe new real Jersey + Grizzly +
@Compresssuite verifies:timeout; and
Additional tests cover explicit HEAD dispatch and metadata-only scheduler access,
gzip trailer failure without a false success metric, request-deadline capping,
interruption propagation, 408 mapping, terminal-page EOF/whitespace/garbage
behavior, packaged timeout defaults, and local versus distributed
result-presence reads.
TaskApiTestexercises the real public HEAD contract inthe API profile; the Linux CI package/start harness is the final end-to-end
API-profile validation.
Does this PR potentially affect the following parts?
stream metrics, and multi-node token-key deployment)
Documentation Status
Doc - TODODoc - DoneDoc - No Need