refactor: move coordinator transport behind runtime ports#2235
Merged
Conversation
📝 WalkthroughWalkthroughThis PR introduces an internal dispatch type abstraction layer to decouple internal task/operation handling from coordinator protobuf types. It defines ChangesDispatch Type Abstraction & Conversion
Runtime Abstractions & Coordinator Integration
Worker Reporting Reorganization
Executor & Task Building
Agent & Service Wiring
Worker Execution Integration
Persistence & Storage
Test Infrastructure
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/service/scheduler/dag_executor.go (1)
151-180:⚠️ Potential issue | 🟠 Major | ⚡ Quick winReject unsupported distributed operations before dispatch.
The distributed branch dispatches whatever
operationit receives, soDispatchOperationUnspecifiedor any future unknown value gets sent to the coordinator instead of failing fast. The local branch already rejects those cases, so this creates inconsistent behavior and can persist invalid dispatch requests downstream.Proposed fix
func (e *DAGExecutor) ExecuteDAG( ctx context.Context, dag *core.DAG, operation exec.DispatchOperation, runID string, previousStatus *exec.DAGRunStatus, triggerType core.TriggerType, scheduleTime string, ) error { if e.shouldUseDistributedExecution(dag) { + switch operation { + case exec.DispatchOperationStart, exec.DispatchOperationRetry: + case exec.DispatchOperationUnspecified: + return fmt.Errorf("operation not specified") + default: + return fmt.Errorf("unsupported operation: %v", operation) + } + // Distributed execution: dispatch to coordinator taskOpts := []executor.TaskOption{ executor.WithWorkerSelector(dag.WorkerSelector), executor.WithPreviousStatus(previousStatus), executor.WithBaseConfig(executor.ResolveBaseConfig(dag.BaseConfigData, e.baseConfigPath)),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/service/scheduler/dag_executor.go` around lines 151 - 180, The distributed-execution branch in shouldUseDistributedExecution(...) currently forwards any operation to executor.CreateTask and dispatchToCoordinator(...) including DispatchOperationUnspecified (and future unknown values); add the same validation used by the local branch to reject DispatchOperationUnspecified (and treat unknown/invalid enum values as errors) before building the task options, returning an error if the operation is invalid so invalid operations are rejected consistently and not dispatched to dispatchToCoordinator.
🧹 Nitpick comments (5)
internal/runtime/agent/agent.go (2)
969-973: ⚡ Quick winUpdate log message to match new variable name.
The cleanup logic now operates on a
dispatcherinstead of a coordinator client, but the log message still refers to "coordinator client". Update the message for consistency:if dispatcher != nil { if err := dispatcher.Cleanup(ctx); err != nil { - logger.Warn(ctx, "Failed to cleanup coordinator client", tag.Error(err)) + logger.Warn(ctx, "Failed to cleanup dispatcher", tag.Error(err)) } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/runtime/agent/agent.go` around lines 969 - 973, The log message refers to "coordinator client" but the code now calls dispatcher.Cleanup; update the logger.Warn invocation to reference the correct variable (dispatcher) so it reads something like "Failed to cleanup dispatcher" (keep the same context and include tag.Error(err)); modify the logger.Warn call that wraps dispatcher.Cleanup to use the new message.
1561-1570: 💤 Low valueConsider clarifying debug log condition in createDispatcher.
The debug message at line 1564 is only logged when
a.registry != nil, which appears intentional (log only when distributed execution might be expected). However, the message "Dispatcher factory is not configured" could be clearer that this is expected behavior in local-only execution scenarios.Consider either:
- Adjusting the log message: "Dispatcher factory not configured; running in local-only mode"
- Adding a comment explaining why logging is conditional on registry presence
This would help developers understand when a nil dispatcher is expected vs. a configuration issue.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/runtime/agent/agent.go` around lines 1561 - 1570, In Agent.createDispatcher, clarify that a nil dispatcher can be expected in local-only execution by either updating the debug log in the conditional that currently checks a.registry to read something like "Dispatcher factory not configured; running in local-only mode" or by adding a one-line comment above the logger.Debug call explaining that logging is conditional on a.registry to avoid noisy logs in non-distributed setups; reference the function name Agent.createDispatcher and the logger.Debug call so you update the message or add the explanatory comment in that exact location.internal/cmd/context.go (1)
589-593: 💤 Low valueUnused context parameter in factory closure.
The
context.Contextparameter at line 590 is declared but never used when creating the dispatcher. If this parameter is required for interface compatibility withDispatcherFactory, consider adding a comment explaining why it's unused. Otherwise, you can use_to make the intent explicit:func (c *Context) RuntimeDispatcherFactory() func(context.Context) runtime.Dispatcher { - return func(context.Context) runtime.Dispatcher { + return func(_ context.Context) runtime.Dispatcher { return coordinator.NewRuntimeDispatcher(c.ServiceRegistry, c.Config.Core.Peer) } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/cmd/context.go` around lines 589 - 593, The closure returned by Context.RuntimeDispatcherFactory declares a context.Context parameter that is unused; update the closure to either use the blank identifier (func(_ context.Context) runtime.Dispatcher) or add a brief comment above or inside the closure explaining the parameter is intentionally unused for interface compatibility with DispatcherFactory, keeping the body that returns coordinator.NewRuntimeDispatcher(c.ServiceRegistry, c.Config.Core.Peer) unchanged and referencing the RuntimeDispatcherFactory method and coordinator.NewRuntimeDispatcher to locate the code.internal/engine/engine.go (1)
205-209: ⚡ Quick winDocument that the factory may return nil.
The factory function calls
coordinator.NewRuntimeDispatcher, which returnsnilwhenserviceRegistryisnil. Callers of this factory should be aware that they may receive a nil dispatcher. Consider adding documentation or an explicit nil check with an error return if a nil dispatcher is not acceptable in this context.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/engine/engine.go` around lines 205 - 209, The runtimeDispatcherFactory currently returns a func(context.Context) runtime.Dispatcher that can yield nil because coordinator.NewRuntimeDispatcher may return nil when e.serviceRegistry is nil; update the factory to perform an explicit nil check and surface that case instead of silently returning nil: change runtimeDispatcherFactory to return func(context.Context) (runtime.Dispatcher, error) (or otherwise return an error/panic) and inside the returned closure call coordinator.NewRuntimeDispatcher(e.serviceRegistry, e.cfg.Core.Peer), check the result for nil and return a clear error if nil (and update all callers of runtimeDispatcherFactory to handle the error), or alternatively add a short doc comment on runtimeDispatcherFactory that it may return nil if e.serviceRegistry is nil if you prefer to keep the current signature.internal/service/coordinator/client.go (1)
960-1000: 💤 Low valueClean conversion at adapter boundary.
The implementation correctly converts coordinator proto responses to internal types. The nil check at line 987 is defensive—if
attemptCallsucceeds,respshould be set. Returningnil, nilwhenresp == nildespiteerr == nilhandles a protocol error gracefully, though an explicit error might be clearer.Optional: Return explicit error for nil response
if err != nil || resp == nil { - return nil, err + if err != nil { + return nil, err + } + return nil, fmt.Errorf("coordinator returned nil response") }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/service/coordinator/client.go` around lines 960 - 1000, GetDAGRunStatus currently returns (nil, nil) if attemptCall returns no error but resp is still nil; update the function (GetDAGRunStatus) to treat a nil resp as an explicit protocol/error case by returning a descriptive error instead of (nil, nil). Modify the post-attemptCall check that currently does "if err != nil || resp == nil { return nil, err }" to return a clear fmt.Errorf (e.g., "nil response from coordinator" or similar) when resp == nil while preserving existing error propagation from attemptCall.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/proto/convert/dispatch.go`:
- Around line 20-22: The owner coordinator port validation is incomplete: ensure
you reject impossible port values (e.g. -1, >65535) in both conversion
directions by adding explicit range checks around task.Owner.Port and any
reciprocal conversion helpers in internal/proto/convert/dispatch.go; update the
outbound check (currently guarding int32 cast) to validate 0 <= port <= 65535
and add the same validation on the inbound conversion path (the function that
accepts proto Owner/Port and produces the internal struct) so malformed owner
metadata is rejected early (reference task.Owner.Port and the conversion helper
functions in dispatch.go).
In `@internal/service/coordinator/handler.go`:
- Around line 310-315: When DispatchTaskToProto(claimed.Task) fails inside the
ClaimNext handling branch, ensure you release the claim before returning so the
task becomes available to other pollers; call the existing claim-release path
(the function that releases/unclaims claims in this component—e.g.,
ReleaseClaim/Unclaim/ReturnClaim on the coordinator or handler) with the
claimed.Task (or claimed) and handle any errors, then return the status.Error as
before; locate the ClaimNext handling code surrounding claimed.Task,
req.WorkerId and convert.DispatchTaskToProto to add this release call just prior
to returning on error.
In `@internal/service/coordinator/runtime_dispatcher.go`:
- Around line 17-24: Before returning the dispatcher, validate the TLS/config by
calling cfg.Validate() and handle any validation error the same way
coordinatorClient in engine.go does: call cfg.Validate() after setting fields
(before calling New(registry, cfg)) and propagate or return the error instead of
constructing a dispatcher with an invalid config; reference cfg.Validate() and
the New(registry, cfg) call when making this change.
---
Outside diff comments:
In `@internal/service/scheduler/dag_executor.go`:
- Around line 151-180: The distributed-execution branch in
shouldUseDistributedExecution(...) currently forwards any operation to
executor.CreateTask and dispatchToCoordinator(...) including
DispatchOperationUnspecified (and future unknown values); add the same
validation used by the local branch to reject DispatchOperationUnspecified (and
treat unknown/invalid enum values as errors) before building the task options,
returning an error if the operation is invalid so invalid operations are
rejected consistently and not dispatched to dispatchToCoordinator.
---
Nitpick comments:
In `@internal/cmd/context.go`:
- Around line 589-593: The closure returned by Context.RuntimeDispatcherFactory
declares a context.Context parameter that is unused; update the closure to
either use the blank identifier (func(_ context.Context) runtime.Dispatcher) or
add a brief comment above or inside the closure explaining the parameter is
intentionally unused for interface compatibility with DispatcherFactory, keeping
the body that returns coordinator.NewRuntimeDispatcher(c.ServiceRegistry,
c.Config.Core.Peer) unchanged and referencing the RuntimeDispatcherFactory
method and coordinator.NewRuntimeDispatcher to locate the code.
In `@internal/engine/engine.go`:
- Around line 205-209: The runtimeDispatcherFactory currently returns a
func(context.Context) runtime.Dispatcher that can yield nil because
coordinator.NewRuntimeDispatcher may return nil when e.serviceRegistry is nil;
update the factory to perform an explicit nil check and surface that case
instead of silently returning nil: change runtimeDispatcherFactory to return
func(context.Context) (runtime.Dispatcher, error) (or otherwise return an
error/panic) and inside the returned closure call
coordinator.NewRuntimeDispatcher(e.serviceRegistry, e.cfg.Core.Peer), check the
result for nil and return a clear error if nil (and update all callers of
runtimeDispatcherFactory to handle the error), or alternatively add a short doc
comment on runtimeDispatcherFactory that it may return nil if e.serviceRegistry
is nil if you prefer to keep the current signature.
In `@internal/runtime/agent/agent.go`:
- Around line 969-973: The log message refers to "coordinator client" but the
code now calls dispatcher.Cleanup; update the logger.Warn invocation to
reference the correct variable (dispatcher) so it reads something like "Failed
to cleanup dispatcher" (keep the same context and include tag.Error(err));
modify the logger.Warn call that wraps dispatcher.Cleanup to use the new
message.
- Around line 1561-1570: In Agent.createDispatcher, clarify that a nil
dispatcher can be expected in local-only execution by either updating the debug
log in the conditional that currently checks a.registry to read something like
"Dispatcher factory not configured; running in local-only mode" or by adding a
one-line comment above the logger.Debug call explaining that logging is
conditional on a.registry to avoid noisy logs in non-distributed setups;
reference the function name Agent.createDispatcher and the logger.Debug call so
you update the message or add the explanatory comment in that exact location.
In `@internal/service/coordinator/client.go`:
- Around line 960-1000: GetDAGRunStatus currently returns (nil, nil) if
attemptCall returns no error but resp is still nil; update the function
(GetDAGRunStatus) to treat a nil resp as an explicit protocol/error case by
returning a descriptive error instead of (nil, nil). Modify the post-attemptCall
check that currently does "if err != nil || resp == nil { return nil, err }" to
return a clear fmt.Errorf (e.g., "nil response from coordinator" or similar)
when resp == nil while preserving existing error propagation from attemptCall.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: e4a1ed45-49db-41ce-bf6c-3683f1e7d4f5
📒 Files selected for processing (59)
internal/cmd/context.gointernal/cmd/dry.gointernal/cmd/progress_remote.gointernal/cmd/restart.gointernal/cmd/retry.gointernal/cmd/start.gointernal/cmn/telemetry/collector_test.gointernal/core/exec/context.gointernal/core/exec/dispatch.gointernal/core/exec/dispatch_test.gointernal/core/exec/distributed.gointernal/engine/engine.gointernal/engine/run.gointernal/launcher/launcher.gointernal/launcher/launcher_test.gointernal/persis/store/distributed_dispatch.gointernal/persis/store/distributed_test.gointernal/proto/convert/dispatch.gointernal/proto/convert/dispatch_test.gointernal/runtime/agent/agent.gointernal/runtime/distributed_stale_run.gointernal/runtime/executor/dag_runner.gointernal/runtime/executor/dag_runner_test.gointernal/runtime/executor/task.gointernal/runtime/executor/task_test.gointernal/runtime/reporter.gointernal/service/coordinator/client.gointernal/service/coordinator/client_test.gointernal/service/coordinator/handler.gointernal/service/coordinator/handler_test.gointernal/service/coordinator/runtime_dispatcher.gointernal/service/frontend/api/v1/dagruns.gointernal/service/frontend/api/v1/dagruns_edit_retry.gointernal/service/frontend/api/v1/dagruns_edit_retry_internal_test.gointernal/service/frontend/api/v1/dagruns_retry_internal_test.gointernal/service/frontend/api/v1/dags.gointernal/service/frontend/api/v1/metrics_test.gointernal/service/frontend/api/v1/proc_liveness_test.gointernal/service/frontend/api/v1/workers_internal_test.gointernal/service/scheduler/dag_executor.gointernal/service/scheduler/dag_executor_test.gointernal/service/scheduler/queue_dispatcher.gointernal/service/scheduler/queue_dispatcher_test.gointernal/service/scheduler/queue_processor_startup_test.gointernal/service/scheduler/queue_processor_test.gointernal/service/scheduler/scheduler.gointernal/service/worker/coordreport/artifact_uploader.gointernal/service/worker/coordreport/artifact_uploader_test.gointernal/service/worker/coordreport/export_test.gointernal/service/worker/coordreport/log_streamer.gointernal/service/worker/coordreport/log_streamer_test.gointernal/service/worker/coordreport/status_pusher.gointernal/service/worker/coordreport/status_pusher_test.gointernal/service/worker/handler.gointernal/service/worker/handler_test.gointernal/service/worker/poller_test.gointernal/service/worker/remote_handler.gointernal/service/worker/remote_handler_test.gointernal/test/helper.go
💤 Files with no reviewable changes (1)
- internal/core/exec/context.go
Comment on lines
+20
to
+22
| if task.Owner.Port < math.MinInt32 || task.Owner.Port > math.MaxInt32 { | ||
| return nil, fmt.Errorf("owner coordinator port out of range: %d", task.Owner.Port) | ||
| } |
There was a problem hiding this comment.
Reject impossible coordinator ports in both directions.
These helpers currently accept values like -1 or 70000: the outbound path only guards the int32 cast, and the inbound path does no validation at all. That lets malformed owner metadata survive conversion and fail later when the worker tries to ack/report back to the owner coordinator.
Proposed fix
+const maxCoordinatorPort = 65535
+
func DispatchTaskToProto(task *exec.DispatchTask) (*coordinatorv1.Task, error) {
if task == nil {
return nil, nil
}
- if task.Owner.Port < math.MinInt32 || task.Owner.Port > math.MaxInt32 {
+ if task.Owner.Port <= 0 || task.Owner.Port > maxCoordinatorPort {
return nil, fmt.Errorf("owner coordinator port out of range: %d", task.Owner.Port)
}
@@
func ProtoToDispatchTask(task *coordinatorv1.Task) (*exec.DispatchTask, error) {
if task == nil {
return nil, nil
}
+ if task.OwnerCoordinatorPort <= 0 || task.OwnerCoordinatorPort > maxCoordinatorPort {
+ return nil, fmt.Errorf("owner coordinator port out of range: %d", task.OwnerCoordinatorPort)
+ }
dispatchTask := &exec.DispatchTask{Also applies to: 99-103
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/proto/convert/dispatch.go` around lines 20 - 22, The owner
coordinator port validation is incomplete: ensure you reject impossible port
values (e.g. -1, >65535) in both conversion directions by adding explicit range
checks around task.Owner.Port and any reciprocal conversion helpers in
internal/proto/convert/dispatch.go; update the outbound check (currently
guarding int32 cast) to validate 0 <= port <= 65535 and add the same validation
on the inbound conversion path (the function that accepts proto Owner/Port and
produces the internal struct) so malformed owner metadata is rejected early
(reference task.Owner.Port and the conversion helper functions in dispatch.go).
Comment on lines
310
to
+315
| if claimed != nil && claimed.Task != nil { | ||
| claimed.Task.WorkerId = req.WorkerId | ||
| return &coordinatorv1.PollResponse{Task: claimed.Task}, nil | ||
| claimed.Task.WorkerID = req.WorkerId | ||
| task, err := convert.DispatchTaskToProto(claimed.Task) | ||
| if err != nil { | ||
| return nil, status.Error(codes.Internal, "failed to encode claimed task: "+err.Error()) | ||
| } |
There was a problem hiding this comment.
Release the claim if encoding the claimed task fails.
ClaimNext has already made this task unavailable to other pollers. Returning here on DispatchTaskToProto error leaves the claim stranded until its TTL expires, so one malformed record can keep surfacing as an internal error and stall queue progress.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/service/coordinator/handler.go` around lines 310 - 315, When
DispatchTaskToProto(claimed.Task) fails inside the ClaimNext handling branch,
ensure you release the claim before returning so the task becomes available to
other pollers; call the existing claim-release path (the function that
releases/unclaims claims in this component—e.g.,
ReleaseClaim/Unclaim/ReturnClaim on the coordinator or handler) with the
claimed.Task (or claimed) and handle any errors, then return the status.Error as
before; locate the ClaimNext handling code surrounding claimed.Task,
req.WorkerId and convert.DispatchTaskToProto to add this release call just prior
to returning on error.
Comment on lines
+17
to
+24
| cfg := DefaultConfig() | ||
| cfg.MaxRetries = 50 | ||
| cfg.CAFile = peerConfig.ClientCaFile | ||
| cfg.CertFile = peerConfig.CertFile | ||
| cfg.KeyFile = peerConfig.KeyFile | ||
| cfg.SkipTLSVerify = peerConfig.SkipTLSVerify | ||
| cfg.Insecure = peerConfig.Insecure | ||
| return New(registry, cfg) |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Add config validation before returning the dispatcher.
The coordinatorClient method in engine.go (line 195) validates the coordinator config by calling cfg.Validate(). This NewRuntimeDispatcher function should do the same to catch invalid TLS configurations early.
🔒 Proposed fix to add validation
cfg.CertFile = peerConfig.CertFile
cfg.KeyFile = peerConfig.KeyFile
cfg.SkipTLSVerify = peerConfig.SkipTLSVerify
cfg.Insecure = peerConfig.Insecure
+ if err := cfg.Validate(); err != nil {
+ return nil
+ }
return New(registry, cfg)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| cfg := DefaultConfig() | |
| cfg.MaxRetries = 50 | |
| cfg.CAFile = peerConfig.ClientCaFile | |
| cfg.CertFile = peerConfig.CertFile | |
| cfg.KeyFile = peerConfig.KeyFile | |
| cfg.SkipTLSVerify = peerConfig.SkipTLSVerify | |
| cfg.Insecure = peerConfig.Insecure | |
| return New(registry, cfg) | |
| cfg.CertFile = peerConfig.CertFile | |
| cfg.KeyFile = peerConfig.KeyFile | |
| cfg.SkipTLSVerify = peerConfig.SkipTLSVerify | |
| cfg.Insecure = peerConfig.Insecure | |
| if err := cfg.Validate(); err != nil { | |
| return nil, err | |
| } | |
| return New(registry, cfg) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/service/coordinator/runtime_dispatcher.go` around lines 17 - 24,
Before returning the dispatcher, validate the TLS/config by calling
cfg.Validate() and handle any validation error the same way coordinatorClient in
engine.go does: call cfg.Validate() after setting fields (before calling
New(registry, cfg)) and propagate or return the error instead of constructing a
dispatcher with an invalid config; reference cfg.Validate() and the
New(registry, cfg) call when making this change.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 59 files
Partial review: This PR has more than 50 files, so cubic reviewed the highest-priority files first. During the trial, paid plans get a higher file limit.
You can try an ultrareview to bypass the file limit, comment @cubic-dev-ai ultrareview. Learn more.
Fix all with cubic | Re-trigger cubic
yohamta0
force-pushed
the
refactor/runtime-reporter-port
branch
from
b36ee6a to
f64be5c
Compare
yohamta0
force-pushed
the
refactor/runtime-reporter-port
branch
from
f64be5c to
b7308bd
Compare
yohamta0
force-pushed
the
refactor/runtime-reporter-port
branch
from
88fdb07 to
9001832
Compare
This was referenced Jun 1, 2026
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
1 participant
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
Testing
Summary by cubic
Moves coordinator transport behind runtime ports and switches the runtime to domain dispatch/status/worker types. Adds a TLS‑validated, coordinator‑backed runtime dispatcher injected via a DispatcherFactory; keeps the runtime proto‑free and preserves distributed sub‑DAG behavior.
New Features
exec.DispatchTaskandexec.DispatchOperation, with converters ininternal/proto/convert/dispatch; used by coordinator client/handler with validation.runtime.StatusPusher,runtime.SchedulerLogStreamer,runtime.ArtifactFinalizer(withAttemptRejected); implemented in service/worker/coordreport.coordinator.NewRuntimeDispatcher(validates TLS/retry); wired through aDispatcherFactoryin CLI, engine, and tests.Refactors
exec.DAGRunStatus.DispatchTaskStorenow persistsexec.DispatchTask; releases claims on encode failure; adaptive cleanup interval reduces contention; can read legacy dispatch records.exec.DispatchTask; telemetry now usesexec.WorkerStats/exec.RunningTask.Written for commit 9001832. Summary will update on new commits.
Summary by CodeRabbit