🏛️ ValkyrAI Workflow Engine v2.0 — System Architecture
The N8N Killer: Built on ThorAPI, Hardened for Production
🎯 High-Level Architecture
┌─────────────────────────────────────────────────────────────────┐
│ CLIENT LAYER │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ React Flow │ │ DLQ Browser │ │ Execution │ │
│ │ Studio │ │ & Replay UI │ │ Monitor │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └──────────────────┴──────────────────┘ │
│ │ │
│ RTK Query Hooks │
└────────────────────────────┬────────────────────────────────────┘
│
│ HTTPS/JSON
↓
┌─────────────────────────────────────────────────────────────────┐
│ REST API LAYER │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ WorkflowExecution API │ Run API │ DLQ API │ │
│ │ /WorkflowExecution │ /Run │ /DeadLetterQueue│ │
│ │ - POST /{id}/execute │ - POST /{id}/heartbeat │ │
│ │ - POST /{id}/cancel │ - GET /{id} │ │
│ │ - POST /{id}/pause │ │ │
│ │ - POST /{id}/resume │ /DeadLetterQueue │
│ │ │ - POST /{id}/requeue │
│ │ │ - POST /{id}/discard │
│ └────────┬───────────┬────────┬──────────┬────────────────┘ │
│ │ │ │ │ │
│ Spring Security │ │ Spring MVC │
│ RBAC + ACL │ │ @RestController │
└───────────┼───────────┼────────┼──────────┼─────────────────────┘
│ │ │ │
↓ ↓ ↓ ↓
┌─────────────────────────────────────────────────────────────────┐
│ SERVICE LAYER │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ ValkyrWorkflowExecutionService (Orchestration) │ │
│ │ - execute(workflowId, triggerType, payload) │ │
│ │ - pause/resume/cancel operations │ │
│ │ - state transition management │ │
│ └────────────┬──────────────────────────────────────────────┘ │
│ │ │
│ ├──→ ValkyrRunService (Idempotency Engine) │
│ │ ├─ generateIdempotencyKey() [SHA-256] │
│ │ ├─ acquireLease() [DB row lock] │
│ │ ├─ updateHeartbeat() [extend lease] │
│ │ ├─ completeRun() [store outputs] │
│ │ ├─ failRun() [classify error] │
│ │ └─ reapZombieRuns() [@Scheduled 30s] │
│ │ │
│ ├──→ ValkyrDLQService (Error Recovery) │
│ │ ├─ quarantine() [failed runs] │
│ │ ├─ requeue() [with overrides] │
│ │ └─ discard() [operator notes] │
│ │ │
│ ├──→ ValkyrCircuitBreakerService (Resilience) │
│ │ ├─ isCircuitOpen() │
│ │ ├─ recordFailure() │
│ │ └─ openCircuit() / closeCircuit() │
│ │ │
│ └──→ ValkyrRunnerService (Execution Engine) │
│ ├─ pollAndExecute() [@Scheduled 100ms] │
│ ├─ executeWithHeartbeat() │
│ └─ enforceResourceLimits() │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ ValkyrBudgetService & ValkyrQuotaService │ │
│ │ - checkBudget() [kill switch] │ │
│ │ - checkQuota() [rate limits] │ │
│ │ - trackCost() [per-principal] │ │
│ └──────────────────────────────────────────────────────────┘ │
└───────────────────┬───────────────────────────────────────────┘
│
│ JPA/Hibernate
↓
┌─────────────────────────────────────────────────────────────────┐
│ REPOSITORY LAYER │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ ValkyrRunRepository (Custom Queries) │ │
│ │ - findByIdempotencyKey() [O(1) lookup] │ │
│ │ - findByLeaseUntilBefore() [zombie detection] │ │
│ │ - findByStateOrderByCreated() [FIFO queue] │ │
│ │ - findRunsReadyForRetry() [auto-retry] │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ ValkyrDLQRepository (DLQ Queries) │ │
│ │ - findByResolution() │ │
│ │ - countByResolution() │ │
│ │ - findByFailureType() │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │
│ + WorkflowExecutionRepository, BudgetRepository, │
│ QuotaRepository, CircuitBreakerStateRepository │
└───────────────────┬───────────────────────────────────────────┘
│
│ JDBC
↓
┌─────────────────────────────────────────────── ──────────────────┐
│ DATA LAYER │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ PostgreSQL Database (ACID Transactions) │ │
│ │ │ │
│ │ Tables: │ │
│ │ - workflow_execution (state tracking) │ │
│ │ - run (idempotency + lease) │ │
│ │ - dead_letter_queue (quarantine) │ │
│ │ - circuit_breaker_state (resilience) │ │
│ │ - budget (cost control) │ │
│ │ - quota (rate limiting) │ │
│ │ │ │
│ │ Indexes: │ │
│ │ - idx_run_idempotency (UNIQUE) │ │
│ │ - idx_run_lease_until (zombie reaper) │ │
│ │ - idx_run_state (queue polling) │ │
│ │ - idx_dlq_resolution (operator queries) │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
⚡ Execution Flow: End-to-End
┌──────────┐
│ USER │ POST /WorkflowExecution/{workflowId}/execute
│ CLIENT │ {params: {...}}
└────┬─────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ WorkflowExecutionController │
│ ├─ @PreAuthorize("hasPermission(...)") │
│ └─ workflowExecutionService.execute(...) │
└────┬────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ ValkyrWorkflowExecutionService │
│ ├─ Create WorkflowExecution (PENDING → RUNNING) │
│ ├─ Load Workflow.tasks │
│ └─ For each Task: │
│ ├─ runService.createRun(execution, task, inputs, 1) │
│ └─ Store Run (PENDING state) │
└────┬──────────── ────────────────────────────────────────────┘
│
│ (Async polling loop — RunnerService)
↓
┌─────────────────────────────────────────────────────────────┐
│ ValkyrRunnerService [@Scheduled(fixedDelay = 100ms)] │
│ ├─ runService.findPendingRuns(limit=10) │
│ └─ For each pending Run: │
│ ├─ runService.acquireLease(runId, runnerId) │
│ │ ├─ Check existing lease holder │
│ │ ├─ Set leaseUntil = now + 2min │
│ │ ├─ Set state = LEASED │
│ │ └─ Save Run │
│ │ │
│ ├─ Start heartbeat thread (every 10s): │
│ │ └─ runService.updateHeartbeat(runId, runnerId) │
│ │ ├─ Verify lease ownership │
│ │ ├─ Extend leaseUntil │
│ │ └─ Update heartbeatAt │
│ │ │
│ ├─ circuitBreakerService.isCircuitOpen(module)? │
│ │ ├─ YES → failRun(CIRCUIT_OPEN) │
│ │ └─ NO → continue │
│ │ │
│ ├─ budgetService.checkBudget(principal)? │
│ │ ├─ EXCEEDED → failRun + kill switch │
│ │ └─ OK → continue │
│ │ │
│ ├─ quotaService.checkQuota(principal, resource)? │
│ │ ├─ EXCEEDED → failRun │
│ │ └─ OK → continue │
│ │ │
│ ├─ Execute VModule: │
│ │ ├─ Load ExecModule │
│ │ ├─ module.execute(inputs) → outputs │
│ │ └─ Track duration & cost │
│ │ │
│ ├─ On SUCCESS: │
│ │ └─ runService.completeRun(runId, outputs, ...) │
│ │ ├─ state = SUCCESS │
│ │ ├─ Store outputs, duration, cost │
│ │ └─ Release lease │
│ │ │
│ └─ On FAILURE: │
│ ├─ Classify error type: │
│ │ ├─ TRANSIENT → failRun(TRANSIENT) │
│ │ │ ├─ Calculate backoff (exp + jitter) │
│ │ │ └─ Set retryReadyAt timestamp │
│ │ │ │
│ │ ├─ PERMANENT → failRun(PERMANENT) │
│ │ │ └─ runService.promoteToDeadLetterQueue() │
│ │ │ └─ dlqService.quarantine() │
│ │ │ │
│ │ └─ TIMEOUT → failRun(TIMEOUT) │
│ │ │
│ └─ circuitBreakerService.recordFailure() │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ Background Jobs (Scheduled) │
│ │
│ ValkyrRunService.reapZombieRuns() [@Scheduled 30s] │
│ ├─ Find runs where leaseUntil < now │
│ ├─ AND state IN (LEASED, RUNNING) │
│ ├─ For each zombie: │
│ │ ├─ Log warning (runner X crashed) │
│ │ ├─ state = PENDING │
│ │ ├─ Clear lease fields │
│ │ └─ Save (will be picked up again) │
│ └─ Auto-recovery in < 30 seconds │
│ │
│ CircuitBreakerService.checkHalfOpen() [@Scheduled 60s] │
│ ├─ Find circuits in OPEN state │
│ ├─ If nextRetryAt < now: │
│ │ └─ state = HALF_OPEN (allow 1 test request) │
│ └─ On success → CLOSED, on failure → OPEN │
│ │
│ QuotaService.resetCounters() [@Scheduled hourly/daily] │
│ ├─ Reset hourly counters at hourResetAt │
│ └─ Reset daily counters at dayResetAt │
└─────────────────────────────────────────────────────────────┘
🎯 Key Differentiators vs n8n
| Feature | n8n | ValkyrAI v2.0 |
|---|---|---|
| Idempotency | ❌ None | ✅ SHA-256 keys + DB checks |
| Lease Management | ❌ None | ✅ 2-min leases + heartbeats |
| Crash Recovery | ❌ Lost work | ✅ Zombie reaper (<30s) |
| Exactly-Once | ❌ At-least-once | ✅ Lease + idempotency |
| DLQ & Replay | ❌ Manual | ✅ Operator-guided w/ overrides |
| Circuit Breakers | ❌ None | ✅ Auto-recovery |
| Budget Controls | ❌ None | ✅ Per-principal + kill switch |
| Cost Tracking | ❌ None | ✅ Per-step token tracking |
| Rate Limiting | ⚠️ Basic | ✅ Concurrent + hourly/daily |
| Observability | ⚠️ Logs only | ✅ OpenTelemetry + flame charts |
| Typed I/O | ❌ None | ✅ JSONSchema validation |
| Multi-Tenancy | ⚠️ Weak | ✅ RBAC + ACL + quotas |
🔐 Security Architecture
┌────────────────────────────────────────────────────────┐
│ Authentication Layer │
│ ├─ Spring Security │
│ ├─ JWT tokens (stateless) │
│ └─ Principal (user/system/agent) │
└────┬───────────────────────────────────────────────────┘
│
↓
┌────────────────────────────────────────────────────────┐
│ Authorization Layer (RBAC) │
│ ├─ Roles: ADMIN, USER, SYSTEM, WORKFLOW │
│ ├─ @PreAuthorize("hasRole('ADMIN')") │
│ └─ Method-level security │
└────┬───────────────────────────────────────────────────┘
│
↓
┌────────────────────────────────────────────────────────┐
│ ACL Layer (Fine-Grained) │
│ ├─ Per-object permissions (READ, WRITE, DELETE, ...)│
│ ├─ @PreAuthorize("hasPermission(#id, 'execute')") │
│ └─ ValkyrACL service │
└────┬───────────────────────────────────────────────────┘
│
↓
┌────────────────────────────────────────────────────────┐
│ Data Encryption │
│ ├─ @SecureField annotation │
│ ├─ AES-256 encryption at rest │
│ ├─ Transparent encrypt/decrypt via AspectJ │
│ └─ KMS integration for key rotation │
└────────────────────────────────────────────────────────┘
📊 Data Flow: Idempotency in Action
Request: Execute Task with inputs {userId: "123", action: "purchase"}
Step 1: Generate Idempotency Key
─────────────────────────────────
Input: executionId + taskId + moduleId + SHA256({userId:"123",action:"purchase"})
Output: "a3f5c9d8e2b1..." (64-char hex)
Step 2: Check Existing Run
─────────────────────────────────
Query: SELECT * FROM run WHERE idempotency_key = 'a3f5c9d8e2b1...'
IF found AND state = SUCCESS:
✅ Return cached outputs (no duplicate work)
Exit
IF found AND state IN (RUNNING, LEASED):
⚠️ Duplicate request detected, return existing run
Exit
ELSE:
➡️ Continue to Step 3
Step 3: Create New Run
─────────────────────────────────
INSERT INTO run (
id, execution_id, task_id, state, idempotency_key, ...
) VALUES (
uuid(), execution_id, task_id, 'PENDING', 'a3f5c9d8e2b1...', ...
)
Commit transaction
Step 4: Runner Picks Up Run
─────────────────────────────────
SELECT * FROM run WHERE state = 'PENDING' ORDER BY created_date LIMIT 10
FOR EACH run:
BEGIN TRANSACTION
UPDATE run
SET state = 'LEASED',
lease_until = NOW() + INTERVAL '2 minutes',
leased_by = 'runner-pod-123'
WHERE id = run.id
AND (lease_until IS NULL OR lease_until < NOW())
COMMIT
IF rows_affected = 1:
✅ Lease acquired
ELSE:
❌ Another runner got it, skip
Step 5: Execute with Heartbeat
─────────────────────────────────
Start heartbeat thread (every 10s):
UPDATE run
SET heartbeat_at = NOW(),
lease_until = NOW() + INTERVAL '2 minutes'
WHERE id = run.id AND leased_by = 'runner-pod-123'
Execute module.execute(inputs)
On success:
UPDATE run
SET state = 'SUCCESS',
finished_at = NOW(),
outputs = '{"orderId":"abc123"}',
duration_ms = 1234,
cost_tokens = 0.05,
lease_until = NULL,
leased_by = NULL
WHERE id = run.id
Step 6: Future Request (Same Inputs)
─────────────────────────────────
Same idempotency key: 'a3f5c9d8e2b1...'
Query: SELECT * FROM run WHERE idempotency_key = 'a3f5c9d8e2b1...'
Result: state = 'SUCCESS', outputs = '{"orderId":"abc123"}'
✅ Return cached outputs immediately (no re-execution)
🚀 Scalability Architecture
┌─────────────────────────────────────────────────────────┐
│ Load Balancer (ALB/nginx) │
│ ├─ Route to available API pods │
│ └─ Health checks: /actuator/health │
└────┬────────────────────────────────────────────────────┘
│
├──────────────┬──────────────┬──────────────┐
↓ ↓ ↓ ↓
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ API Pod 1│ │ API Pod 2│ │ API Pod 3│ │ API Pod N│
│ (Stateless) │ (Stateless) │ (Stateless) │ (Stateless)
└──────────┘ └──────────┘ └──────────┘ └──────────┘
│ │ │ │
└──────────────┴──────────────┴──────────────┘
│
↓
┌──────────────────────────────────────────┐
│ Shared Database (PostgreSQL) │
│ ├─ Connection pooling (HikariCP) │
│ ├─ Read replicas for queries │
│ └─ Write leader for mutations │
└────────────── ────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│ Runner Pods (Scheduled Workers) │
│ ├─ Auto-scaling: scale by queue depth │
│ ├─ Each pod polls queue independently │
│ ├─ Lease-based coordination (no leader election) │
│ └─ Horizontal: 1 → N pods (linear scale) │
│ │
│ Runner Pod 1 (runnerId: runner-abc-123) │
│ └─ @Scheduled pollAndExecute() every 100ms │
│ ├─ Fetch 10 pending runs │
│ ├─ Acquire leases (DB row lock) │
│ ├─ Execute with heartbeat │
│ └─ Release on completion │
│ │
│ Runner Pod 2 (runnerId: runner-def-456) │
│ └─ (same) │
│ │
│ Runner Pod N (runnerId: runner-xyz-789) │
│ └─ (same) │
└─────────────────────────────────────────────────────────┘
Performance Targets:
- 10k concurrent executions
- 250k active tasks
- P95 dispatch < 75ms
- Runner CPU < 70% at scale
🎯 Summary
ValkyrAI Workflow Engine v2.0 is architected for production resilience with:
- Exactly-once execution via idempotency keys + lease management
- Automatic crash recovery via zombie reaper (<30s)
- Operator-friendly DLQ with replay + input overrides
- Cost control via budgets + quotas + kill switches
- Horizontal scaling via stateless design + DB coordination
- Security by default via RBAC + ACL + field encryption
This is the n8n killer. Built on ThorAPI. Shipping Q4 2025. 🚀