npm package: @nsshunt/stsrunnerframework
Description: STS Runner Framework
Author: STS
Latest version: 2.0.17

# STS Runner Framework

A **high-performance asynchronous runner orchestration framework** for Node.js and browser environments.

The framework enables you to build systems that execute large numbers of **concurrent test runners or workload generators** across **worker processes**, with:

* asynchronous runner lifecycle management
* distributed worker execution
* structured telemetry
* instrumentation integration
* dynamic runner configuration
* state synchronization
* archive history tracking
* event-driven observability

The framework is designed for scenarios such as:

* load testing
* distributed task execution
* asynchronous job orchestration
* testing frameworks
* automation systems
* telemetry-driven workload engines

---

# Table of Contents

* Overview
* Architecture
* Core Concepts
* System Components
* Execution Flow
* Telemetry and Observability
* Runner Lifecycle
* Worker Lifecycle
* State Synchronization
* Archiving
* Client Usage
* Creating Custom Runners
* Example Test Runner
* Example Client
* Threading Model
* Design Goals
* Future Enhancements

---

# Overview

The **STS Runner Framework** provides a scalable architecture for executing **multiple runners across multiple workers**, with centralized orchestration.

The system consists of three primary layers:

```
Client Application
        │
        ▼
STSWorkerManager (Coordinator)
        │
        ▼
Workers (Execution Hosts)
        │
        ▼
Runners (User Workload Logic)
```

Key properties:

* **Workers isolate execution**
* **Runners perform actual work**
* **Manager coordinates everything**
* **Telemetry flows upward**
* **Commands flow downward**

---

# Architecture

The framework is composed of several subsystems:

```
Client Application
        │
        ▼
STSWorkerManager
│
├── WorkerRegistry
├── WorkerInstanceManager
├── RunnerLifecycleManager
├── WorkerStateSynchroniser
├── WorkerCommandCoordinator
├── AsyncRunnerInstanceManager
├── STSMessageBroker
└── ArchiveManager
```

Each subsystem has a clearly defined responsibility.

---

# Core Concepts

## Worker

A **Worker** represents an execution host.

Workers may run:

* inside Node.js worker threads
* inside browser workers
* in mocked in-process environments

Workers host **multiple runners**.

---

## Runner

A **Runner** is the unit that performs actual work.

Examples:

* HTTP load test
* API validation
* database operation
* workflow automation
* synthetic telemetry generator

Runners implement:

```ts
interface IRunnerInstance
```

Typical methods:

```
ExecuteRunner()
StartRunner()
StopRunner()
PauseRunner()
ResumeRunner()
ResetRunner()
TerminateRunner()
UpdateRunner()
Completed()
```

---

## Manager

The **STSWorkerManager** is the central coordinator.

It:

* creates workers
* creates runners
* dispatches commands
* collects telemetry
* synchronizes state
* archives completed runners

---

## Telemetry

Telemetry represents runtime metrics generated by runners.

Examples:

```
requestCount
errorCount
velocity
latency
duration
activeRequestCount
tx
rx
message[]
```

Telemetry flows from:

```
Runner → Worker → Manager → Client
```

---

# System Components

## STSWorkerManager

The **core orchestrator** of the framework.

Responsibilities:

* worker creation
* runner creation
* lifecycle command dispatch
* state synchronization
* archive management
* telemetry processing

Example initialization:

```ts
const wm = new STSWorkerManager({
    workerFactory,
    maxArchiveListLength: 200,
    logger: defaultLogger,
    logLevel: 4,
    messageTimeout: 10000
});
```

---

## WorkerRegistry

Stores the in-memory runtime graph of the system.

```
WorkerRegistry
   ├── WorkerEx
   │       └── RunnerEx
```

Provides:

```
AddWorker()
AddRunner()
DeleteWorker()
DeleteRunner()
GetWorkersMap()
GetWorkersCoreMap()
GetNextAvailableWorker()
GetBusiestWorker()
```

---

## WorkerInstanceManager

Responsible for **creating worker instances** and wiring system events.

Handles:

* worker creation
* exit events
* error events
* message error events

Supports:

```
Node Worker Threads
Browser Workers
Mock Workers
```

---

## STSMessageBroker

Provides **message routing between manager and workers**.

Responsibilities:

```
SendMessageToWorkerForRunnerWithCallBack()
SendMessageToWorkerForWorkerWithCallBack()
SetupMessagePort()
SetupMessagePortListener()
```

Uses `MessagePort` communication channels.

---

## AsyncRunnerInstanceManager

Creates and manages **runner instances inside workers**.

Responsibilities:

* runner creation
* runner execution control
* instrument controller attachment

---

## RunnerLifecycleManager

Handles **runner telemetry and lifecycle state updates**.

Responsibilities:

```
ProcessTelemetry()
RunnerStateChange()
EmitRunnerEvent()
SyncRunnerData()
```

Ensures the in-memory model stays synchronized.

---

## WorkerCommandCoordinator

Handles **command dispatching**.

Two command scopes:

### Worker Commands

```
StartWorkers
StopWorkers
PauseWorkers
ResumeWorkers
ExecuteWorkers
ResetWorkers
TerminateWorkers
UpdateWorkers
```

### Runner Commands

```
StartRunners
StopRunners
PauseRunners
ResumeRunners
ExecuteRunners
ResetRunners
TerminateRunners
UpdateRunners
```

Commands are executed asynchronously across workers.

---

## WorkerStateSynchroniser

Ensures that the **manager's in-memory model matches actual worker state**.

Mechanism:

```
Manager → Worker : GetRunners
Worker → Manager : Runner states
```

Methods:

```
GetSyncedWorkers()
GetSyncedWorkersCore()
```

---

## ArchiveManager

Stores completed runner history.

Archived runners contain:

```
runnerId
runnerOptions
runnerHistory[]
telemetrySnapshots
```

Provides:

```
GetArchiveList()
```

---

# Execution Flow

### Worker Creation

```
Client → STSWorkerManager.AddWorker()
```

Flow:

```
Manager
   ↓
WorkerInstanceManager
   ↓
WorkerFactory
   ↓
WorkerInstance
   ↓
WorkerRegistry
```

---

### Runner Creation

```
Client → Worker.AddRunner()
```

Flow:

```
Manager
   ↓
AsyncRunnerInstanceManager
   ↓
Runner created
   ↓
Worker receives AddRunner command
```

---

### Runner Execution

```
StartRunner()
```

Execution loop:

```
Runner.ExecuteRunner()
    ↓
Update telemetry
    ↓
Sleep or perform work
    ↓
Publish telemetry
```

---

# Telemetry and Observability

Telemetry is buffered and published using a hybrid strategy:

```
Flush if:
    buffer size reached
    OR
    timeout reached
```

Example logic:

```
maxBufferSize = 50
timeout = 1000ms
```

Telemetry metrics include:

```
requestCount
errorCount
retryCount
velocity
latency
duration
activeRequestCount
network RX/TX
```

Instrumentation integrates with:

```
@nsshunt/stsobservability
@nsshunt/stsinstrumentmanagerclient
```

---

# Runner Lifecycle

Runner states include:

```
initializing
running
paused
stopped
completed
error
terminated
```

Lifecycle transitions generate events:

```
Telemetry
StateChange
```

Example:

```ts
runner.on('Telemetry', handler)
runner.on('StateChange', handler)
```

---

# Worker Lifecycle

Worker events include:

```
exit
error
messageerror
```

The WorkerInstanceManager wires these events and forwards them to the manager.

---

# State Synchronization

The manager maintains an **in-memory graph**.

Before returning model queries:

```
GetWorkers()
GetWorkersCore()
```

The system performs:

```
SyncAllWorkers()
```

Which queries every worker for current runner state.

---

# Archiving

Completed runners are archived by the **ArchiveManager**.

The archive contains:

```
runner metadata
execution history
telemetry snapshots
state transitions
```

Example:

```ts
const archiveList = await wm.GetArchiveList({});
```

---

# Client Usage

Example initialization:

```ts
const wm = new STSWorkerManager({
    workerFactory,
    maxArchiveListLength: 200,
    logger: defaultLogger,
    logLevel: 4,
    messageTimeout: 10000
});
```

---

## Create Worker

```ts
const worker = await wm.AddWorker({
    hostName: "host01",
    agentId: "agent01",
    mocked: true,
    tags: ["testing"],
    logLevel: 4
});
```

---

## Create Runner

```ts
const runner = await worker.AddRunner({
    testType: "TestCase01",
    executionProfile: {
        iterations: 40,
        delayBetweenIterations: 250
    }
});
```

---

## Runner Events

```ts
runner.on("Telemetry", handler);
runner.on("StateChange", handler);
```

---

## Start Runner

```ts
await runner.Start();
```

or

```ts
await wm.StartRunners(worker.id, [runner.id]);
```

---

## Pause / Resume

```ts
await runner.Pause();
await runner.Resume();
```

---

## Update Runner

```ts
await runner.Update({
    executionProfile: {
        iterations: 30
    }
});
```

---

# Creating Custom Runners

Implement:

```ts
IRunnerInstance
```

Example skeleton:

```ts
class MyRunner implements IRunnerInstance {

    async ExecuteRunner() {
        // perform work
        return true;
    }

    async StartRunner() {}
    async StopRunner() {}
    async PauseRunner() {}
    async ResumeRunner() {}
}
```

---

# Example Test Runner

Two reference implementations exist:

```
TestCase01
TestCase02
```

They demonstrate:

* telemetry batching
* simulated workload
* logging
* lifecycle transitions

---

# Example Client

Example scenarios included:

```
client test 1:
multiple workers + runners

client test 2:
single runner dynamic update
```

Both demonstrate full framework usage.

---

# Threading Model

The framework supports multiple execution models:

```
Node Worker Threads
Browser Workers
Mock Workers
```

Mock workers are useful for testing.

---

# Design Goals

The framework was designed with the following principles:

### Scalability

Supports large numbers of runners across workers.

---

### Observability

First-class telemetry and instrumentation integration.

---

### Flexibility

Works in:

* Node
* Browser
* test environments

---

### Isolation

Workers isolate execution workloads.

---

### Event-Driven

Runners emit structured events.

---

### Dynamic Control

Runners can be:

```
paused
resumed
reset
updated
terminated
```

at runtime.

---

# Future Enhancements

Potential roadmap features:

* distributed multi-machine worker clusters
* built-in load test profiles
* Web UI dashboard
* Prometheus exporters
* distributed tracing
* adaptive execution control
* auto-scaling workers

---

# License

MIT License.


# Architecture Diagrams

## System Architecture

```mermaid
flowchart TB

    Client["Client Application"]

    subgraph Manager["STSWorkerManager"]
        WorkerRegistry
        WorkerInstanceManager
        RunnerLifecycleManager
        WorkerCommandCoordinator
        WorkerStateSynchroniser
        AsyncRunnerInstanceManager
        ArchiveManager
        STSMessageBroker
    end

    subgraph Workers["Worker Threads / Web Workers"]
        Worker1["WorkerInstance"]
        Worker2["WorkerInstance"]
    end

    subgraph Runners["Runner Instances"]
        Runner1["RunnerInstance"]
        Runner2["RunnerInstance"]
        Runner3["RunnerInstance"]
    end

    Client --> Manager

    Manager --> WorkerRegistry
    Manager --> WorkerInstanceManager
    Manager --> RunnerLifecycleManager
    Manager --> WorkerCommandCoordinator
    Manager --> WorkerStateSynchroniser
    Manager --> AsyncRunnerInstanceManager
    Manager --> ArchiveManager
    Manager --> STSMessageBroker

    WorkerInstanceManager --> Worker1
    WorkerInstanceManager --> Worker2

    Worker1 --> Runner1
    Worker1 --> Runner2
    Worker2 --> Runner3

    STSMessageBroker --> Worker1
    STSMessageBroker --> Worker2
```
## Component Interaction

```mermaid
flowchart LR

Client --> STSWorkerManager

STSWorkerManager --> WorkerRegistry
STSWorkerManager --> WorkerInstanceManager
STSWorkerManager --> WorkerCommandCoordinator
STSWorkerManager --> RunnerLifecycleManager
STSWorkerManager --> WorkerStateSynchroniser
STSWorkerManager --> AsyncRunnerInstanceManager
STSWorkerManager --> ArchiveManager
STSWorkerManager --> STSMessageBroker

WorkerCommandCoordinator --> STSMessageBroker
WorkerStateSynchroniser --> STSMessageBroker

STSMessageBroker --> Worker

Worker --> Runner

Runner --> RunnerLifecycleManager

RunnerLifecycleManager --> WorkerRegistry

RunnerLifecycleManager --> ArchiveManager
```

## Runner Lifecycle

```mermaid
stateDiagram-v2

[*] --> initializing

initializing --> running : Start

running --> paused : Pause

paused --> running : Resume

running --> completed : Iterations reached

running --> stopped : Stop

running --> error : Exception

running --> terminated : Terminate

paused --> terminated : Terminate

completed --> [*]

stopped --> [*]

error --> [*]

terminated --> [*]
```

## Execution Sequence

```mermaid
sequenceDiagram

    participant Client
    participant Manager as STSWorkerManager
    participant Broker as MessageBroker
    participant Worker
    participant Runner

    Client->>Manager: AddWorker()
    Manager->>Worker: Create WorkerInstance

    Client->>Manager: AddRunner()
    Manager->>Broker: Send AddRunner
    Broker->>Worker: AddRunner message
    Worker->>Runner: Create RunnerInstance

    Client->>Manager: StartRunner()

    Manager->>Broker: StartRunner command
    Broker->>Worker: StartRunner
    Worker->>Runner: ExecuteRunner()

    loop Execution Loop
        Runner->>Runner: Perform Work
        Runner->>Worker: PostTelemetry
        Worker->>Broker: Telemetry Message
        Broker->>Manager: Telemetry Event
        Manager->>Client: Emit Telemetry Event
    end

    Runner->>Worker: Completed
    Worker->>Broker: RunnerCompleted
    Broker->>Manager: StateChange
    Manager->>Client: StateChange Event
```

## Worker Load Balancing / Runner Placement Diagram

This shows how the manager can choose a worker, typically by using the registry helpers like:
GetNextAvailableWorker()
GetBusiestWorker()
and then place runners onto workers.

```mermaid
flowchart TB

    NewRunner["New Runner Request"]

    subgraph Manager["STSWorkerManager"]
        Registry["WorkerRegistry"]
        Coordinator["Worker / Runner Coordination"]
        Selector{"Select Worker"}
    end

    subgraph WorkerPool["Live Workers"]
        W1["Worker A\n2 runners"]
        W2["Worker B\n5 runners"]
        W3["Worker C\n1 runner"]
    end

    NewRunner --> Manager
    Manager --> Registry
    Registry --> Selector
    Coordinator --> Selector

    Selector -->|"GetNextAvailableWorker()"| W3
    Selector -->|"GetBusiestWorker()"| W2

    W3 -->|"AddRunner()"| Added["Runner assigned to least-loaded worker"]
```

- The registry maintains the current live worker/runners graph
- worker selection can be based on current runner counts
- the least-loaded worker can be selected for simple balancing
- after selection, the runner is created and attached to that worker

## Telemetry Pipeline Diagram

This shows the full telemetry path from a concrete runner up to the client.

```mermaid
flowchart LR

    subgraph WorkerRuntime["Worker Runtime"]
        Runner["RunnerInstance"]
        ExecWorker["AbstractRunnerExecutionWorker"]
    end

    subgraph ManagerSide["Manager Side"]
        Broker["STSMessageBroker"]
        Lifecycle["RunnerLifecycleManager"]
        Registry["WorkerRegistry"]
        Sync["WorkerStateSynchroniser"]
    end

    Client["Client Application"]
    Instrument["PublishInstrumentController / Telemetry Output"]

    Runner -->|"updates instrumentData"| ExecWorker
    ExecWorker -->|"PostTelemetryById()"| Broker
    Broker -->|"InstrumentTelemetry message"| Lifecycle
    Lifecycle -->|"SyncRunnerData()"| Registry
    Lifecycle -->|"ProcessTelemetry()"| Instrument
    Registry --> Sync
    Sync --> Client
    Lifecycle -->|"runner.on('Telemetry')"| Client
    Lifecycle -->|"runner.on('StateChange')"| Client

```

- runners update their local instrumentData
- the worker runtime posts telemetry upward via PostTelemetryById()
- the message broker routes telemetry to the manager
- the runner lifecycle manager applies the update into the live registry model
- optional instrumentation publishers can also receive metric updates
- clients can observe telemetry through:
- event subscriptions
- synchronized model queries like GetWorkers()

<br>
──────────────────── ✦ ────────────────────
<br>
<br>
<br>

# stsrunnerframework (Old Legacy Doco - Previous Version)

terminated - this means the process has been terminated and no end-of-process steps should execute such as reports etc.

stopped - end the test and report as if completed normnally

# State Machine
```mermaid
---
config:
  theme: neo
  look: neo
title: STSRunnerFramework
---
stateDiagram
  direction TB
  [*] --> Created:Select Start
  Created --> Running
  Created --> Error:Error Caught
  Running --> Completed:Iterations Exhausted
  Running --> Error:Error Caught
  Running --> Stopped:Select Stop
  Completed --> [*]
  Running --> Paused:Select Pause
  Paused --> Running:Select Resume
  Paused --> Paused:Select Execute or Reset
  Paused --> Stopped:Select Stop
  Running --> Terminated:Select Terminate
  Paused --> Terminated:Select Terminate
  Paused --> Completed:Select Execute<br>Iterations Exhausted
  Terminated --> [*]
  Error --> [*]
  Stopped --> [*]
```

# Nodejs Factory Instance Example
```typescript
/* eslint @typescript-eslint/no-unused-vars: 0,  @typescript-eslint/no-explicit-any: 0 */  // --> OFF
import { IRunnerInstance, ITestRunnerTelemetryPayload, eIWMessageCommands, WorkerInstance } from './../index'
import { TestCase01 } from './testCase01'

import { parentPort } from 'worker_threads';

export class WorkerTestCases extends WorkerInstance {
    constructor() {
        super()
    }

    override CreateAsyncRunner = (testRunnerTelemetryPayload: ITestRunnerTelemetryPayload): IRunnerInstance | null => {
        const { runner } = testRunnerTelemetryPayload;
        switch (runner.options.testType) {
        case 'TestCase01' :
            return new TestCase01(this, runner)
        }
        return null;
    }
}

const worker = new WorkerTestCases();

parentPort?.on('message', (data: any) => {
    if (isNode) {
        const { command, payload } = data;
        if (command === eIWMessageCommands.MessagePort) {
            payload.port.on('message', (data: any) => {
                worker.ProcessMessage(data);
            });
            worker.ProcessMessage(data);
        } else {
            throw new Error(`Invalid command: [${command}]`)
        }
    }
});
```

# Agent Factory Instance Example
```typescript
/* eslint @typescript-eslint/no-unused-vars: 0,  @typescript-eslint/no-explicit-any: 0 */  // --> OFF
import { IRunnerInstance, ITestRunnerTelemetryPayload, WorkerInstance, eIWMessageCommands } from './../index'
import { TestCase01 } from './testCase01'

export class WorkerTestCases extends WorkerInstance {
    constructor() {
        super()
    }

    override CreateAsyncRunner = (testRunnerTelemetryPayload: ITestRunnerTelemetryPayload): IRunnerInstance | null => {
        const { runner } = testRunnerTelemetryPayload;
        switch (runner.options.testType) {
        case 'TestCase01' :
            return new TestCase01(this, runner)
        }
        return null;
    }
}

const worker = new WorkerTestCases();

onmessage = async function(data: any) {
    const { command, payload } = data.data;
    if (command === eIWMessageCommands.MessagePort) {
        payload.port.start();
        payload.port.addEventListener('message', (data: any) => {
            worker.ProcessMessage(data.data); // browser version
        });

        /*
        payload.port.addEventListener('messageerror', (error: any) => {
            console.error(`onmessageerror(): [${error}]`)
            worker.ProcessMessage(data.data); // browser version
        });
        */

        worker.ProcessMessage(data.data);
    } else {
        throw new Error(`Invalid command: [${command}]`)
    }
}
```

# Mocked Factory Instance Example (for testing only)
```typescript
/* eslint @typescript-eslint/no-unused-vars: 0,  @typescript-eslint/no-explicit-any: 0 */  // --> OFF
import { IRunnerInstance, ITestRunnerTelemetryPayload, WorkerInstance } from './../index'
import { TestCase01 } from './testCase01'

import { MessagePort } from 'worker_threads';

import { JSONObject } from '@nsshunt/stsutils';

export class WorkerTestCases extends WorkerInstance {
    #port: MessagePort | null = null;

    constructor() {
        super()
    }

    SetPort = (message: JSONObject) => {
        this.#port = message.payload.port as MessagePort;

        this.#port.on('message', (data: any) => {
            if (isNode) {
                this.ProcessMessage(data);
            } else {
                this.ProcessMessage(data.data); // browser version
            }
        });

        this.ProcessMessage(message);
    }

    override CreateAsyncRunner = (testRunnerTelemetryPayload: ITestRunnerTelemetryPayload): IRunnerInstance | null => {
        const { runner } = testRunnerTelemetryPayload;
        switch (runner.options.testType) {
        case 'TestCase01' :
            return new TestCase01(this, runner)
        }
        return null;
    }
}
```