Machines

A workflow step parks a row and waits. A person resolves it from the dashboard, a service over HTTP, and with the machine surface a machine resolves it over MQTT. The row, the wait and the result are the same in every case. This page is the contract for a host: the line file, handoff rows, command rows, the wire codec and the test rig.

Declare a line

index.ts

import { start, Machines } from '@hotmeshio/long-tail';

await start({
  database: { connectionString: process.env.DATABASE_URL },
  workers: [/* your route workflows */],
  machines: {
    lines: [Machines.loadLine('/etc/line.json')],
    codec: myCodec, // optional: the machines' wire protocol (default: the generic line protocol)
    rows: myRows,   // optional: how your rows read as handoffs (default: the standard handoff facets)
  },
});

Install mqtt in the host; it is an optional peer dependency. Transports come from MACHINE_* environment variables (see Configuration).

For each line, one replica leads by holding a Postgres advisory lock on its own connection. The leader makes sure the line’s bot accounts exist and runs the gateway. Other replicas wait on the lock and take over when the leader’s session ends.

The line file

line.json

{
  "siteId": "site-1",
  "lineId": "line-1",
  "roles": { "exit": "inspector", "operator": "line-operator", "receivers": ["packer"] },
  "machines": [
    {
      "machineId": "bond-cell-1", "process": "bond", "role": "bond-cell", "carrier": "none",
      "cycleSec": 8, "queueDepth": 4,
      "recipes": { "RCP-BOND-A": ["2"] },
      "allowedSoftware": ["1.4.0"]
    }
  ]
}

Field	Meaning
machines[].role	The role whose handoff rows the machine resolves. Default line-<process>.
machines[].address	The physical machine on the wire: topic level, client id and certificate name. Default machineId.
machines[].allowedSoftware	Versions the line accepts. Another version is refused before any dispatch.
machines[].queueDepth	Most jobs the machine is sent or holds at once. Unset means no limit.
machines[].requireArtifact	The machine needs a job file. A handoff without one is refused before anything is sent.
addressing	declared (default) or learned. See Learned addresses.
roles.exit	The person who receives units at the end of a route. Default line-exit.
roles.operator	Gets failed handoffs, stopping alarms, and offline or quiet machines. Default line-operator.
roles.receivers	Other people’s roles that take units straight from a machine.

start({ machines }) declares the station, exit and operator roles unless your roles already declare them. Set declareRoles: false to own them all.

Handoff rows

A handoff is a row on the receiving machine’s role (or a receiver role, for a person) whose metadata says what moves:

workflows/route.ts

import { conditionalAccumulator } from '@hotmeshio/long-tail';

const result = await conditionalAccumulator(`route-${routeKey}-hop-2`, {
  role: 'cut-cell',
  type: 'handoff',
  subtype: 'cut',
  metadata: {
    siteId: 'site-1', lineId: 'line-1',
    units: [{ unitId: 'U1', carrierId: 'C7' }, { unitId: 'U2', carrierId: 'C7' }],
    jobId: `${routeKey}-cut`, senderJobId: `${routeKey}-bond`,
    fromMachine: 'bond-cell-1', toMachine: 'cut-01',
    recipeId: 'RCP-CUT-A', recipeVersion: '1', process: 'cut',
    attempt: 1,
  },
  accumulate: {},
  timeout: '30m',
});

Facet	Meaning
siteId, lineId	The line. A gateway acts only on its own line’s rows.
units	[{ unitId, carrierId, artifact?, parameters? }]: units that move together. A unit’s artifact is { url, sha256 }.
jobId, senderJobId	The receiver’s job and the sender’s. A second open row naming the same job resolves dispatch-refused.
fromMachine, toMachine	Machine ids, or null for line entry and for a person.
recipeId, recipeVersion, process	What the receiver runs.
artifactUrl, artifactSha256	The job file, by reference.
parameters	Job settings passed to the machine as given. Their meaning is between the host and its codec.
attempt	Retry number.

The gateway dispatches to toMachine when the row is created, claims the row as that machine once it accepts, records each release and arrival as items, and resolves the row as the receiver once every unit has arrived:

{ received: true, receiver: 'cut-01', units: ['U1', 'U2'] }

A row on a person’s role resolves the same way from the dashboard. A route can mix machines and people and read every hop the same way.

Machines.handoffProgress(units, items) reads where units stand from a row’s items, or from $accumulated on a finished wait: { released, arrived, missing }.

When a handoff fails

The row resolves { received: false, errorCode, unitId?, reason?, detail }:

errorCode	Cause
dispatch-refused	The machine refused or failed the dispatch (mode, recipe, queue, artifact), or a required artifact is missing.
dispatch-failed	The machine accepted the job, then could not prepare it.
no-reply	No reply after every attempt, and the machine’s snapshot does not list the job. The dispatch is cancelled.
software-unsupported	The machine runs a version the line does not allow.
sender-rejected	The sender finished a unit as reject, hold, scrap or rework.
jam, incomplete-handshake, carrier-mismatch, unreadable-id	The move between machines failed.
withdrawn	The receiver gave the job back.

When a handoff ends without every unit, the gateway withdraws only the jobs none of whose units reached the receiver. A job already under way stays, and the workflow decides what happens next.

Machines.handoffDecisions(errorCode) returns a default set of decisions. A failed move leaves units on the line, so retry, manual and scrap fit. After sender-rejected or a partial arrival, only manual and scrap fit. A workflow may offer its own decisions instead.

Traceability

When a sender finishes a unit, the gateway records the release and, in the same statement, merges two facets into the row:

Facet	Holds
trace	Per unit: { unitId, machineId, jobId, result, materials: [{ material, lot }] }
lots	Every lot those units consumed, once each

Both are containment-searchable, so every row that used a lot is one query: GET /api/escalations?facets={"lots":["LOT-7"]}. A unit’s path is every handoff row whose units holds it.

Your own row facets

A host whose rows already carry their own facets supplies a HandoffRowAdapter:

const rows: Machines.HandoffRowAdapter = {
  openRows: () => [{ type: 'station-step' }],
  toHandoff: (row) => row.type === 'station-step' ? { siteId: row.metadata.plant /* ... */ } : null,
};

Pacing

A machine with a queueDepth is sent no more jobs than that at once. Beyond it, a handoff waits in the machine’s role queue, pending and unclaimed, the way work waits for a busy person. The oldest waiting row goes out when there is room. A machine with nothing in hand always takes the next row.

A machine’s load counts rows dispatched into it, rows whose units still wait on it, and units that arrived and have not left. A replica taking the line reads every open row back before it sends anything, so pacing survives a restart.

Commands

Ask a machine to start, stop, pause, resume, change over or cancel a job by parking a machine-command row on its role:

workflows/changeover.ts

import { conditional, Machines } from '@hotmeshio/long-tail';

const reply = await conditional<Machines.CommandResolution>(`changeover-${orderId}`, {
  ...Machines.commandRow(Machines.commandTarget(line, 'bond-cell-1'), 'changeover', {
    recipe: { id: 'RCP-BOND-B', version: '1' },
  }),
  timeout: '10m',
});

The gateway sends the command and resolves the row as the machine:

{ accepted: true, machineId: 'bond-cell-1', op: 'changeover', status: 'accepted', reason: 'changing over to RCP-BOND-B v1' }

status	Meaning
accepted	The machine took the command.
refused, failed	The machine said no. Its reason is in reason.
no-reply	No reply after every attempt. The command expired, so a late copy is refused at the machine.
unsupported	Not a command a row may ask for, not a machine on this line, or a command the codec has no message for. Nothing was sent.

cancel needs args.jobId. Parking a command row takes write access to the machine’s role, the same as any row there. Every command carries a cmdId, and a repeat is answered, never run twice.

Rows for people

The gateway raises these on the operator role:

Row	Raised when	Resolved when
line-alarm	A critical or emergency alarm is raised	The machine clears it. If a person resolves it first, the gateway sends ack-alarm.
machine-offline	Presence goes offline past a debounce	The machine is heard again
machine-quiet	A connected machine is silent for MACHINE_QUIET_MS	It is heard again
contract-violation	A strict codec decodes a message that breaks the contract	A person resolves it
machine-registration	A learned-address claim disagrees with a binding	A person resolves it with { decision: 'approve' } or reject

A replica taking a line reads back its learned addresses, every open row and each handoff’s arrivals, then the machines’ retained state. A clear or a return that happened while no gateway was listening still resolves its row.

The wire

The gateway speaks the generic line protocol. A codec translates your machines’ protocol:

interface LineCodec {
  root: string;
  channels: { events: string[]; telemetry: string[]; commands: string[] };
  topicFor?(scope: { siteId?, lineId?, address? }, channel: string): string;
  unitsPerDispatch: number | null;      // 1: one dispatch per unit
  dispatchOn?: 'created' | 'arrival';   // when a handoff's job goes out
  strict?: boolean;                     // broken messages go to a person, never acted on
  decode(topic: string, payload: string, retained?: boolean): Inspection[];
  encode(ident: MachineIdent, command: MachineCommand): WireMessage[];
  receipt?(ident: MachineIdent, messageIds: string[]): WireMessage[];
}

Option	Effect
decode returning several events	Lets a codec infer a receipt from another event. Machines.genericCodec({ root, receiptFrom: 'started' | 'bound' }) does this.
unitsPerDispatch: 1	One job per unit. The row is claimed once every job is accepted, and the accepted jobs are cancelled if any is refused.
dispatchOn: 'arrival'	For machines that take work only for units they have scanned in. Each unit carries its arrival event, so encode can echo the visit id.
strict: true	A message with contract violations is confirmed, never acted on, and shown in one contract-violation row.
receipt	Confirms messages for machines that resend until confirmed, in batches of up to 100 after MACHINE_RECEIPT_DELAY_MS.
encode returning []	The machines lack that command. A command row resolves unsupported.

The generic protocol’s topics are {root}/{site}/{line}/{machine}/{channel}, and every message is wrapped in { v, id, at, source, body }:

Channel	Purpose
presence (retained)	Birth and will, with the software version
snapshot (retained)	Mode, recipe, jobs, alarms, counters, boot id
job	Per unit: queued, ready, arrived, bound, started, progress, finished, released, transfer-failed, withdrawn
alarm/<id> (retained while active)	Raise, acknowledgement, clear
telemetry	Readings
report/<name>	Production, availability and inventory reports
cmd / cmd/reply	dispatch, cancel, start, stop, pause, resume, changeover, ack-alarm, snapshot; replies accepted, completed, refused, failed

Topology and identity

 machines ──MQTT, TLS, client cert──▶ machine broker ──bridge──▶ site NATS
                                         (Mosquitto)               JetStream
                                              ▲                        ▲
                                              └── gateway (commands)    └── gateway (streams, read only)

• Machines log in over MQTT 3.1.1 or 5 with their own certificate. The CN ({site}-{line}-{address}) is the username. The broker ACL lets each write only its own channels and read only its own commands.
• Streams: the bridge carries the line’s topics into the site NATS. JetStream keeps LINE_EVENTS for 30 days and LINE_TELEMETRY for 3 days.
• Commands go to the machine broker at QoS 1, so they queue for a machine that is away.
• Bot accounts: each machine is a bot account on its station and receiver roles, so claimed_by and resolved_by name the machine. The line will not start while a person’s account holds one of those names.
• Rows are the authority: an escalation event tells the gateway which row to read. It reads the row back and acts on the stored role, status and metadata.
• Rendering: Machines.renderSiteConf, renderMosquittoConf and renderMosquittoAcl(line, codec) render broker configuration from the line.

Learned addresses

With addressing: 'learned', a machine tells the line which station it is, so it can be commissioned, moved or replaced without editing the file.

1. Every message carries the machine’s address and the station it claims (source.station in the generic protocol).

2. A station with no address binds to the first address that claims it, and its waiting work goes out. Until then, handoffs wait in its role queue and command rows resolve unsupported.

3. A claim that disagrees with an existing binding raises one machine-registration row. The claimant’s messages have no effect until a person approves.

Bindings live on each station’s bot account as the user property machineAddress. A replica taking the line reads them back before it sends anything.

Configuration

Variable	Default	Meaning
MACHINE_TRANSPORT		nats (site NATS streams) or mqtt (any MQTT broker)
MACHINE_ESCALATION_FEED	in-process	core subscribes to Long Tail’s NATS for a standalone gateway
MACHINE_TOPIC_ROOT	line/v1	The generic codec’s root
MACHINE_LINE_CONFIG		Path of the line JSON
MACHINE_SITE_NATS_URL, _USER, _PASSWORD		Site NATS for the streams
MACHINE_MQTT_URL, _PROTOCOL	protocol 5	Machine broker. Use 4 for MQTT 3.1.1.
MACHINE_MQTT_CA, _CERT, _KEY		TLS files. {clientId} in a path picks each identity’s certificate.
MACHINE_REPLY_TIMEOUT_MS	5000	Wait for a command reply
MACHINE_DISPATCH_ATTEMPTS	3	Tries per command
MACHINE_DISPATCH_TTL_MS	600000	How long a dispatch stays valid at the machine
MACHINE_OFFLINE_DEBOUNCE_MS	15000	Offline this long before a row
MACHINE_QUIET_MS	90000	Silent this long before a row. 0 turns it off.
MACHINE_RECEIPT_DELAY_MS	250	How long confirmed ids gather before a receipt

Test a codec

Prove a codec and row adapter against the real gateway in memory. The test plays the machine:

codec.test.ts

import { Machines } from '@hotmeshio/long-tail';
import { FakeLongTail, until } from '@hotmeshio/long-tail/build/services/machines/testing';

const line = Machines.loadLine('./line.json');
const broker = new Machines.MemoryBroker({ redeliverMs: 5 });
const feed = new Machines.MemoryEscalationFeed(broker);
const lt = new FakeLongTail(feed);
const gateway = new Machines.MachineGateway({
  bus: new Machines.MemoryMachineBus(broker, Machines.lineFilter(myCodec.root, line.siteId, line.lineId)),
  feed, lt, line, codec: myCodec, rows: myRows,
  settings: { topicRoot: myCodec.root, replyTimeoutMs: 50, dispatchAttempts: 2, offlineDebounceMs: 10, dispatchTtlMs: 60_000, quietMs: 0 },
});
await gateway.start();

broker.connect('bond-cell-1');
broker.subscribe('bond-cell-1', myCommandTopic('bond-cell-1'), (msg) => broker.publish(myReplyTopic, myAcceptedReply(msg)));

const row = lt.parkRow({ type: 'station-step', role: 'bond-cell', metadata: { /* your facets */ } });
await until(() => lt.rows.get(row.id)?.assigned_to === 'bond-cell-1', 2_000);
broker.publish(capturedArrival.topic, capturedArrival.payload);
expect((await lt.ended(row.id)).resolver_payload).toMatchObject({ received: true, receiver: 'bond-cell-1' });

The same rig covers the unhappy paths: cut the machine off for no-reply, answer with a refusal for dispatch-refused, publish a not-ok finish for sender-rejected, or restart the gateway to prove a takeover reads its rows back.

Related

Batches and accumulatorsHow a handoff collects unit arrivals.

Factory line

A machine is a resolver

Command a machine with a row