edge-service-sdk

README

edge-service-sdk

edge-service-sdk is the shared runtime foundation for the edge-service family.

It extracts the common runtime, control, and transport capabilities out of protocol-specific projects so that S7, Fanuc, Modbus, Mitsubishi, and other device services can focus on protocol adapters and device-domain logic only.

Core Capabilities

• unified config loading, profile merge, and runtime normalization
• auth bootstrap, token issuance, credential query, and protected request verification
• ops HTTP runtime endpoints for health, readiness, runtime status, model query, and control tracing
• device status tracking and optional MQTT status reporting
• telemetry event normalization and a SQLite-first durable telemetry outbox
• built-in EVENT engine for connection, OEE, alarm, fault, pulse and rise-clear rules
• event profile loading from device.eventDir, device-level eventProfile binding, state persistence, summary windows, and a separate durable event outbox
• MQTT lifecycle management with explicit reconnect, subscription recovery, and health checks
• property request execution over HTTP and MQTT, including auto-report and progress/failure context
• command registry, synchronous/asynchronous command execution, and persisted command result tracking
• control job persistence, result history, diagnostics, export, and MQTT query handling
• dependency checks, worker supervision, and shared logging contracts

Reliability Hardening In v0.10.0

• MQTT delivery now has a second SQLite acceptance boundary with one durable row and ACK per named destination. Telemetry, EVENT, property reports, and control results retry independently for every MQTT group; one unavailable mirror cannot duplicate traffic to a healthy group.

• Periodic property reports now carry a stable trace_id and survive broker outages and process restarts. Runtime status exposes aggregate and per-group MQTT queue depth, oldest age, and dead-letter count.

• Pending rows bind to stable mqtt.groups[].name values. Removing or renaming a group while it still has pending deliveries fails startup and requires the operator to restore, drain, or explicitly migrate that destination.

• A driver call that does not return after its deadline is classified as stuck and requests a graceful hard restart. The external supervisor should restart exit code 75; driver.Stop is bounded to five seconds.

• Startup and runtime failures now propagate through app.Run / app.RunWithOptions; legacy Bootstrap wrappers exit non-zero on failure and use exit code 75 after a graceful restart request.

• Driver I/O has a 30-second default deadline. Implement driver.ContextProtocolDriver to cancel transport I/O at its source. Legacy drivers are serialized per device so a stuck call cannot create unbounded goroutines; timed-out writes return the explicit ambiguous code 409.

• Control execution claims and final result/outbox writes are transactional. Interrupted physical operations become final ambiguous results on startup and are never blindly replayed.

• MQTT reconnects retain a stable configured client ID, restore every subscription before becoming healthy, and replace half-open connections after publish or subscribe timeout.

• SQLite acceptance paths use WAL + synchronous=FULL, quarantine malformed outbox rows, enforce configurable database limits, and surface corruption or 95% capacity through readiness.

• HTTP has body, header, request-concurrency, query-size, and read/write/idle limits. Blank service.host now binds only to 127.0.0.1.

• Shutdown cancels workers and control operations, waits for in-flight work, persists final state, then closes MQTT. Worker failures restart with bounded exponential backoff.

What's New In v0.9.9

• Single telemetry outbox path — every telemetry report that passes the configured collection filter is committed to a dedicated SQLite database before MQTT delivery. Recovery rows are drained before newly collected rows.
• Stable delivery metadata — telemetry uses collection time, exact attempt send_at, and boolean is_replayed; dynamic point data is retained as JSON. The removed reliableQueue config and old cache are not migrated.
• EVENT payload fields — the existing data.event_code and data.type fields remain the wire identifiers for event code and lifecycle action.
• Explicit EVENT lifecycle — raise emits phase=start, status=active; clear emits phase=end, status=resolved; pulse emits phase=record, status=recorded.
• Write-ahead EVENT outbox — every event is committed to SQLite before MQTT delivery and acknowledged only after a successful send. The old event queue is intentionally not migrated.
• Graceful shutdown — process signals stop the driver, flush EVENT state, close the event outbox and telemetry outbox, and then close MQTT.

What's New In v0.9.1

• Multi-MQTT broker support — group-based architecture: top-level parallel groups, each group with its own failover broker chain. Per-group dataFormat overrides, per-group heartbeatInterval, and per-broker clientId/username/password. NewPublisher() factory auto-detects groups vs single-broker mode.
• Per-group device status — deviceStatusPublisher creates independent heartbeat goroutines per MQTT group via StartHeartbeatOnly().
• Runtime ops services — configsvc.ConfigService for dynamic config read/write, diff, backup, restore, and hot-reload callback support. logsvc.LogSearcher for log file browsing with pagination and filtering. ops.Restarter for service restarts.
• Service port auto-assign & range search — service.port: 0 now auto-assigns a free port (OS-picked). service.portEnd enables range search: try each port in [port, portEnd], use the first available. Previously port <= 0 disabled the server.
• Telemetry compact format — now includes trace_id, time, and send_at alongside device-name-keyed data.
• bitMerge config — bitMerge field on TelemetryConfig and PropertyConfig for bit-level data merging.
• ClientId lifecycle — an omitted ID is generated once per process; a configured ID remains stable and enables a persistent MQTT session.

EVENT Configuration

EVENT is an SDK capability shared by protocol services. It is enabled only when device.eventDir, device.eventProfile, and (for cloud delivery) eventReport.topic are configured. Existing configurations without these keys continue to run their status, telemetry, property, and command paths unchanged.

device:
  profilesDir: "./configs/profiles"
  devicesDir: "./configs/devices"
  eventDir: "./configs/events"

eventReport:
  topic: "v1/gateway/{productCode}/event/report"
  qos: 1
  retain: false

Each device selects one named EVENT profile with eventProfile. The SDK evaluates telemetry snapshots and standard connection observations, keeps state separately from the durable event outbox, preserves the original event time during replay, and updates only transport metadata such as send_at and is_replayed. EVENT profiles do not contain middleware pipelines or MQTT connection settings. The v0.9.9 wire contract keeps event_code and type, and does not migrate the legacy SQLite event queue.

Telemetry Outbox Configuration

Telemetry is persisted after onChange, deadband, watched-field, and heartbeat filtering. Therefore onChange: false persists every due report, while onChange: true persists only reports selected by that filter strategy. There is no direct MQTT bypass: SQLite commit is the telemetry acceptance boundary. Protocol drivers use ReportAsyncValues; its nil return confirms that commit, so the SDK no longer exposes a process-memory telemetry channel.

telemetryOutbox:
  sqlitePath: "./data/telemetry-outbox.db"
  retentionDays: 0        # 0 = never silently discard pending telemetry
  sendBatchSize: 100
  maxSendRatePerSec: 100
  retryInitialMs: 1000
  retryMaxMs: 30000
  maxDatabaseBytes: 2147483648

telemetryReport:
  topic: "v1/gateway/{productCode}/telemetry/report"
  qos: 1

The outbox database must be different from storage.sqlitePath. Dynamic telemetry points are stored in a JSON column, so each report may have a different data shape. Pending rows are selected by time, id; after startup or network recovery, an ID cutoff keeps the recovery backlog ahead of records collected after recovery. time never changes, send_at is the exact current MQTT attempt time, and is_replayed is true for startup, offline, and failed delivery rows. A new online row waiting behind recovery data remains false.

The outbox is at-least-once: a crash after MQTT acceptance but before SQLite acknowledgement can produce a duplicate, so consumers should deduplicate by trace_id. QoS 1 is the default. The old reliableQueue key and reliable_queue table are neither read nor migrated.

The shared and control SQLite files are also bounded. If both paths point to the same file, the lower configured limit applies:

storage:
  sqlitePath: "./data/runtime.db"
  maxDatabaseBytes: 2147483648

controlStore:
  sqlitePath: "./data/runtime.db"
  retentionDays: 7
  maxDatabaseBytes: 2147483648

At the limit, SQLite rejects new durable writes instead of deleting unsent data. Operators should alert on /api/v1/ready and outbox depth/age before the 95% readiness threshold is reached.

What's New In v0.7.5

• kind: struct — standalone struct type (no index), supports multi-field nested structures: PropertyStructField now has recursive Kind, Fields, MaxItems, IndexStride for nesting
• Nested arrays — kind: array fields support multi-field struct elements and nested array/struct sub-types (max 2 array levels)
• Unified array path — struct_array internally maps to array (multi-field), full backward compatibility with existing configs
• Recursive address calculation — cumulative offset accumulation through nested struct/array levels for correct PLC address generation
• Telemetry struct support — TelemetryConfig.Structs and TelemetryGroup.Structs reuse PropertyStruct, auto-reported structs emit nested JSON objects
• Input-driven writes — property write path processes only fields present in the input, supporting partial updates on nested structures
• Added flattenStructFields, buildStructWriteFields, buildStructReadFields, buildNestedArrayRead, buildNestedArrayWrite, buildStructFieldSelection, BuildTelemetryStructReadRequests
• Added IsScalar(), IsStruct(), IsArray() helper methods on PropertyStructField

What's New In v0.6.7

• property get/set now supports name[index] format for struct arrays: BuildPropertyReadSelectionFromNames accepts wheels[1], wheels[1,3,5], wheels[1-10] to read specific indices
• BuildPropertyWriteRequests accepts wheels[2] single-index writes
• BuildPropertyReadRequests resolves name[index] keys for readback support
• added parseStructNameWithIndices, parseIndexList, deduplicateAndSort, buildStructSelectionForIndices helpers

What's New In v0.6.5

• added shared command and control packages for command descriptors, execution contracts, and control request/result models
• added runtime/command and runtime/control for async command execution, SQLite-backed control state, and resume-on-restart behavior
• expanded ops/http with device-model query APIs, control job list/export/diagnostics APIs, and trace-based property/command result lookup
• added MQTT query handling in runtime/app so device model and control state can be queried over MQTT request/reply topics
• refined property runtime progress, failure context, and helper utilities for telemetry/property reads inside runtime commands

Quick Start

A protocol-specific service typically keeps only the protocol driver and command registration code locally, then boots through the SDK:

package main

import (
    cmdapi "github.com/punk-one/edge-service-sdk/command"
    "github.com/punk-one/edge-service-sdk/runtime/app"
)

func main() {
    registry := cmdapi.NewRegistry()
    // registry.MustRegister(yourCommand)

    app.Bootstrap("edge-service-yourproto", "v0.10.0", newDriver(), registry)
}

The bootstrap flow loads config, initializes auth/MQTT/telemetry outbox/control store, wires property + command + query handlers, starts runtime HTTP APIs, and then supervises protocol workers.

Applications that own their process policy can call app.Run instead. It returns startup/runtime errors and app.ErrRestartRequested; service managers should restart exit code 75. For reliable cancellation, protocol drivers should additionally implement driver.ContextProtocolDriver while retaining the original ProtocolDriver methods for compatibility.

Integration Checklist

1. Keep protocol-specific address parsing, connections, and driver calls in your service repository.
2. Put shared runtime config into configs/config.yaml, devices/*.yaml, and profiles/*.yaml.
3. Register command implementations through command.Registry when the service exposes callable commands.
4. Route telemetry/property/status/command topics through the SDK MQTT publisher instead of re-implementing transport logic.
5. Reuse the SDK HTTP endpoints and control store so HTTP, MQTT, async execution, and result queries stay aligned.

Optional embedded JetStream bus

The MQTT, SQLite, collection, property, and command paths remain authoritative. The embedded JetStream bus is disabled by default and mirrors MQTT traffic only when natsBus.enabled is true. A bus startup or publish failure is reported as a degraded optional dependency and does not stop the existing service path.

The embedded server binds 127.0.0.1 on a random operating-system-selected port. Subjects and the port are SDK conventions and are not application configuration.

Fixed subjects

Logical type	JetStream subject	Direction
telemetry.report	edge.v1.telemetry.report	SDK/process to monitoring or MQTT
property.report	edge.v1.property.report	SDK/process to monitoring or MQTT
property.result	edge.v1.property.result	SDK/process to monitoring or MQTT
command.result	edge.v1.command.result	SDK/process to monitoring or MQTT
property.set	edge.v1.property.set	MQTT/process/NATS to property service
property.get	edge.v1.property.get	MQTT/process/NATS to property service
command.call	edge.v1.command.call.{identifier}	MQTT/process/NATS to command service
event.report	edge.v1.event.report	SDK/process to monitoring or MQTT
status.report	edge.v1.status.report	SDK/process to monitoring or MQTT

JetStream data is the same payload produced for the corresponding MQTT route. Telemetry therefore follows the effective telemetryReport.dataFormat; it is not forced to rule. The SDK adds routing metadata only as NATS headers: Edge-Origin, Edge-Process-Name, Edge-Message-Type, Edge-Data-Format, Edge-Trace-Id, Edge-Product-Code, Edge-Device-Code, and Edge-Hop.

Minimal optional configuration:

natsBus:
  enabled: true

device:
  profilesDir: "./configs/profiles"
  devicesDir: "./configs/devices"
  # Optional; defaults to ./configs/process.
  processDir: "./configs/process"

The default JetStream store is ./data/natsbus. maxAge defaults to 72h and maxBytes defaults to 1 GiB. The optional values can be overridden under natsBus.

Processes are enabled per device in configs/devices/*.yaml:

deviceList:
  - name: device-01
    profileName: profile-01
    productCode: product-01
    processNames:
      - telemetry-alarm
      - external-query

Each distinct referenced Process is started once and receives all fixed SDK message types for only its bound devices. Omitting natsBus preserves the previous runtime behavior; omitting processNames starts no Processes. See docs/process-development-spec.md for the application process API and lifecycle contract.

Package Layout

• config Compatibility-facing configuration model, device/profile loading, normalization helpers, and property/command lookup helpers.
• driver Shared driver contracts, device models, command request/value types, and value type constants.
• property Shared property request/response models.
• command Shared command registry, descriptors, request/response aliases, progress payloads, and command context contracts.
• control Shared control request, metadata, result, and result-code contracts.
• auth Credential bootstrap, token issuance, and request authorization.
• ops/http Runtime health, readiness, auth, model-query, property, command, and control-job HTTP endpoints.
• ops/status Device status tracking and runtime snapshots.
• event Protocol-independent event model, YAML validation, expression evaluation, connection/OEE/alarm state machines, payload selection, and summary windows.
• runtime/event EVENT runtime lifecycle, state-file persistence, event dispatch, and integration with the MQTT event publisher.
• bus / runtime/bus Fixed message contracts plus the optional embedded JetStream server, mirror, durable consumers, and random-port lifecycle.
• process / runtime/process Application handler registry, YAML definitions, independent durable consumers, self-loop prevention, timeout handling, and output publication.
• runtime/app SDK bootstrap facade, runtime assembly, status publishing, and MQTT query wiring.
• runtime/config Runtime-facing config access layer used by bootstrap modules.
• runtime/property Property request execution, MQTT property topic integration, auto-report, and result persistence.
• runtime/command Command execution, async command resume, MQTT command topic integration, and result publishing.
• runtime/control SQLite-backed control job/result store, diagnostics, and export helpers.
• runtime/dependency Runtime dependency checks.
• runtime/scheduler Worker supervision and restart logic.
• telemetry Unified telemetry event model and trace identifiers.
• telemetry/reliable SQLite-first telemetry delivery plus the independent event outbox namespace.
• transport/mqtt MQTT client lifecycle, publishing, subscriptions, and health checks.
• logging Shared logging interface and default implementation.

Runtime APIs And Topics

The SDK now keeps HTTP and MQTT control flows aligned around the same trace_id and control-store model.

HTTP runtime APIs include:

• /api/v1/health
• /api/v1/ready
• /api/v1/runtime/status
• /api/v1/device/model/properties
• /api/v1/device/model/telemetry
• /api/v1/device/model/commands
• /api/v1/device/control/property/get
• /api/v1/device/control/property/set
• /api/v1/device/control/command/call/:identifier
• /api/v1/device/control/jobs
• /api/v1/device/control/jobs/export
• /api/v1/device/control/jobs/diagnostics
• /api/v1/device/control/property/result/:trace_id
• /api/v1/device/control/command/result/:trace_id

MQTT runtime capabilities include telemetry/property/status publishing, property/command request handling, and optional MQTT query request/reply handling for model and control-state lookup.

Current Consumers

• edge-service-s7 boots from runtime/app and reuses the shared runtime, telemetry, MQTT, telemetry outbox, property, command, and control capabilities.
• edge-service-fanuc reuses SDK runtime/config/property/auth/http/status/reliable modules and keeps only protocol-specific driver logic locally.

Documentation

• docs/系统实现参考架构.md Shared runtime layering, package boundaries, startup flow, and integration guidance for new edge-service projects.
• CHANGELOG.md Release notes by SDK version.

Version

This repository version is v0.10.0.