Example 12: Worker Containers

What It Is

Worker Containers let an application run a DAG body outside the main execution context while keeping the parent graph unchanged. The Cartographer binds a scatter body to a browser worker pool; the Node companion uses the same container seam for worker-thread style deployments.

The important contract is the role binding: the DAG declares a logical container role, and the host decides which DagContainerInterface implementation satisfies that role.

How It Works

The parent placement declares a logical container role. The host binds that role to a DagContainerInterface implementation, and the worker loads a registry module that reconstructs the DAG bundle and services inside the isolated context. The dispatcher sends clone tasks to the container, receives outcomes, and applies the same gather and reducer semantics it uses in-process.

This keeps worker adoption incremental. Application authors do not fork their DAG into "local" and "worker" versions; they bind a role when isolation or parallel throughput is needed.

Diagrams, Examples, and Outputs

DAG registration and diagram

The parent DAG declares the container-bound scatter; the worker DAG is the body registered inside the worker-host registry. The Cartographer is the in-browser owner for this principle through its container-role stream processing path.

Cartographer worker parent DAG

6 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:cartographer",
  "@type": "DAG",
  "name": "dag:cartographer",
  "version": "1.0",
  "entrypoints": {
    "position-ping": "urn:noocodec:dag:cartographer/node/intake-gather",
    "facility-scan": "urn:noocodec:dag:cartographer/node/intake-gather",
    "sensor-reading": "urn:noocodec:dag:cartographer/node/intake-gather",
    "customs-event": "urn:noocodec:dag:cartographer/node/intake-gather",
    "delivery-confirmation": "urn:noocodec:dag:cartographer/node/intake-gather"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:cartographer/node/intake-gather",
      "@type": "GatherNode",
      "name": "dag:cartographer/node/intake-gather",
      "sources": {
        "urn:noocodec:dag:cartographer/entrypoint/position-ping": {},
        "urn:noocodec:dag:cartographer/entrypoint/facility-scan": {},
        "urn:noocodec:dag:cartographer/entrypoint/sensor-reading": {},
        "urn:noocodec:dag:cartographer/entrypoint/customs-event": {},
        "urn:noocodec:dag:cartographer/entrypoint/delivery-confirmation": {}
      },
      "gather": {
        "strategy": "source-intake"
      },
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/process-stream",
        "error": "urn:noocodec:dag:cartographer/node/process-stream",
        "empty": "urn:noocodec:dag:cartographer/node/failed"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/process-stream",
      "@type": "ScatterNode",
      "name": "dag:cartographer/node/process-stream",
      "source": "sources",
      "body": {
        "dag": "urn:noocodec:dag:stream-event"
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:cartographer/node/fold-insights",
        "partial": "urn:noocodec:dag:cartographer/node/fold-insights",
        "all-error": "urn:noocodec:dag:cartographer/node/fold-insights",
        "empty": "urn:noocodec:dag:cartographer/node/summarize-insights"
      },
      "itemKey": "source-payload",
      "reducer": "aggregate",
      "container": "cpu",
      "execution": {
        "mode": "reservoir",
        "concurrency": 16,
        "reservoir": {
          "keyField": "eventType",
          "capacity": 1000
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/fold-insights",
      "@type": "GatherNode",
      "name": "dag:cartographer/node/fold-insights",
      "sources": {
        "urn:noocodec:dag:cartographer/node/process-stream": {}
      },
      "gather": {
        "strategy": "insights-fold"
      },
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/summarize-insights",
        "error": "urn:noocodec:dag:cartographer/node/summarize-insights",
        "empty": "urn:noocodec:dag:cartographer/node/summarize-insights"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/summarize-insights",
      "@type": "EmbeddedDAGNode",
      "name": "dag:cartographer/node/summarize-insights",
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/done",
        "error": "urn:noocodec:dag:cartographer/node/failed"
      },
      "dag": "urn:noocodec:dag:insights-summary",
      "container": "io"
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/done",
      "@type": "TerminalNode",
      "name": "dag:cartographer/node/done",
      "outcome": "completed"
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/failed",
      "@type": "TerminalNode",
      "name": "dag:cartographer/node/failed",
      "outcome": "failed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:cartographer (v1.0)
  entry_position-ping(["position-ping"])
  entry_position-ping --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_facility-scan(["facility-scan"])
  entry_facility-scan --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_sensor-reading(["sensor-reading"])
  entry_sensor-reading --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_customs-event(["customs-event"])
  entry_customs-event --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_delivery-confirmation(["delivery-confirmation"])
  entry_delivery-confirmation --> urn_noocodec_dag_cartographer/node/intake-gather
  urn_noocodec_dag_cartographer/node/intake-gather{"dag_cartographer/node/intake-gather"}
  urn_noocodec_dag_cartographer/node/intake-gather -->|success| urn_noocodec_dag_cartographer/node/process-stream
  urn_noocodec_dag_cartographer/node/intake-gather -->|error| urn_noocodec_dag_cartographer/node/process-stream
  urn_noocodec_dag_cartographer/node/intake-gather -->|empty| urn_noocodec_dag_cartographer/node/failed
  urn_noocodec_dag_cartographer/node/process-stream[/"dag:cartographer/node/process-stream ▣ eventType ×1000"/]
  urn_noocodec_dag_cartographer/node/process-stream -->|all-success| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|partial| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|all-error| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|empty| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/fold-insights{"dag_cartographer/node/fold-insights"}
  urn_noocodec_dag_cartographer/node/fold-insights -->|success| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/fold-insights -->|error| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/fold-insights -->|empty| urn_noocodec_dag_cartographer/node/summarize-insights
  urn_noocodec_dag_cartographer/node/summarize-insights[["dag:cartographer/node/summarize-insights"]]
  urn_noocodec_dag_cartographer/node/summarize-insights -->|success| urn_noocodec_dag_cartographer/node/done
  urn_noocodec_dag_cartographer/node/summarize-insights -->|error| urn_noocodec_dag_cartographer/node/failed
  urn_noocodec_dag_cartographer/node/done((("dag:cartographer/node/done")))
  urn_noocodec_dag_cartographer/node/failed>"dag:cartographer/node/failed"]
  classDef contained-cpu fill:#b45309,stroke:#d97706,color:#eef3f7
  class urn_noocodec_dag_cartographer/node/process-stream contained-cpu
  classDef contained-io fill:#7c3aed,stroke:#8b5cf6,color:#eef3f7
  class urn_noocodec_dag_cartographer/node/summarize-insights contained-io
  classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
  class urn_noocodec_dag_cartographer/node/process-stream reservoir

stream-event worker body DAG

9 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:stream-event",
  "@type": "DAG",
  "name": "dag:stream-event",
  "version": "1.0",
  "entrypoints": {
    "main": "urn:noocodec:dag:stream-event/node/decode-payload"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:stream-event/node/decode-payload",
      "@type": "SingleNode",
      "name": "dag:stream-event/node/decode-payload",
      "node": "urn:noocodec:node:decode-payload",
      "outputs": {
        "decoded": "urn:noocodec:dag:stream-event/node/route-event-type-variant",
        "invalid": "urn:noocodec:dag:stream-event/node/rejected"
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/route-event-type-variant",
      "@type": "SingleNode",
      "name": "dag:stream-event/node/route-event-type-variant",
      "node": "urn:noocodec:node:route-event-type-variant",
      "outputs": {
        "position-ping": "urn:noocodec:dag:stream-event/node/pipeline-position-ping",
        "sensor-reading": "urn:noocodec:dag:stream-event/node/pipeline-sensor-reading",
        "customs-event": "urn:noocodec:dag:stream-event/node/pipeline-customs-event",
        "facility-scan": "urn:noocodec:dag:stream-event/node/pipeline-facility-scan",
        "delivery-confirmation": "urn:noocodec:dag:stream-event/node/pipeline-delivery-confirmation"
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-position-ping",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-position-ping",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-position-ping",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "legKm": "legKm",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-sensor-reading",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-sensor-reading",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-sensor-reading",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "coldChainBreach": "coldChainBreach",
          "legKm": "legKm",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-customs-event",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-customs-event",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-customs-event",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "customsDwellHours": "customsDwellHours",
          "legKm": "legKm",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-facility-scan",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-facility-scan",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-facility-scan",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "pricedOrder": "pricedOrder",
          "shippingQuote": "shippingQuote",
          "deliveryEstimate": "deliveryEstimate",
          "legKm": "legKm",
          "gdprResult": "gdprResult",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/pipeline-delivery-confirmation",
      "@type": "EmbeddedDAGNode",
      "name": "dag:stream-event/node/pipeline-delivery-confirmation",
      "outputs": {
        "success": "urn:noocodec:dag:stream-event/node/done",
        "error": "urn:noocodec:dag:stream-event/node/rejected"
      },
      "dag": "urn:noocodec:dag:pipeline-delivery-confirmation",
      "stateMapping": {
        "output": {
          "canonicalVariant": "canonicalVariant",
          "raw": "raw",
          "normalized": "normalized",
          "currentEvent": "currentEvent",
          "geoContext": "geoContext",
          "resolvedGeo": "resolvedGeo",
          "legKm": "legKm",
          "gdprResult": "gdprResult",
          "routing": "routing",
          "enriched": "enriched",
          "capturedErrors": "capturedErrors"
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/done",
      "@type": "TerminalNode",
      "name": "dag:stream-event/node/done",
      "outcome": "completed"
    },
    {
      "@id": "urn:noocodec:dag:stream-event/node/rejected",
      "@type": "TerminalNode",
      "name": "dag:stream-event/node/rejected",
      "outcome": "failed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:stream-event (v1.0)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_stream-event/node/decode-payload
  urn_noocodec_dag_stream-event/node/decode-payload["dag:stream-event/node/decode-payload"]
  urn_noocodec_dag_stream-event/node/decode-payload -->|decoded| urn_noocodec_dag_stream-event/node/route-event-type-variant
  urn_noocodec_dag_stream-event/node/decode-payload -->|invalid| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/route-event-type-variant["dag:stream-event/node/route-event-type-variant"]
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|position-ping| urn_noocodec_dag_stream-event/node/pipeline-position-ping
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|sensor-reading| urn_noocodec_dag_stream-event/node/pipeline-sensor-reading
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|customs-event| urn_noocodec_dag_stream-event/node/pipeline-customs-event
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|facility-scan| urn_noocodec_dag_stream-event/node/pipeline-facility-scan
  urn_noocodec_dag_stream-event/node/route-event-type-variant -->|delivery-confirmation| urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation
  urn_noocodec_dag_stream-event/node/pipeline-position-ping[["dag:stream-event/node/pipeline-position-ping"]]
  urn_noocodec_dag_stream-event/node/pipeline-position-ping -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-position-ping -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-sensor-reading[["dag:stream-event/node/pipeline-sensor-reading"]]
  urn_noocodec_dag_stream-event/node/pipeline-sensor-reading -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-sensor-reading -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-customs-event[["dag:stream-event/node/pipeline-customs-event"]]
  urn_noocodec_dag_stream-event/node/pipeline-customs-event -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-customs-event -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-facility-scan[["dag:stream-event/node/pipeline-facility-scan"]]
  urn_noocodec_dag_stream-event/node/pipeline-facility-scan -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-facility-scan -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation[["dag:stream-event/node/pipeline-delivery-confirmation"]]
  urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation -->|success| urn_noocodec_dag_stream-event/node/done
  urn_noocodec_dag_stream-event/node/pipeline-delivery-confirmation -->|error| urn_noocodec_dag_stream-event/node/rejected
  urn_noocodec_dag_stream-event/node/done((("dag:stream-event/node/done")))
  urn_noocodec_dag_stream-event/node/rejected>"dag:stream-event/node/rejected"]

This example runs a scatter-dag-body placement over a real browser WebWorkerContainer pool from @studnicky/dagonizer-executor-web. Each scatter clone's sub-DAG executes in a worker; the results are gathered back into the parent state identically to the in-process path.

Run

npm run docs:dev

Open The Cartographer, click Run, and watch process-stream execute through the worker-backed body DAG.

What It Lets You Do

Worker containers let applications execute a DAG body in an isolated worker pool while preserving the same parent DAG, state mapping, gather, and lifecycle contracts. Use this when a browser or Node host needs CPU isolation, parallel throughput, crash containment, or a deployment seam for clone-level work.

Code Samples

Read the snippets with the diagrams nearby so the TypeScript behavior, JSON-LD graph shape, and runtime output line up as one contract.

Container-bound DAG placement

DAG authoring does not change between the in-process and worker-thread paths. The only difference is the container key on the scatter placement and the containers option on the dispatcher:

/** Default reservoir capacity for the process-stream scatter in the workers DAG. */
export const DEFAULT_RESERVOIR_CAPACITY = 1000;

/**
 * CartographerWorkersDag: static factory for the cartographer-workers DAG and
 * its associated dispatcher bundle. Consumers call CartographerWorkersDag.build()
 * (DAG only) or CartographerWorkersDag.bundle() (full dispatcher bundle) with an
 * optional reservoir capacity override.
 *
 * DAG topology — identical to cartographerDAG with containerized boundaries:
 *   - container: 'cpu' so each stream-event body runs inside a
 *     WorkerThreadContainer/WebWorkerContainer rather than in-process.
 *   - container: 'io' so the final summary runs through the same embedded-DAG
 *     interface used by plugins and nested flows.
 *   - reservoir.capacity is parameterised; callers pass their UI-controlled
 *     batch size rather than relying on the compile-time default.
 *
 *   5 data-type entrypoints → gather('intake-gather', source-intake)
 *     → scatter('process-stream', 'sources', { dag: 'stream-event' },
 *               concurrency: 16, container: 'cpu', reservoir: { capacity })
 *     → gather('fold-insights', strategy: insights-fold)
 *     → embed('summarize-insights', 'insights-summary', container: 'io')
 *     → done
 */
export class CartographerWorkersDag {
  private constructor() { /* static-only */ }

  /**
   * Build the cartographer-workers DAG with the given reservoir capacity.
   * CLI, smoke tests, and dag-validate consumers use cartographerWorkersDAG
   * (the pre-built constant); the browser demo calls this with a UI-controlled value.
   */
  static build(capacity: number = DEFAULT_RESERVOIR_CAPACITY): DAGType {
    return new DAGBuilder(CARTOGRAPHER_DAG_IRI, '1.0')

      .gather(
        CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_DAG_IRI),
        { 'strategy': 'source-intake' },
        {
          'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
          'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
          'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'),
        },
      )

      .scatter(
        CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
        'sources',
        { 'dag': STREAM_EVENT_DAG_IRI },
        {
          'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
          'partial':     CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
          'all-error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
          'empty':       CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
        },
        {
          'itemKey':     'source-payload',
          'container':   'cpu',
          'execution': { 'mode': 'reservoir', 'concurrency': 16, 'reservoir': { 'keyField': 'eventType', 'capacity': capacity } },
        },
      )
      .gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'), {
        [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream')]: {},
      }, { 'strategy': 'insights-fold' }, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
        'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'),
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize-insights'), INSIGHTS_SUMMARY_DAG_IRI, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'),
      }, {
        'container': 'io',
      })

      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'), { outcome: 'completed' })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'failed'), { outcome: 'failed' })

      .entrypoints({
        'position-ping':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'facility-scan':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'sensor-reading':         CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'customs-event':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
        'delivery-confirmation':  CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
      })

      .build();
  }

  /**
   * Build the workers bundle with a configurable reservoir capacity. The returned
   * bundle is identical to cartographerWorkersBundle except that its cartographer
   * DAG is built with CartographerWorkersDag.build(capacity) so the process-stream
   * scatter uses the caller-supplied batch size.
   *
   * Used by the browser demo to wire UI-controlled knobs into each run() without
   * mutating the shared default-capacity constants.
   */
  static bundle(
    capacity: number = DEFAULT_RESERVOIR_CAPACITY,
  ): DispatcherBundleType<CartographerState> {
    return {
      'nodes': [
        ...cartographerWorkerRuntimeBundle.nodes,
      ],
      'dags': [
        ...cartographerWorkerRuntimeBundle.dags,
        CartographerWorkersDag.build(capacity),
      ],
    };
  }
}

/**
 * cartographerWorkersDAG: pre-built workers DAG at DEFAULT_RESERVOIR_CAPACITY.
 * CLI, smoke tests, and dag-validate consumers use this constant; the browser
 * demo uses CartographerWorkersDag.build(capacity) with a UI-controlled value.
 */
export const cartographerWorkersDAG: DAGType = CartographerWorkersDag.build();

The dispatcher resolves "cpu" to the bound backend. If "cpu" is not bound, the scatter runs in-process and fires contractWarning. The scatter inbox / work-queue, gather strategies, and outcome reducer are identical in both cases.

Container and dispatcher setup

The browser runner constructs registry-backed worker containers and binds them by role. process-stream uses cpu; the same page also binds io for the summary embedded DAG.

// Separate role bindings use the same worker entry and registry while still
// keeping the DAG's execution boundaries explicit: process-stream runs on
// `cpu`, summarize-insights runs on `io`.
const cpuContainer = new CartographerWorkerContainer({
  'registryModule':  new URL('./cartographerWorkerEntry.ts', import.meta.url).href,
  'registryVersion': '1.0.0',
  'servicesConfig':  { 'useRecordedIp': true },
  'poolSize':        clampedPoolSize.value,
});
const ioContainer = new CartographerWorkerContainer({
  'registryModule':  new URL('./cartographerWorkerEntry.ts', import.meta.url).href,
  'registryVersion': '1.0.0',
  'servicesConfig':  { 'useRecordedIp': true },
  'poolSize':        1,
});

dispatcher = new CartographerBrowserObserver(_cartographerLogger, {
  'containers': {
    'cpu': cpuContainer,
    'io':  ioContainer,
  },
});

Worker registry module

Web workers load a separate module — the main thread's in-memory registry is not accessible across the worker boundary. The registry module exports a RegistryModuleInterface default that reconstructs the bundle and services inside the worker from an opaque servicesConfig JSON object:

const registry: RegistryModuleInterface = {
  async instantiate(servicesConfig: JsonObjectType): Promise<RegistryBundleInterface> {
    const useRecorded = servicesConfig['useRecordedIp'] !== false;
    const services = useRecorded ? GeoResolvers.recorded() : GeoResolvers.live();
    const geoBundle = GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder);
    return {
      'bundle': {
        'nodes': [...geoBundle.nodes, ...cartographerWorkerRuntimeBundle.nodes],
        'dags':  [...geoBundle.dags,  ...cartographerWorkerRuntimeBundle.dags],
      },
      'registryVersion': '1.0.0',
      'restoreState': {
        restore(snapshot: JsonObjectType) {
          return CartographerState.restore(snapshot);
        },
      },
    };
  },
};

export default registry;

Vite chunks the worker entry for the docs site; the runner supplies the module URL to WebWorkerContainer.

Key APIs

Symbol	Import	Role
WebWorkerContainer	@studnicky/dagonizer-executor-web	DagContainerInterface over a browser Web Worker pool
DagContainerInterface	@studnicky/dagonizer/contracts	Adapter contract: runDag(task)
RegistryModuleInterface	@studnicky/dagonizer/contracts	Default export shape loaded by DagHost inside each worker
RegistryBundleInterface	@studnicky/dagonizer/contracts	Return type of instantiate: bundle, services, version, restoreState
DagonizerOptionsType.containers	@studnicky/dagonizer	Binds logical role strings to backend instances

Details for Nerds

Other Node.js backends

@studnicky/dagonizer-executor-node exports WorkerThreadContainer, ForkContainer, ClusterContainer, and SpawnContainer for Node deployments. The repository keeps examples/12-workers.ts as the Node companion for that backend family; the browser runnable for this page is the Cartographer code above.

• container key on a scatter placement. A ScatterNode with a dag body and container: "cpu" delegates each clone's sub-DAG to the bound WebWorkerContainer. Node-body scatter (no dag key in body) is not containable; validation rejects container on a node body.
• containers option. new Dagonizer({ containers: { cpu: workerContainer } }) binds the "cpu" role. Any scatter or embedded-DAG placement declaring container: "cpu" uses this backend.
• Pool lifecycle. WebWorkerContainer manages a pool of workers. Workers initialize on first use (sending init with the registry module URL and services config) and reuse across requests. Call await container.destroy() or await dispatcher.destroy() to shut down the pool cleanly.
• registryVersion handshake. The container sends the registryVersion to each worker during init. The worker's DagHost rejects an init message whose version does not match the string from RegistryModuleInterface.instantiate. This prevents a stale bundle from executing state from a newer bundle's run.
• In-process execution. Remove the container key from the scatter placement (or omit containers from the dispatcher options) to run the scatter in-process. The output is byte-identical; no code changes are needed in node implementations.
• V8 resource limits. WorkerThreadContainerOptions.resourceLimits accepts a maxOldGenerationSizeMb cap to prevent runaway heap growth in individual workers.

Related Concepts

• Guide: Distribution and Cloud - in-fleet containment, registry module contract, pool sizing
• Example 11: Operator Hand-Off - cross-host state pass-over via DAGHandoff channels
• Example 04: Scatter Scout - scatter mechanics: source, body DAG, gather placement, reduce
• Example 05: Embedded DAGs - embedded DAG placements with stateMapping
• Reference: Contracts, DagContainerInterface