# tarungka/wire architecture

> Architecture diagram of the GitHub repository tarungka/wire, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-25). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/tarungka/wire
- Repository: https://github.com/tarungka/wire

## Overview

Wire is a distributed stream-processing engine: users define a job through the Go SDK or submit/manage jobs through the CLI and coordinator API; the coordinator schedules work to workers, which execute stream tasks and emit results. Checkpointing and persistent state support recovery, while RPC and transport connect the cluster. The README also identifies exactly-once processing, event ordering, HA, and observability as product capabilities. The excerpts show coordinator dispatch and worker execution; links for unsampled subsystems are limited to relationships indicated by the documented runtime and source names.

Main entry points: [`internal/worker/worker.go`](https://github.com/tarungka/wire/blob/master/internal/worker/worker.go) (Worker Service), [`cmd/main.go`](https://github.com/tarungka/wire/blob/master/cmd/main.go) (Binary Entrypoint), [`sdk/environment.go`](https://github.com/tarungka/wire/blob/master/sdk/environment.go) (Go SDK), [`internal/coordinator/job_manager.go`](https://github.com/tarungka/wire/blob/master/internal/coordinator/job_manager.go) (Job Management)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_authoring["Job Authoring"]
  node_sdk["Go SDK<br/>[environment.go]"]
  node_yaml["YAML Pipelines<br/>[pipeline_yaml.go]"]
  node_connectors["Connector Adapters"]
end

subgraph group_control["Control Plane"]
  node_cli["CLI Commands<br/>[run.go]"]
  node_entry["Binary Entrypoint<br/>[main.go]"]
  node_api["Coordinator API<br/>[http.go]"]
  node_jobs["Job Management<br/>[job_manager.go]"]
  node_scheduler["Task Scheduling<br/>[scheduler.go]"]
  node_ha["Leader Election<br/>[election.go]"]
end

subgraph group_runtime["Execution Runtime"]
  node_worker["Worker Service<br/>[worker.go]"]
  node_engine["Task Engine<br/>[operator_chain.go]"]
  node_io["Stream I/O<br/>[input_reader.go]"]
end

subgraph group_reliability["State and Recovery"]
  node_metadata[("Coordinator Store<br/>[store.go]")]
  node_checkpoint["Checkpoint Control"]
  node_state["Task State<br/>[state_backend.go]"]
  node_state_db[("Pebble State")]
end

subgraph group_operations["Connectivity and Operations"]
  node_rpc["RPC Services<br/>[server.go]"]
  node_transport["Data Transport<br/>[data_stream.go]"]
  node_observability["Metrics and Telemetry<br/>[otel.go]"]
end

node_user(("Job Author"))
node_source_sink["External Data Systems"]
node_kubernetes["Kubernetes Leases"]

node_user -->|"defines jobs"| node_sdk
node_sdk -->|"parses pipeline"| node_yaml
node_sdk -->|"resolves connectors"| node_connectors
node_user -->|"runs commands"| node_cli
node_cli -->|"invokes binary"| node_entry
node_entry -->|"starts worker"| node_worker
node_entry -->|"starts coordinator"| node_api
node_user -->|"submits jobs"| node_api
node_api -->|"manages jobs"| node_jobs
node_jobs -->|"persists metadata"| node_metadata
node_jobs -->|"wakes scheduler"| node_scheduler
node_scheduler -->|"dispatches tasks"| node_rpc
node_rpc -->|"delivers commands"| node_worker
node_worker -->|"executes tasks"| node_engine
node_engine -->|"reads and writes"| node_io
node_io -->|"exchanges events"| node_source_sink
node_worker -->|"uses data streams"| node_transport
node_transport -->|"delivers task data"| node_worker
node_engine -->|"uses keyed state"| node_state
node_state -->|"persists state"| node_state_db
node_checkpoint -->|"triggers snapshots"| node_rpc
node_checkpoint -->|"stores checkpoint records"| node_metadata
node_worker -->|"reports acknowledgements"| node_checkpoint
node_ha -->|"coordinates leadership"| node_metadata
node_ha -.->|"uses leases"| node_kubernetes
node_entry -->|"initializes telemetry"| node_observability
node_worker -->|"records runtime metrics"| node_observability

click node_sdk "https://github.com/tarungka/wire/blob/master/sdk/environment.go"
click node_yaml "https://github.com/tarungka/wire/blob/master/sdk/pipeline_yaml.go"
click node_connectors "https://github.com/tarungka/wire/tree/master/sdk/connectors"
click node_cli "https://github.com/tarungka/wire/blob/master/internal/jobcli/run.go"
click node_entry "https://github.com/tarungka/wire/blob/master/cmd/main.go"
click node_api "https://github.com/tarungka/wire/blob/master/internal/coordinator/http.go"
click node_jobs "https://github.com/tarungka/wire/blob/master/internal/coordinator/job_manager.go"
click node_scheduler "https://github.com/tarungka/wire/blob/master/internal/coordinator/scheduler.go"
click node_metadata "https://github.com/tarungka/wire/blob/master/internal/coordinator/store.go"
click node_ha "https://github.com/tarungka/wire/blob/master/internal/coordinator/election.go"
click node_worker "https://github.com/tarungka/wire/blob/master/internal/worker/worker.go"
click node_engine "https://github.com/tarungka/wire/blob/master/internal/engine/operator_chain.go"
click node_io "https://github.com/tarungka/wire/blob/master/internal/engine/input_reader.go"
click node_rpc "https://github.com/tarungka/wire/blob/master/internal/rpc/server.go"
click node_transport "https://github.com/tarungka/wire/blob/master/internal/transport/data_stream.go"
click node_checkpoint "https://github.com/tarungka/wire/blob/master/internal/coordinator/checkpoint_manager.go"
click node_state "https://github.com/tarungka/wire/blob/master/internal/engine/state_backend.go"
click node_state_db "https://github.com/tarungka/wire/blob/master/internal/engine/state_backend_pebble.go"
click node_observability "https://github.com/tarungka/wire/blob/master/internal/observability/otel.go"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_sdk,node_yaml,node_connectors toneBlue
class node_cli,node_entry,node_api,node_jobs,node_scheduler,node_ha toneAmber
class node_worker,node_engine,node_io toneMint
class node_metadata,node_checkpoint,node_state,node_state_db toneRose
class node_rpc,node_transport,node_observability,node_user,node_source_sink,node_kubernetes toneIndigo
```

## Components

### Job Authoring

- [Go SDK](https://github.com/tarungka/wire/blob/master/sdk/environment.go): `sdk/environment.go`
- [YAML Pipelines](https://github.com/tarungka/wire/blob/master/sdk/pipeline_yaml.go): `sdk/pipeline_yaml.go`
- [Connector Adapters](https://github.com/tarungka/wire/tree/master/sdk/connectors): `sdk/connectors`

### Control Plane

- [CLI Commands](https://github.com/tarungka/wire/blob/master/internal/jobcli/run.go): `internal/jobcli/run.go`
- [Binary Entrypoint](https://github.com/tarungka/wire/blob/master/cmd/main.go): `cmd/main.go`
- [Coordinator API](https://github.com/tarungka/wire/blob/master/internal/coordinator/http.go): `internal/coordinator/http.go`
- [Job Management](https://github.com/tarungka/wire/blob/master/internal/coordinator/job_manager.go): `internal/coordinator/job_manager.go`
- [Task Scheduling](https://github.com/tarungka/wire/blob/master/internal/coordinator/scheduler.go): `internal/coordinator/scheduler.go`
- [Leader Election](https://github.com/tarungka/wire/blob/master/internal/coordinator/election.go): `internal/coordinator/election.go`

### Execution Runtime

- [Worker Service](https://github.com/tarungka/wire/blob/master/internal/worker/worker.go): `internal/worker/worker.go`
- [Task Engine](https://github.com/tarungka/wire/blob/master/internal/engine/operator_chain.go): `internal/engine/operator_chain.go`
- [Stream I/O](https://github.com/tarungka/wire/blob/master/internal/engine/input_reader.go): `internal/engine/input_reader.go`

### State and Recovery

- [Coordinator Store](https://github.com/tarungka/wire/blob/master/internal/coordinator/store.go): `internal/coordinator/store.go`
- [Checkpoint Control](https://github.com/tarungka/wire/blob/master/internal/coordinator/checkpoint_manager.go): `internal/coordinator/checkpoint_manager.go`
- [Task State](https://github.com/tarungka/wire/blob/master/internal/engine/state_backend.go): `internal/engine/state_backend.go`
- [Pebble State](https://github.com/tarungka/wire/blob/master/internal/engine/state_backend_pebble.go): `internal/engine/state_backend_pebble.go`

### Connectivity and Operations

- [RPC Services](https://github.com/tarungka/wire/blob/master/internal/rpc/server.go): `internal/rpc/server.go`
- [Data Transport](https://github.com/tarungka/wire/blob/master/internal/transport/data_stream.go): `internal/transport/data_stream.go`
- [Metrics and Telemetry](https://github.com/tarungka/wire/blob/master/internal/observability/otel.go): `internal/observability/otel.go`

### Other components

- **Job Author**
- **External Data Systems**
- **Kubernetes Leases**

## Connections

- Job Author → Go SDK: defines jobs
- Go SDK → YAML Pipelines: parses pipeline
- Go SDK → Connector Adapters: resolves connectors
- Job Author → CLI Commands: runs commands
- CLI Commands → Binary Entrypoint: invokes binary
- Binary Entrypoint → Worker Service: starts worker
- Binary Entrypoint → Coordinator API: starts coordinator
- Job Author → Coordinator API: submits jobs
- Coordinator API → Job Management: manages jobs
- Job Management → Coordinator Store: persists metadata
- Job Management → Task Scheduling: wakes scheduler
- Task Scheduling → RPC Services: dispatches tasks
- RPC Services → Worker Service: delivers commands
- Worker Service → Task Engine: executes tasks
- Task Engine → Stream I/O: reads and writes
- Stream I/O → External Data Systems: exchanges events
- Worker Service → Data Transport: uses data streams
- Data Transport → Worker Service: delivers task data
- Task Engine → Task State: uses keyed state
- Task State → Pebble State: persists state
- Checkpoint Control → RPC Services: triggers snapshots
- Checkpoint Control → Coordinator Store: stores checkpoint records
- Worker Service → Checkpoint Control: reports acknowledgements
- Leader Election → Coordinator Store: coordinates leadership
- Leader Election → Kubernetes Leases: uses leases
- Binary Entrypoint → Metrics and Telemetry: initializes telemetry
- Worker Service → Metrics and Telemetry: records runtime metrics