Ditch Glue Code: Skill‑Based Distributed Engine for Multi‑API Orchestration and Real‑Time DB Checks

Problem Background: Why Traditional API Automation Gets Heavier

In monolithic apps, automation only needs three steps: obtain a token, call a business API, and assert the HTTP status and fields. In microservice and cloud‑native environments a "order success" flow traverses many services, databases, and asynchronous messages, making simple sequential scripts insufficient.

Four Pain Points of Hand‑Written Scripts

Scenario reuse is poor because calls, SQL assertions, and context variables are scattered across scripts.

Maintenance cost spikes when service addresses, auth methods, or DB schemas change, requiring dozens of scripts to be updated.

Concurrent execution quickly exhausts HTTP and DB connection pools during large CI runs.

Lack of engineering governance: scripts ignore timeout/retry policies, side‑effects, asynchronous DB state, worker crash recovery, and failure pinpointing.

Thus the real abstraction target is not the script but the "automation execution capability" itself.

Business Case: E‑commerce Order Chain Verification

A typical promotion‑time order flow involves user, inventory, order, payment services, a message bus, and three databases (order_db, inventory_db, payment_db). The platform must support:

Multi‑API chaining

Automatic context variable propagation

Real‑time DB validation

Asynchronous message verification

Delayed consistency polling

Automatic retries with result tracing

High‑concurrency batch execution

Core Idea: Skill + DAG + Execution Engine

The solution replaces hand‑written glue with a declarative model:

AutomationScenario = Skill + DAG + DistributedExecutionEngine

1. What Is a Skill?

A Skill is the smallest reusable unit, e.g.: http-call: invoke an HTTP endpoint db-check: run SQL and assert results mq-produce: send a message mq-consume: consume and verify a message wait-until: poll until a condition is met script-transform: perform complex data transformations

A Skill defines its input parameters, output structure, side‑effects, timeout/retry policy, and how it writes to the execution context.

2. Why Use a DAG Instead of Linear Scripts?

Real business flows contain parallel branches. For example, after order creation the system can simultaneously verify the order DB and the order event, or pre‑reserve inventory while checking the inventory table. A DAG naturally captures these non‑linear dependencies.

3. Why a Unified Execution Engine?

Without a central engine each script must handle dependency resolution, concurrency scheduling, context propagation, timeout/retry, idempotency, and logging. Centralizing these cross‑cutting concerns makes the platform governable.

Skill Model Design: From "Runnable" to "Governable"

Skill Contract

public interface Skill {
    String type();
    SkillResult execute(SkillExecutionContext context) throws Exception;
    default boolean isIdempotent() { return true; }
}

The contract requires:

Parameter definition

Output definition

Side‑effect declaration

Timeout/retry strategy

Context writing rules

Result exposure for reporting

Execution Context

The context is more than a Map<String, Object>. It must store global variables (e.g., userId, sku, traceId), node outputs (e.g., create_order.orderId), runtime metadata (retry count, start time), and audit information (request/response payloads).

public class SkillExecutionContext {
    private final String taskId;
    private final String traceId;
    private final SkillNode currentNode;
    private final ExecutionSnapshot snapshot;
    private final ConcurrentMap<String, Object> variables;
    private final ConcurrentMap<String, NodeExecutionRecord> records;
    // resolveTemplate, putOutput, etc.
}

Two‑level variable writing is recommended: nodeId.key to avoid parallel overwrites, and a plain key for simple references.

SkillResult Design

public record SkillResult(
    boolean success,
    SkillStatus status,
    Map<String, Object> outputs,
    List<AssertionResult> assertions,
    RetryDirective retryDirective,
    String errorCode,
    String errorMessage,
    long costMs
) { }

The result carries status (SUCCESS, FAILED, SKIPPED, TIMEOUT), detailed assertions for reporting, and optional retry directives.

Orchestration File Design: Readable, Auditable, Versionable

YAML (or JSON) is chosen because it works natively with Git version control, CI change‑review, visual editors, and engine parsing.

dag:
  dagId: order-create-and-verify
  version: 1.0.0
  failureStrategy: FAIL_FAST
  timeoutMs: 30000
  context:
    userId: U10001
    sku: SKU-IPHONE-001
    quantity: 1
  nodes:
    - id: get_user
      type: http-call
      config:
        service: user-service
        method: GET
        path: /api/users/${userId}
        timeoutMs: 1000
        extract:
          memberLevel: $.data.memberLevel
          userName: $.data.name
    - id: reserve_stock
      type: http-call
      dependsOn: [get_user]
      config:
        service: inventory-service
        method: POST
        path: /api/stocks/reservations
        body:
          sku: ${sku}
          quantity: ${quantity}
          userId: ${userId}
        timeoutMs: 1500
        retry:
          maxAttempts: 2
          backoffMs: 200
        extract:
          reservationId: $.data.reservationId
    # ... other nodes omitted for brevity ...

Key points:

Dependencies are expressed with dependsOn.

Template variables use ${var} syntax.

All node types (http‑call, db‑check, mq‑consume, wait‑until) share the same Skill contract.

Each node can configure its own timeout and retry policy.

Overall Architecture: From Request Submission to Distributed Execution

The platform separates "definition" from "runtime execution":

Automation API : receives task submissions, validates DAG, generates taskId / traceId, persists the task, and pushes it to a message queue.

Task Dispatcher : decouples submission from execution, preventing burst overload.

Worker : stateless process that executes the DAG, using dedicated thread pools per node type.

Redis Snapshot : stores task snapshots, node states, and context for crash recovery.

MySQL : persists task metadata, node execution records, and assertion details for reporting.

Kafka (or other MQ) : transports tasks to workers; optional KEDA‑based autoscaling based on consumer lag.

Metrics & Tracing : Prometheus counters and Jaeger/Tempo trace IDs for observability.

Component Responsibilities

Automation API : validate, generate IDs, persist, enqueue.

Task Dispatcher : rate‑limit and fan‑out tasks.

Worker : load snapshot, build execution graph, schedule runnable nodes, merge outputs, handle failures.

Redis Snapshot : keep in‑memory state for fast recovery.

MySQL : store DAG versions, task logs, node logs, assertion logs.

Execution Engine Mechanics

1. Execution Flow

The engine repeatedly:

Finds nodes whose dependencies are satisfied (using indegree counters).

Selects an appropriate executor ( httpExecutor or dbExecutor) based on node type.

Runs nodes concurrently, records SkillResult, merges outputs into the context, and persists snapshots.

Stops early if FAIL_FAST is configured and a node fails.

2. DAG Scheduling Core

Determine which nodes are ready.

Decide concurrency limits per node type (e.g., limit http-call parallelism).

Safely merge results, handling write conflicts, variable overrides, and visibility of failed node outputs.

3. Core Executor Implementation (simplified production version)

public class DagExecutionEngine {
    private final SkillRegistry skillRegistry;
    private final SnapshotRepository snapshotRepository;
    private final ExecutorService httpExecutor;
    private final ExecutorService dbExecutor;

    public DagExecutionResult execute(DagDefinition dag, TaskRuntime runtime) {
        ExecutionGraph graph = ExecutionGraph.from(dag);
        ExecutionSnapshot snapshot = snapshotRepository.loadOrCreate(runtime.taskId(), dag);
        SkillExecutionContext context = SkillContexts.from(runtime, snapshot);
        while (graph.hasRunnableNodes(snapshot)) {
            List<SkillNode> runnable = graph.nextRunnableNodes(snapshot);
            Map<SkillNode, CompletableFuture<SkillResult>> futures = new LinkedHashMap<>();
            for (SkillNode node : runnable) {
                Executor executor = chooseExecutor(node.type());
                futures.put(node, CompletableFuture.supplyAsync(() -> executeNode(node, context), executor));
            }
            for (Map.Entry<SkillNode, CompletableFuture<SkillResult>> e : futures.entrySet()) {
                SkillNode node = e.getKey();
                SkillResult result = waitNodeResult(node, e.getValue(), dag.timeoutMs());
                snapshot.markNodeFinished(node.id(), result);
                mergeOutputs(node, result, context);
                snapshotRepository.save(runtime.taskId(), snapshot);
                if (!result.success() && dag.failureStrategy() == FailureStrategy.FAIL_FAST) {
                    return DagExecutionResult.failed(runtime.taskId(), node.id(), result.errorMessage());
                }
            }
        }
        return snapshot.hasFailedNode()
            ? DagExecutionResult.partialFailed(runtime.taskId(), snapshot.failedNodes())
            : DagExecutionResult.success(runtime.taskId(), snapshot.nodeResults());
    }
    // executeNode, mergeOutputs, chooseExecutor, waitNodeResult omitted for brevity
}

The code demonstrates explicit handling of idempotency, timeout, and failure strategies.

Production‑Grade Skill Implementations

HttpCallSkill

public class HttpCallSkill implements Skill {
    private final ServiceDiscovery serviceDiscovery;
    private final WebClientFactory webClientFactory;
    private final JsonExtractor jsonExtractor;
    private final AssertionEvaluator assertionEvaluator;

    @Override public String type() { return "http-call"; }

    @Override public SkillResult execute(SkillExecutionContext ctx) {
        HttpNodeConfig cfg = ctx.currentNode().configAs(HttpNodeConfig.class);
        URI uri = serviceDiscovery.resolve(cfg.service(), cfg.path(), ctx);
        Map<String, Object> body = ctx.renderObject(cfg.body());
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        headers.set("X-Trace-Id", ctx.traceId());
        headers.set("X-Task-Id", ctx.taskId());
        cfg.headers().forEach((k,v) -> headers.set(k, ctx.resolveTemplate(v)));
        long start = System.currentTimeMillis();
        ClientResponseEnvelope resp = webClientFactory.create(cfg.timeoutMs())
            .exchange(cfg.method(), uri, headers, body);
        List<AssertionResult> asserts = assertionEvaluator.evaluateHttp(resp, cfg.assertions(), ctx);
        Map<String, Object> outputs = jsonExtractor.extract(resp.body(), cfg.extract());
        ctx.recordHttpExchange(HttpExchangeRecord.builder()
            .nodeId(ctx.currentNode().id())
            .method(cfg.method())
            .url(uri.toString())
            .requestBody(body)
            .responseCode(resp.statusCode())
            .responseBody(resp.body())
            .costMs(System.currentTimeMillis() - start)
            .build());
        boolean success = resp.success() && asserts.stream().allMatch(AssertionResult::passed);
        return SkillResult.builder()
            .success(success)
            .status(success ? SkillStatus.SUCCESS : SkillStatus.FAILED)
            .outputs(outputs)
            .assertions(asserts)
            .costMs(System.currentTimeMillis() - start)
            .build();
    }
}

DbCheckSkill

public class DbCheckSkill implements Skill {
    private final DataSourceRouter dataSourceRouter;
    private final SqlTemplateRenderer sqlTemplateRenderer;
    private final AssertionEvaluator assertionEvaluator;

    @Override public String type() { return "db-check"; }

    @Override public SkillResult execute(SkillExecutionContext ctx) throws SQLException {
        DbCheckNodeConfig cfg = ctx.currentNode().configAs(DbCheckNodeConfig.class);
        DataSource ds = dataSourceRouter.get(cfg.datasource());
        String sql = sqlTemplateRenderer.render(cfg.sql(), ctx.variablesView());
        long start = System.currentTimeMillis();
        List<Map<String, Object>> rows = new ArrayList<>();
        try (Connection conn = ds.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql);
             ResultSet rs = stmt.executeQuery()) {
            ResultSetMetaData meta = rs.getMetaData();
            while (rs.next()) {
                Map<String, Object> row = new LinkedHashMap<>();
                for (int i = 1; i <= meta.getColumnCount(); i++) {
                    row.put(meta.getColumnLabel(i), rs.getObject(i));
                }
                rows.add(row);
            }
        }
        List<AssertionResult> asserts = assertionEvaluator.evaluateRows(rows, cfg.assertions(), ctx);
        boolean success = asserts.stream().allMatch(AssertionResult::passed);
        ctx.recordDbQuery(DbQueryRecord.builder()
            .nodeId(ctx.currentNode().id())
            .datasource(cfg.datasource())
            .sql(sql)
            .rowCount(rows.size())
            .rows(rows)
            .costMs(System.currentTimeMillis() - start)
            .build());
        return SkillResult.builder()
            .success(success)
            .status(success ? SkillStatus.SUCCESS : SkillStatus.FAILED)
            .assertions(asserts)
            .outputs(Map.of("rowCount", rows.size()))
            .costMs(System.currentTimeMillis() - start)
            .build();
    }
}

WaitUntilSkill (asynchronous verification)

public class WaitUntilSkill implements Skill {
    @Override public String type() { return "wait-until"; }
    @Override public SkillResult execute(SkillExecutionContext ctx) {
        WaitUntilNodeConfig cfg = ctx.currentNode().configAs(WaitUntilNodeConfig.class);
        for (int attempt = 1; attempt <= cfg.maxAttempts(); attempt++) {
            boolean matched = evaluateCondition(cfg.condition(), ctx);
            if (matched) {
                return SkillResult.builder()
                    .success(true)
                    .status(SkillStatus.SUCCESS)
                    .outputs(Map.of("attempt", attempt))
                    .build();
            }
            sleep(cfg.intervalMs());
        }
        return SkillResult.builder()
            .success(false)
            .status(SkillStatus.TIMEOUT)
            .errorCode("WAIT_CONDITION_TIMEOUT")
            .errorMessage("condition not satisfied")
            .build();
    }
}

The wait‑until node abstracts polling logic, preventing duplicated sleep‑retry code in scripts.

Data Model & Table Design

Three‑layer tables support reporting and recovery:

CREATE TABLE automation_task (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    task_id VARCHAR(64) NOT NULL UNIQUE,
    dag_id VARCHAR(128) NOT NULL,
    dag_version VARCHAR(32) NOT NULL,
    status VARCHAR(32) NOT NULL,
    trigger_source VARCHAR(32) NOT NULL,
    trace_id VARCHAR(64) NOT NULL,
    context_json JSON NOT NULL,
    start_time DATETIME,
    end_time DATETIME,
    error_message VARCHAR(1024),
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_dag_status_created (dag_id, status, created_at),
    INDEX idx_trace_id (trace_id)
);

CREATE TABLE automation_task_node (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    task_id VARCHAR(64) NOT NULL,
    node_id VARCHAR(128) NOT NULL,
    node_type VARCHAR(64) NOT NULL,
    status VARCHAR(32) NOT NULL,
    retry_count INT NOT NULL DEFAULT 0,
    cost_ms BIGINT NOT NULL DEFAULT 0,
    request_snapshot JSON,
    response_snapshot JSON,
    error_code VARCHAR(64),
    error_message VARCHAR(1024),
    started_at DATETIME,
    finished_at DATETIME,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_task_node (task_id, node_id),
    INDEX idx_task_status (task_id, status)
);

CREATE TABLE automation_assertion_result (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    task_id VARCHAR(64) NOT NULL,
    node_id VARCHAR(128) NOT NULL,
    assertion_type VARCHAR(64) NOT NULL,
    field_name VARCHAR(128),
    expected_value VARCHAR(512),
    actual_value VARCHAR(512),
    passed TINYINT(1) NOT NULL,
    message VARCHAR(1024),
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_task_node (task_id, node_id)
);

This separation enables task‑level statistics, node‑level troubleshooting, and fine‑grained assertion reporting.

Process Flows

Normal Flow

Submit Task → Parse DAG → Init Context → Execute Ready Nodes → Merge Outputs →
Check for New Ready Nodes → If All Success → Generate Report
Otherwise → Record Failure

Exception Flow

Node Failure → Is Retryable? → Exponential Backoff → Retry Success?
If still failing → FAIL_FAST? → Stop DAG or Continue other branches

Recovery Flow

Worker Crash → Task stays RUNNING → Scheduler detects heartbeat timeout →
New Worker picks up task → Load snapshot from Redis → Skip completed idempotent nodes →
Resume pending nodes

Recovery requires complete snapshots and idempotent Skills.

High Concurrency & Scalability: The Real Challenge Is Resource Governance

Thread‑pool isolation : separate pools for HTTP, DB, and waiting nodes.

Connection‑pool limits : enforce concurrency semaphores for db-check nodes.

Task‑level rate limiting : API throttling at entry and worker‑side consumption limits.

Horizontal scaling : workers are stateless, can be scaled via Kubernetes Pods; snapshots stored in Redis, results in MySQL.

KEDA auto‑scaling based on Kafka consumer lag (example YAML omitted for brevity).

Observability Construction

Logging

Task‑level logs (start, end, overall status).

Node‑level logs (input, output, duration).

Audit logs (raw HTTP request/response, SQL text, message payload) with sensitive fields masked.

Metrics (Prometheus)

automation_task_total

automation_task_success_total

automation_task_duration_ms

automation_node_duration_ms

automation_node_retry_total

automation_db_query_slow_total

Tracing

Each task gets a traceId propagated via HTTP headers, enabling end‑to‑end tracing in Jaeger/Tempo.

Alerting

Multi‑level alerts: large‑scale task failures, DAG success‑rate drops, node latency spikes, and sudden DB‑validation error rate increases.

Security & Environment Governance

Fine‑grained RBAC for task submission, result viewing, and write‑operation execution.

Separate environments (local, integration, pre‑release, production‑shadow) with configuration‑driven service addresses and DB connections.

Protection for dangerous operations (real payment, data deletion, SMS/Email sending) via approval workflows or whitelist.

Common Pitfalls & Wrong Solutions

Treating Skill as ad‑hoc script fragments – loses the contract and governance.

Embedding all logic in prompts or scripts – makes retries, waiting, and compensation unmanageable.

Only asserting HTTP 200 without DB verification – masks eventual consistency failures.

Ignoring asynchronous flows – leads to incomplete success detection.

Jumping to full platformization when only a handful of simple scenarios exist – start with a single‑machine DAG engine.

Evolution Path: From Single‑Machine Tool to Team‑Level Platform

Stage 1 – Single‑Machine Engine : local YAML, single‑process executor, HikariCP + WebClient, MySQL result storage.

Stage 2 – Service‑ified Platform : add task API, web UI, independent workers, Redis snapshots.

Stage 3 – Distributed Platform : Kafka‑driven dispatch, K8s + KEDA scaling, multi‑tenant RBAC, full observability suite.

Checklist Before Going Live

All Skills expose a clear input/output contract.

DAG validator detects cycles, orphan nodes, and missing variables.

HTTP/DB/MQ nodes have isolated resources (thread pools, connection limits).

No non‑idempotent node can be retried unintentionally.

Snapshots support fault recovery.

Assertions cover both immediate responses and eventual consistency.

Logging, metrics, and tracing are fully integrated.

All environment‑specific values are externalized.

Dangerous operations are protected by approvals or whitelists.

Slow‑SQL, timeout, and retry‑storm safeguards are in place.

Conclusion

Replacing glue scripts with a Skill‑based, DAG‑driven automation engine transforms fragile, hard‑to‑maintain test code into a scalable, observable, and recoverable engineering capability. The three key benefits are:

Skill abstraction unifies HTTP calls, DB checks, message verification, and waiting logic.

DAG representation turns procedural scripts into declarative, version‑controlled workflows.

A distributed execution engine provides concurrency control, retry/compensation, snapshot recovery, and resource governance, enabling platform‑level adoption.

Teams still using a few serial scripts can adopt the single‑machine engine immediately; teams already facing scenario explosion, flaky CI runs, and opaque failures should invest in the full Skill orchestration platform.