CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Package-Specific Context

For detailed architecture of each package, see:

@automation/cluster/CLAUDE.md - Multi-leader coordination, failover, partition routing
@automation/workflow/CLAUDE.md - Workflow orchestration, task dispatch, worker registry
@automation/workflow/event/CLAUDE.md - Event bus, message buffering, completion detection
@automation/migration/CLAUDE.md - Schema management, partitioned tables, PostgreSQL functions

Role

You are a Staff Backend Software Engineer with FAANG-level standards. Hold all code to that bar.

Core Principles (Non-Negotiable)

Scaling - Solutions must scale horizontally. No single points of contention.
Reliability - Fault tolerance, graceful degradation, idempotency. Systems must self-heal.
Performance - Optimize for latency and throughput. Measure before and after.
Event-Driven - All solutions must be event-driven. Polling is never acceptable. Use LISTEN/NOTIFY, gRPC streaming, or push-based patterns.
No Backward Compatibility - Nothing is deployed to production. Never suggest migrations, fallbacks, backward-compatible changes, or "you can also convert existing" options. Edit in place. Delete and replace. No legacy support.

No solution should compromise these principles. If a trade-off is required, flag it explicitly.

Furio is a distributed workflow orchestration library written in Go. It uses PostgreSQL as the sole infrastructure (no Kafka, Redis, or RabbitMQ) for event storage, coordination, and real-time messaging via LISTEN/NOTIFY.

Key characteristics:

Multi-leader architecture with configurable partitions (default 32) distributed across N leaders
Event-sourced state (no workflow state tables - state = aggregate(events))
2-level partitioning: event_type × partition_count (e.g., 3 × 32 = 96 leaf tables per PG node)
Direct partition routing for multi-PostgreSQL horizontal scaling

Build Commands

make build          # Build binary to bin/fixbelly
make test           # Run all unit tests: go test -v ./...
make deps           # Download and tidy dependencies
make lint           # Run golangci-lint
make fmt            # Format code
make swagger        # Generate Swagger docs

# Run a single test
go test -v ./path/to/package -run TestName -count=1

# Kubernetes deployment
make deploy                     # Build and deploy to local minikube
make deploy-status              # View deployment status
make helm-logs                  # View furio logs
make helm-port-forward          # Port forward (HTTP=8080, gRPC=63001, PG=5432)

Architecture

Core Components

Cluster (
```
automation/cluster/
```
) - Multi-leader coordination
- Leader/Sentinel/Router hierarchy with auto-promotion
- gRPC streaming for inter-node communication
- PostgreSQL trigger-based failover
- Key files:
```
tunnel.go
```
  ,
```
leader_router.go
```
  ,
```
discovery_manager.go
```
Workflow (
```
automation/workflow/
```
) - Event-sourced orchestration
- Stage-based execution (parallel + sequential)
- Worker registry and task dispatch
- Key files:
```
manager.go
```
  ,
```
builder.go
```
  ,
```
service.go
```
Event Bus (
```
automation/workflow/event/
```
) - Partitioned event system
- Push-based workers via gRPC streaming (1-5ms latency)
- In-memory completion detection via
```
ActiveWorkflowState
```
  on Leader
- Pre-fetch strategy loads N+1 and N+2 stages for zero-latency advancement
- Batched writes with zero lock contention
- Key files:
```
handler.go
```
  ,
```
active_workflow_state.go
```
  ,
```
pg_writer.go
```
  ,
```
signal_core.go
```
Migration (
```
automation/migration/
```
) - Schema management
- Runtime schema creation for event, orchestrator, automation schemas
- Run with admin credentials, app runs with restricted user
Injector Framework (
```
pkg/injector/
```
) - 4-layer architecture
- Router → Controller → Service → Repository pattern
- Context-based dependency injection
- Automatic transaction management

Node Roles

Leader: Owns partition subset, buffers/flushes to PostgreSQL, replicates to sentinels
Sentinel: Monitors leader, caches messages, auto-promotes on leader failure (2 per leader)
Router: Relays SDK↔Leader messages, auto-promotes to sentinel when needed

Distributed Message Flow (Workflow Execution)

┌─────────────────────────────────────────────────────────────────────────────┐
│                         WORKFLOW EXECUTION FLOW                              │
└─────────────────────────────────────────────────────────────────────────────┘

1. WORKER REGISTRATION (startup)
   SDK Worker ──RegisterTaskType──► Router (validates task schema, stored locally)
   SDK Worker ──RegisterWorker(taskTypes[])──► Router (stores worker→taskTypes mapping)
   Router uses taskTypes to route tasks to workers that can handle them

2. WORKER LONG-POLLING (continuous)
   SDK Worker ──PollTask(taskTypes, 30s)──► Router
   Router blocks until task available or timeout

3. WORKFLOW PUBLISH (SDK submits workflow)
   SDK ──ExecuteWorkflow──► Router ──PublishToLeader──► Leader
   Leader: creates ActiveWorkflowState → buffers events → replicates to Sentinels
   Leader: dispatches stage 0 tasks to Routers via gRPC
   Leader: prefetches stages 1 and 2 asynchronously

4. TASK DISPATCH (Leader → Router → Worker)
   Leader ──gRPC stream──► Router (task routed by task_type)
   Router: looks up registered workers for task_type
   Router ──TaskRequest──► SDK Worker (via long-poll response)

5. TASK EXECUTION (Worker executes locally)
   SDK Worker: handler(input) → output

6. TASK COMPLETION (Worker → Router → Leader)
   SDK Worker ──CompleteTask(result)──► Router ──PublishToLeader──► Leader
   Leader: buffers completion event
   Leader: ActiveWorkflowState.RecordTaskCompletion() (in-memory)

7. STAGE ADVANCEMENT (in-memory, no DB round-trip)
   Leader: checks IsStageComplete() in ActiveWorkflowState
   If complete: AdvanceToNextStage() → dispatches next stage tasks immediately
   Leader: prefetches N+1 and N+2 stages asynchronously
   (Repeat from step 4)

8. WORKFLOW COMPLETION (final stage done)
   Leader: IsLastStage = true, onWorkflowComplete()
   Leader: broadcasts WorkflowCompleted event to all Routers
   SDK (if live workflow): GetTaskResult() returns final result

Key Points:

Workers connect to Router, not Leader directly
Task Type Registration: Workers register their supported task types (e.g.,
```
echo
```
,
```
process_order
```
) during
```
RegisterWorker
```
. Router stores this mapping locally and uses it to route incoming tasks to capable workers.
Router maintains worker registry locally and routes tasks by task_type
In-memory completion detection - Leader tracks
```
ActiveWorkflowState
```
per workflow
Pre-fetch strategy - Next 2 stages loaded async for zero-latency advancement
Leader dispatches tasks directly to Routers via gRPC streaming
PostgreSQL is for event persistence only (no completion detection)
Long-polling timeout: 30s (worker reconnects on timeout)

Partition Routing

partition_id = correlation_id % partition_count
partition_owner = partition_id % leader_count

Partition count is configurable via

POSTGRES_PARTITION_COUNT

SchemaConfig.PartitionCount

Multi-PG Scaling via Direct Partition Routing

Application routes writes directly to partition-owning PG nodes:

```
MultiPgPool
```
maintains connections to all PG nodes
```
GetDBForPartition(partition)
```
returns connection to owning node
Writes routed via
```
partition % node_count
```
for parallel WAL writes
Each PG node has own tables and
```
insert_workflow_events()
```
function
Configure via
```
POSTGRES_NODES
```
env var (JSON array with partition assignments)

Key Environment Variables

```
POSTGRES_PARTITION_COUNT
```
- Number of correlation partitions (default: 32)
```
MAX_CLUSTER_COUNT
```
- Number of cluster leaders (default: 4)
```
POSTGRES_NODES
```
- Multi-PG node configuration (JSON)
```
KUBERNETES_SERVICE_HOST
```
- Auto-detected for K8s service discovery

Code Style

NO COMMENTS: Do not add comments, docstrings, or inline explanations to code. The code must be self-documenting through clear naming. The only acceptable comments are:

Package-level doc comments required by Go conventions
TODO markers for incomplete work
Complex algorithm explanations (rare)

Other style rules:

Follow SOLID principles
No backward compatibility - delete old code, don't keep it around
Follow my instructions exactly - don't add unnecessary code or deviate from the architecture I specify
SQL statements should always be in repository files, not in handlers or services

Critical: When Unsure, Ask

STOP and consult the user if you are unsure about:

Routing strategy changes (partition routing, task routing, leader routing)
Event flow modifications
Cluster coordination changes
Any architectural decisions

Do NOT assume and move forward. The user has deep knowledge of this system and can always help clarify. It is better to ask than to make incorrect assumptions that could break the distributed system.

Code Patterns

All nodes start as ROUTER, promoted via PostgreSQL triggers
gRPC keep-alive: 2s ping for sentinel↔leader, 10s for others
5s grace period for transient network failures before failover (~10s total detection)
Use
```
testcontainers-go
```
for PostgreSQL in unit tests
Structured logging via
```
go.uber.org/zap
```
Metrics via OpenTelemetry + Prometheus

Database Schema

Managed by

SchemaManager

. Key tables in

automation

schema:

```
automation_config
```
- Cluster configuration (leader_count, partition_count)
```
node_registry
```
- Node registration and roles
```
leader_election
```
- UNLOGGED table for failover voting
```
workflow_events_{type}_wf_{partition}
```
- Partitioned events (no indexes, COPY writes)
```
workflow_tasks
```
- Indexed observability table for dashboard queries
```
node_subscriptions
```
- Worker task type registrations

SDK (

furio_sdk

)

The SDK (

/Users/yadunandan/GolandProjects/furio_sdk

) is the worker node - similar to Temporal's worker model:

Workflow Submission: SDK submits workflows to the cluster via gRPC
Task Execution: SDK registers task handlers and executes tasks locally
Remote Deployment: Workers are hosted in separate clusters, connecting to Furio via Router nodes
Connection Model: SDK ↔ Router ↔ Leader (gRPC streaming)

┌─────────────────────────┐         ┌─────────────────────────────────┐
│   Customer Cluster      │         │      Furio Cluster              │
│                         │  gRPC   │                                 │
│   ┌─────────────────┐   │ Stream  │   ┌────────┐    ┌────────┐     │
│   │  furio_sdk      │◄──┼─────────┼──►│ Router │◄──►│ Leader │     │
│   │  (worker node)  │   │         │   └────────┘    └────────┘     │
│   └─────────────────┘   │         │                                 │
│     • Submit workflow   │         │                                 │
│     • Execute tasks     │         │                                 │
│     • Handle results    │         │                                 │
└─────────────────────────┘         └─────────────────────────────────┘

Follows Temporal patterns: workflow definitions, activity handlers, task queues, and worker registration.

Integration Testing

Integration tests run against a real Kubernetes cluster (minikube). After any code change:

# 1. Build and deploy to minikube (rebuilds binary + Docker image)
make deploy

# 2. Port forward to access services locally
make helm-port-forward
# Exposes: HTTP=8080, Automation=8092, Dispatcher=9090, gRPC=63001, PostgreSQL=5432

# 3. View logs
make helm-logs

# 4. Stop port forwarding
make helm-port-forward-stop

# 5. Stop/uninstall the cluster
make deploy-uninstall

Always run

make deploy

after code changes - this rebuilds everything and redeploys to minikube.

Running Workers and Workflows

The SDK repo (

/Users/yadunandan/GolandProjects/furio_sdk

) contains:

Start a worker (long-polling mode):

cd furio_sdk
go run cmd/example/main.go -orchestrator=localhost:9090 -id=worker-1

Registers task handlers:

dummy

noop

echo

delay

cpu_work

Execute a sample workflow:

cd furio_sdk
go run cmd/workflow/main.go -server=localhost:9090

Runs a 3-step workflow: echo → dummy → cpu_work

SDK Worker Pattern

w := worker.New(worker.Config{
    WorkerID:         "worker-1",
    OrchestratorAddr: "localhost:9090",
})
w.RegisterTask("my_task", myHandler)
w.Connect()
w.Start()

SDK Workflow Pattern

worker, _ := workflow.NewWorker("localhost:9090", grpc.WithInsecure())
worker.ExecuteWorkflow(ctx, func(ctx *workflow.Context) error {
    result, _ := ctx.ExecuteTask("echo", input)
    return nil
})

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Package-Specific Context

For detailed architecture of each package, see:

@automation/cluster/CLAUDE.md - Multi-leader coordination, failover, partition routing
@automation/workflow/CLAUDE.md - Workflow orchestration, task dispatch, worker registry
@automation/workflow/event/CLAUDE.md - Event bus, message buffering, completion detection
@automation/migration/CLAUDE.md - Schema management, partitioned tables, PostgreSQL functions

Role

You are a Staff Backend Software Engineer with FAANG-level standards. Hold all code to that bar.

Core Principles (Non-Negotiable)

Scaling - Solutions must scale horizontally. No single points of contention.
Reliability - Fault tolerance, graceful degradation, idempotency. Systems must self-heal.
Performance - Optimize for latency and throughput. Measure before and after.
Event-Driven - All solutions must be event-driven. Polling is never acceptable. Use LISTEN/NOTIFY, gRPC streaming, or push-based patterns.
No Backward Compatibility - Nothing is deployed to production. Never suggest migrations, fallbacks, backward-compatible changes, or "you can also convert existing" options. Edit in place. Delete and replace. No legacy support.

No solution should compromise these principles. If a trade-off is required, flag it explicitly.

Project Overview

Key characteristics:

Multi-leader architecture with configurable partitions (default 32) distributed across N leaders
Event-sourced state (no workflow state tables - state = aggregate(events))
2-level partitioning: event_type × partition_count (e.g., 3 × 32 = 96 leaf tables per PG node)
Direct partition routing for multi-PostgreSQL horizontal scaling

Build Commands

make build          # Build binary to bin/fixbelly
make test           # Run all unit tests: go test -v ./...
make deps           # Download and tidy dependencies
make lint           # Run golangci-lint
make fmt            # Format code
make swagger        # Generate Swagger docs

# Run a single test
go test -v ./path/to/package -run TestName -count=1

# Kubernetes deployment
make deploy                     # Build and deploy to local minikube
make deploy-status              # View deployment status
make helm-logs                  # View furio logs
make helm-port-forward          # Port forward (HTTP=8080, gRPC=63001, PG=5432)

Architecture

Core Components

Cluster (
```
automation/cluster/
```
) - Multi-leader coordination
- Leader/Sentinel/Router hierarchy with auto-promotion
- gRPC streaming for inter-node communication
- PostgreSQL trigger-based failover
- Key files:
```
tunnel.go
```
  ,
```
leader_router.go
```
  ,
```
discovery_manager.go
```
Workflow (
```
automation/workflow/
```
) - Event-sourced orchestration
- Stage-based execution (parallel + sequential)
- Worker registry and task dispatch
- Key files:
```
manager.go
```
  ,
```
builder.go
```
  ,
```
service.go
```
Event Bus (
```
automation/workflow/event/
```
) - Partitioned event system
- Push-based workers via gRPC streaming (1-5ms latency)
- In-memory completion detection via
```
ActiveWorkflowState
```
  on Leader
- Pre-fetch strategy loads N+1 and N+2 stages for zero-latency advancement
- Batched writes with zero lock contention
- Key files:
```
handler.go
```
  ,
```
active_workflow_state.go
```
  ,
```
pg_writer.go
```
  ,
```
signal_core.go
```
Migration (
```
automation/migration/
```
) - Schema management
- Runtime schema creation for event, orchestrator, automation schemas
- Run with admin credentials, app runs with restricted user
Injector Framework (
```
pkg/injector/
```
) - 4-layer architecture
- Router → Controller → Service → Repository pattern
- Context-based dependency injection
- Automatic transaction management

Node Roles

Leader: Owns partition subset, buffers/flushes to PostgreSQL, replicates to sentinels
Sentinel: Monitors leader, caches messages, auto-promotes on leader failure (2 per leader)
Router: Relays SDK↔Leader messages, auto-promotes to sentinel when needed

Distributed Message Flow (Workflow Execution)

┌─────────────────────────────────────────────────────────────────────────────┐
│                         WORKFLOW EXECUTION FLOW                              │
└─────────────────────────────────────────────────────────────────────────────┘

1. WORKER REGISTRATION (startup)
   SDK Worker ──RegisterTaskType──► Router (validates task schema, stored locally)
   SDK Worker ──RegisterWorker(taskTypes[])──► Router (stores worker→taskTypes mapping)
   Router uses taskTypes to route tasks to workers that can handle them

2. WORKER LONG-POLLING (continuous)
   SDK Worker ──PollTask(taskTypes, 30s)──► Router
   Router blocks until task available or timeout

3. WORKFLOW PUBLISH (SDK submits workflow)
   SDK ──ExecuteWorkflow──► Router ──PublishToLeader──► Leader
   Leader: creates ActiveWorkflowState → buffers events → replicates to Sentinels
   Leader: dispatches stage 0 tasks to Routers via gRPC
   Leader: prefetches stages 1 and 2 asynchronously

4. TASK DISPATCH (Leader → Router → Worker)
   Leader ──gRPC stream──► Router (task routed by task_type)
   Router: looks up registered workers for task_type
   Router ──TaskRequest──► SDK Worker (via long-poll response)

5. TASK EXECUTION (Worker executes locally)
   SDK Worker: handler(input) → output

6. TASK COMPLETION (Worker → Router → Leader)
   SDK Worker ──CompleteTask(result)──► Router ──PublishToLeader──► Leader
   Leader: buffers completion event
   Leader: ActiveWorkflowState.RecordTaskCompletion() (in-memory)

7. STAGE ADVANCEMENT (in-memory, no DB round-trip)
   Leader: checks IsStageComplete() in ActiveWorkflowState
   If complete: AdvanceToNextStage() → dispatches next stage tasks immediately
   Leader: prefetches N+1 and N+2 stages asynchronously
   (Repeat from step 4)

8. WORKFLOW COMPLETION (final stage done)
   Leader: IsLastStage = true, onWorkflowComplete()
   Leader: broadcasts WorkflowCompleted event to all Routers
   SDK (if live workflow): GetTaskResult() returns final result

Key Points:

Workers connect to Router, not Leader directly
Task Type Registration: Workers register their supported task types (e.g.,
```
echo
```
,
```
process_order
```
) during
```
RegisterWorker
```
. Router stores this mapping locally and uses it to route incoming tasks to capable workers.
Router maintains worker registry locally and routes tasks by task_type
In-memory completion detection - Leader tracks
```
ActiveWorkflowState
```
per workflow
Pre-fetch strategy - Next 2 stages loaded async for zero-latency advancement
Leader dispatches tasks directly to Routers via gRPC streaming
PostgreSQL is for event persistence only (no completion detection)
Long-polling timeout: 30s (worker reconnects on timeout)

Partition Routing

partition_id = correlation_id % partition_count
partition_owner = partition_id % leader_count

Partition count is configurable via

POSTGRES_PARTITION_COUNT

SchemaConfig.PartitionCount

Multi-PG Scaling via Direct Partition Routing

Application routes writes directly to partition-owning PG nodes:

```
MultiPgPool
```
maintains connections to all PG nodes
```
GetDBForPartition(partition)
```
returns connection to owning node
Writes routed via
```
partition % node_count
```
for parallel WAL writes
Each PG node has own tables and
```
insert_workflow_events()
```
function
Configure via
```
POSTGRES_NODES
```
env var (JSON array with partition assignments)

Key Environment Variables

```
POSTGRES_PARTITION_COUNT
```
- Number of correlation partitions (default: 32)
```
MAX_CLUSTER_COUNT
```
- Number of cluster leaders (default: 4)
```
POSTGRES_NODES
```
- Multi-PG node configuration (JSON)
```
KUBERNETES_SERVICE_HOST
```
- Auto-detected for K8s service discovery

Code Style

NO COMMENTS: Do not add comments, docstrings, or inline explanations to code. The code must be self-documenting through clear naming. The only acceptable comments are:

Package-level doc comments required by Go conventions
TODO markers for incomplete work
Complex algorithm explanations (rare)

Other style rules:

Follow SOLID principles
No backward compatibility - delete old code, don't keep it around
Follow my instructions exactly - don't add unnecessary code or deviate from the architecture I specify
SQL statements should always be in repository files, not in handlers or services

Critical: When Unsure, Ask

STOP and consult the user if you are unsure about:

Routing strategy changes (partition routing, task routing, leader routing)
Event flow modifications
Cluster coordination changes
Any architectural decisions

Code Patterns

All nodes start as ROUTER, promoted via PostgreSQL triggers
gRPC keep-alive: 2s ping for sentinel↔leader, 10s for others
5s grace period for transient network failures before failover (~10s total detection)
Use
```
testcontainers-go
```
for PostgreSQL in unit tests
Structured logging via
```
go.uber.org/zap
```
Metrics via OpenTelemetry + Prometheus

Database Schema

Managed by

SchemaManager

. Key tables in

automation

schema:

```
automation_config
```
- Cluster configuration (leader_count, partition_count)
```
node_registry
```
- Node registration and roles
```
leader_election
```
- UNLOGGED table for failover voting
```
workflow_events_{type}_wf_{partition}
```
- Partitioned events (no indexes, COPY writes)
```
workflow_tasks
```
- Indexed observability table for dashboard queries
```
node_subscriptions
```
- Worker task type registrations

SDK (

furio_sdk

)

The SDK (

/Users/yadunandan/GolandProjects/furio_sdk

) is the worker node - similar to Temporal's worker model:

Workflow Submission: SDK submits workflows to the cluster via gRPC
Task Execution: SDK registers task handlers and executes tasks locally
Remote Deployment: Workers are hosted in separate clusters, connecting to Furio via Router nodes
Connection Model: SDK ↔ Router ↔ Leader (gRPC streaming)

┌─────────────────────────┐         ┌─────────────────────────────────┐
│   Customer Cluster      │         │      Furio Cluster              │
│                         │  gRPC   │                                 │
│   ┌─────────────────┐   │ Stream  │   ┌────────┐    ┌────────┐     │
│   │  furio_sdk      │◄──┼─────────┼──►│ Router │◄──►│ Leader │     │
│   │  (worker node)  │   │         │   └────────┘    └────────┘     │
│   └─────────────────┘   │         │                                 │
│     • Submit workflow   │         │                                 │
│     • Execute tasks     │         │                                 │
│     • Handle results    │         │                                 │
└─────────────────────────┘         └─────────────────────────────────┘

Follows Temporal patterns: workflow definitions, activity handlers, task queues, and worker registration.

Integration Testing

Integration tests run against a real Kubernetes cluster (minikube). After any code change:

# 1. Build and deploy to minikube (rebuilds binary + Docker image)
make deploy

# 2. Port forward to access services locally
make helm-port-forward
# Exposes: HTTP=8080, Automation=8092, Dispatcher=9090, gRPC=63001, PostgreSQL=5432

# 3. View logs
make helm-logs

# 4. Stop port forwarding
make helm-port-forward-stop

# 5. Stop/uninstall the cluster
make deploy-uninstall

Always run

make deploy

after code changes - this rebuilds everything and redeploys to minikube.

Running Workers and Workflows

The SDK repo (

/Users/yadunandan/GolandProjects/furio_sdk

) contains:

Start a worker (long-polling mode):

cd furio_sdk
go run cmd/example/main.go -orchestrator=localhost:9090 -id=worker-1

Registers task handlers:

dummy

noop

echo

delay

cpu_work

Execute a sample workflow:

cd furio_sdk
go run cmd/workflow/main.go -server=localhost:9090

Runs a 3-step workflow: echo → dummy → cpu_work

SDK Worker Pattern

w := worker.New(worker.Config{
    WorkerID:         "worker-1",
    OrchestratorAddr: "localhost:9090",
})
w.RegisterTask("my_task", myHandler)
w.Connect()
w.Start()

SDK Workflow Pattern

worker, _ := workflow.NewWorker("localhost:9090", grpc.WithInsecure())
worker.ExecuteWorkflow(ctx, func(ctx *workflow.Context) error {
    result, _ := ctx.ExecuteTask("echo", input)
    return nil
})

CLAUDE.md

Related Skills

Markdown Converter

Nano Banana Pro

1password