wip: feat(engine): dags-as-durable-tasks by mrkaye97 · Pull Request #4266 · hatchet-dev/hatchet

mrkaye97 · 2026-06-23T20:36:44Z

Description

WIP

Initial implementation of the dag operator, which basically creates an internal durable task that processes the dag via child spawning

to do:

Fixes # (issue)

Type of change

New feature (non-breaking change which adds functionality)

What's Changed

Add a list of tasks or features here...

Checklist

Changes have been:

Tested (unit, integration, or manually with steps specified)
Linted and formatted
Documented (where applicable)
Added to CHANGELOG (where applicable) -- see Keep a Changelog

🤖 AI Disclosure

I acknowledge that an LLM was used in the creation of this Pull Request, in accordance with Hatchet's AI_POLICY.md.

Details: [e.g. generating tests, writing docs]

…or is created

vercel · 2026-06-23T20:36:49Z

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
hatchet-docs	Ready	Preview, Comment	Jul 15, 2026 6:52pm

github-actions · 2026-06-23T20:38:36Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-06-23T20:48:43Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=false). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-06-23T20:48:49Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=true). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-06-24T17:43:21Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

This reverts commit abc0dc7.

github-actions · 2026-06-24T17:53:32Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=true). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-06-24T17:53:35Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=false). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-06-24T20:30:25Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-07-13T16:00:33Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-07-13T16:09:19Z

⚠️ Optional test failure: The old-engine-new-sdk (Python) job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-07-13T16:10:58Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=false). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-07-13T16:14:59Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

mrkaye97 · 2026-07-13T19:18:37Z

    ),
    cell: ({ row }) => {
-      if (row.getCanExpand()) {
+      if (row.original.type === 'DAG') {


this lets us open the workflow run detail page for an operator DAG before any of the tasks have actually been spawned

mrkaye97 · 2026-07-13T19:18:57Z

-    [taskRuns],
+    () =>
+      taskRuns
+        .filter((t) => t.isDurable && t.taskExternalId !== id)


there's a bunch of stuff like this - basically removing the orchestrator from the ids

mrkaye97 · 2026-07-13T19:19:46Z

+		if e.isDagOrchestrator {
+			continue
+		}


don't want to render spans for the orchestrator task since it messes up nesting and is "internal" / not user facing

mrkaye97 · 2026-07-13T19:21:10Z

-					d.l.Warn().Ctx(ctx).Err(err).Msg("failed to unmarshal input")
-					continue
+		if len(currInput.DagParentTaskRunIds) > 0 {
+			dagParentOutputs, err := d.repov1.Tasks().GetDagParentOutputs(ctx, tenantId, currInput.DagParentTaskRunIds)


not sure what to do about this - seems very suboptimal to need a db lookup here, will circle back

mrkaye97 · 2026-07-13T19:22:09Z

+						continue
+					}
+				} else {
+					childEvents, err := s.repov1.Tasks().ListDurableOrchestratorChildOutputEvents(ctx, tenantId, event.TaskExternalId)


same here - really not ideal to need the db lookup, will circle back

mrkaye97 · 2026-07-13T19:24:06Z

+	return nil
+}
+
+func (d *DispatcherServiceImpl) TriggerDAGStep(ctx context.Context, tenantId uuid.UUID, req *operator.DAGStepTriggerRequest) (*operator.DAGStepTriggerResult, error) {


this is one of the important methods, since it's what gets called by the operator to trigger tasks

mrkaye97 · 2026-07-13T19:29:48Z

+		TriggerTaskData: &v1.TriggerTaskData{
+			WorkflowName:         req.WorkflowName,
+			TargetActionId:       &req.ActionId,
+			UserMessage:          &stepLabel,
+			Data:                 []byte(req.Input),
+			AdditionalMetadata:   task.AdditionalMetadata,
+			ParentExternalId:     &task.ExternalID,
+			ParentTaskId:         &task.ID,
+			ParentTaskInsertedAt: &task.InsertedAt.Time,
+			ChildIndex:           &childIndex,
+			DagParentTaskRunIds:  req.DagParentTaskRunIds,
+			IsSkipped:            req.IsSkipped,
+			IsCancelled:          req.IsCancelled,
+			DesiredWorkerLabels:  req.DesiredWorkerLabels,
+			WorkflowRunId:        &orchestratorWorkflowRunId,


important stuff here:

we basically pass most things along (e.g. the metadata, the input, etc.)

we tie the workflow run id manually to the orchestrator's workflow run id

mrkaye97 · 2026-07-13T19:40:00Z

+		if t.output != nil {
+			b, err := json.Marshal(t.output)
+			if err == nil {
+				output[t.readableId] = json.RawMessage(b)


this results in a DAG output that maps step ids to outputs, similar to what we see in the dash and over the API

mrkaye97 · 2026-07-13T20:36:19Z

+		-- fixme: this might be really slow
+		AND NOT EXISTS (
+			SELECT 1
+			FROM v1_tasks_olap t
+			WHERE
+				t.external_id = r.parent_task_external_id
+				AND COALESCE(t.is_dag_orchestrator, FALSE)
+				AND t.tenant_id = r.tenant_id
+		)


so, this makes me very nervous, but I'm not sure how else to do this. Basically, we need to remove any dag children from the runs list query, since they'll be double counted (we should only include dag parents, dags, and standalone tasks)

probably a broader conversation, but should we not be treating the OLAP side of things as static, so we just insert data as if we're inserting DAGs?

ah, I didn't even think about that... 🤔 it would make all the API surface stuff a lot easier

I guess the negative is we'd still need v1_dags_olap, v1_dag_to_task_olap, etc.?

yeah - I'm not sure it's a long-term fix, but short-term it would make our lives easier (obviously just moves complexity somewhere else)

lemme see if I can get it to work. maybe when we eventually do V2, we can do that second push for OLAP since we potentially won't need to support as much old API surface

github-actions · 2026-07-13T20:40:56Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-07-13T20:48:28Z

⚠️ Optional test failure: The old-engine-new-sdk (Python) job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

mrkaye97 · 2026-07-13T20:51:03Z

+		-- fixme: this might be really slow
+		AND NOT EXISTS (
+			SELECT 1
+			FROM v1_tasks_olap t
 			WHERE
-				lt.tenant_id = $1::uuid
-				AND lt.external_id = $11::UUID
+				t.external_id = r.parent_task_external_id
+				AND COALESCE(t.is_dag_orchestrator, FALSE)
+				AND t.tenant_id = r.tenant_id
 		)


github-actions · 2026-07-13T20:51:16Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=true). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

abelanger5 · 2026-07-14T14:00:05Z

 type dag struct {
-	requestCh chan<- *v1contracts.DurableTaskRequest
+	requestCh   chan<- *v1contracts.DurableTaskRequest
+	matchRepo   repository.MatchRepository


I think pulling out the EvaluateBoolExpr into a subpackage with essentially no dependencies (particularly no DB dependencies) would make it much easier to test this file.

This also seems like a good candidate for table-test unit testing with a set of comprehensive examples; in particular, we'd likely want code coverage on all of the feature paths (wait conditions, etc).

I'm aware that building out testing for this file would require the test to sort of act like its own mini workflow engine, since it needs to be able to receive and send durable task req/responses over the channels...I'm assuming that hardcoding when we send data over these channels for each test is simple enough that this is doable, but I could also see a world where it's complex enough that we don't want to do this

abelanger5 · 2026-07-14T14:03:38Z

+			if !ok {
+				return fmt.Errorf("durable task session closed")
+			}
+			d.taskConsumer(resp)


q: did we ever merge that "stable callback ordering" PR? That's required for this to work

not yet, but yeah, agreed - there are some open comments on that one still

abelanger5 · 2026-07-14T14:06:03Z

+	isTriggered  bool
+	isSkipped    bool
+	errorMessage string
+	output       map[string]interface{}


I don't think this is tractable, I think we need to find a way to not keep the output of the task in memory for longer than necessary. Perhaps we can invert the expression evaluation to evaluate the output when the parent completes instead of evaluating it on the child.

abelanger5 · 2026-07-14T14:09:18Z

-		for _, parent := range task.parents {
-			if !parent.isCompleted {
-				areParentsSatisfied = false
+	for _, t := range d.tasks {


I wonder if it's more beneficial to keep a list of pendingTasks versus finalizedTasks so that we only loop through pendingTasks which could theoretically still emit an event

abelanger5 · 2026-07-14T14:13:51Z

+			}
+		}
+
+		result, err := d.triggerStep(ctx, t.actionId, t.workflowName, t.index, parentTaskRunIds, skip, cancelled)


does this need to be synchronous / blocking the loop, or can we add to a waitgroup and evaluate at the bottom of taskEmitter?

Perhaps this wouldn't get us anything though, because we need to guard the event log which I imagine we touch when we call this method, so we serialize it anyway

abelanger5 · 2026-07-14T14:18:48Z

nit: a quick pass adding spans/traces throughout this file would be awesome

abelanger5 · 2026-07-14T14:22:25Z

+		-- fixme: this might be really slow
+		AND NOT EXISTS (
+			SELECT 1
+			FROM v1_tasks_olap t
+			WHERE
+				t.external_id = r.parent_task_external_id
+				AND COALESCE(t.is_dag_orchestrator, FALSE)
+				AND t.tenant_id = r.tenant_id
+		)


probably a broader conversation, but should we not be treating the OLAP side of things as static, so we just insert data as if we're inserting DAGs?

abelanger5 · 2026-07-14T14:26:46Z

+		return fmt.Errorf("could not marshal DAG operator config: %w", err)
+	}
+
+	_, err = a.repo.Operators().CreateOperator(ctx, tenantId, v1.CreateOperatorOpts{


it looks like we only call this on PutWorkflow. Should we be calling this from other places, or is the thinking that this only really enables when a user restarts their workers?

I was thinking it was fine to only do this here since the isDagOperator flag is tied to the WorkflowVersion, so until someone registers a new workflow version with a dag, we'd never end up needing to call this, which I think is fine. We could also call it on startup I guess?

github-actions · 2026-07-15T17:45:59Z

⚠️ Optional test failure: The load-deadlock job failed on this PR (optimistic-scheduling=true). This check is non-mandatory and does not block merging, but may be worth investigating. View logs

github-actions · 2026-07-15T18:51:31Z

⚠️ Optional test failure: The load-online-migrate job failed on this PR. This check is non-mandatory and does not block merging, but may be worth investigating. View logs

mrkaye97 added 7 commits June 23, 2026 12:19

feat: sql, move some logic over from my old branch

ac147cb

feat: call ensureDAGOperator on PutWorkflow so we know the operat…

a043707

…or is created

chore: gen

5b65eda

feat: first pass at operator implementation

7c6f92e

feat: task repo wiring

3a3f6ae

feat: more initial wiring, still very buggy

85029fd

feat: it sort of works?

58048bb

github-actions Bot added the engine Related to the core Hatchet engine label Jun 23, 2026

vercel Bot deployed to Preview June 23, 2026 20:37 View deployment

fix: handle panic and shutdown / cancel

a058731

mrkaye97 added 8 commits June 24, 2026 12:20

fix: get rid of the need to flatten workflow tasks into standalones

cc05929

feat: pass user message through to fe

61b2beb

feat: add TriggerDagStep method on the dispatcher

1059be1

feat: triggerDagStep wrapper

9d5ac6b

chore: lint, internals

de5c970

feat: start wiring up the olap side

07ba3d7

feat: map dag tasks to their parents so we get proper nesting

50a344f

fix: handle expansion in side panel

abc0dc7

github-actions Bot added the dashboard Related to the Hatchet dashboard label Jun 24, 2026

vercel Bot deployed to Preview June 24, 2026 17:44 View deployment

Revert "fix: handle expansion in side panel"

e812939

This reverts commit abc0dc7.

fix: render minimap correctly

be7655d

vercel Bot deployed to Preview July 13, 2026 16:01 View deployment

feat: render dag even when empty

97006a9

vercel Bot deployed to Preview July 13, 2026 16:16 View deployment

mrkaye97 commented Jul 13, 2026

View reviewed changes

feat: more slice helpers

5ff8383

mrkaye97 commented Jul 13, 2026

View reviewed changes

fix: improve error handling a bit

2c6c710

vercel Bot deployed to Preview July 13, 2026 20:42 View deployment

mrkaye97 commented Jul 13, 2026

View reviewed changes

abelanger5 reviewed Jul 14, 2026

View reviewed changes

Merge branch 'belanger/operators' into mk/durable-tasks-as-operator

52c7b5d

vercel Bot deployed to Preview July 15, 2026 17:37 View deployment

feat: fable one-shot rewire to spoof the OLAP DAGs

b06cd7a

vercel Bot deployed to Preview July 15, 2026 18:52 View deployment

Uh oh!

Conversation

mrkaye97 commented Jun 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Type of change

What's Changed

Checklist

Uh oh!

vercel Bot commented Jun 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Jun 23, 2026

Uh oh!

github-actions Bot commented Jun 23, 2026

Uh oh!

github-actions Bot commented Jun 23, 2026

Uh oh!

github-actions Bot commented Jun 24, 2026

Uh oh!

github-actions Bot commented Jun 24, 2026

Uh oh!

github-actions Bot commented Jun 24, 2026

Uh oh!

github-actions Bot commented Jun 24, 2026

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrkaye97 commented Jun 23, 2026 •

edited

Loading

vercel Bot commented Jun 23, 2026 •

edited

Loading