219 lines
7.7 KiB
Go
219 lines
7.7 KiB
Go
// Package backend observes contracts and correlates only this process's own commands.
|
|
package backend
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"robot-market/test-mqtt-5/internal/contracts"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
"robot-market/test-mqtt-5/internal/mqtt"
|
|
|
|
"github.com/eclipse/paho.golang/paho"
|
|
)
|
|
|
|
type Result struct {
|
|
Ack *contracts.Ack
|
|
Err error
|
|
}
|
|
type pending struct {
|
|
command contracts.Command
|
|
correlation []byte
|
|
started, expires time.Time
|
|
result chan Result
|
|
terminal bool
|
|
accepted bool
|
|
}
|
|
type Backend struct {
|
|
Client *mqtt.Client
|
|
mu sync.Mutex
|
|
pending map[string]*pending
|
|
receipts chan *paho.Publish
|
|
PublishReceipt func(context.Context, *paho.Publish) error
|
|
}
|
|
|
|
func New(client *mqtt.Client) *Backend {
|
|
return &Backend{Client: client, pending: make(map[string]*pending), receipts: make(chan *paho.Publish, client.Config.Limits.InboundQueue), PublishReceipt: client.PublishRaw}
|
|
}
|
|
func ID() string {
|
|
var data [16]byte
|
|
if _, err := rand.Read(data[:]); err != nil {
|
|
panic("system random source unavailable")
|
|
}
|
|
return hex.EncodeToString(data[:])
|
|
}
|
|
func Timestamp(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) }
|
|
func NewCommand(device, name, id string, lifetime time.Duration) contracts.Command {
|
|
now := time.Now().UTC()
|
|
return contracts.Command{SchemaVersion: 1, DeviceID: device, RequestID: id, Command: name, IssuedAt: Timestamp(now), ExpiresAt: Timestamp(now.Add(lifetime))}
|
|
}
|
|
func (b *Backend) register(cmd contracts.Command) (*pending, error) {
|
|
if err := cmd.Validate(b.Client.Config.Topics.DeviceID); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := cmd.CheckDeadline(time.Now()); err != nil {
|
|
return nil, err
|
|
}
|
|
expires, _ := time.Parse(time.RFC3339Nano, cmd.ExpiresAt)
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if len(b.pending) >= b.Client.Config.Limits.PendingCommands {
|
|
return nil, errors.New("pending command limit reached")
|
|
}
|
|
if _, ok := b.pending[cmd.RequestID]; ok {
|
|
return nil, errors.New("request ID already pending")
|
|
}
|
|
p := &pending{command: cmd, correlation: []byte(cmd.RequestID), started: time.Now(), expires: expires, result: make(chan Result, 1)}
|
|
b.pending[cmd.RequestID] = p
|
|
return p, nil
|
|
}
|
|
func (b *Backend) Command(ctx context.Context, cmd contracts.Command) (*contracts.Ack, error) {
|
|
if !b.Client.Ready() {
|
|
return nil, mqtt.ErrNotReady
|
|
}
|
|
p, err := b.register(cmd)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { b.mu.Lock(); delete(b.pending, cmd.RequestID); b.mu.Unlock() }()
|
|
packet := mqtt.Packet(mqtt.Topic(b.Client.Config, "commands"), 1, false, cmd)
|
|
packet.Properties.ResponseTopic = mqtt.Topic(b.Client.Config, "acks")
|
|
packet.Properties.CorrelationData = append([]byte(nil), p.correlation...)
|
|
expiry := uint32(max(1, int(time.Until(p.expires).Seconds()+0.999)))
|
|
packet.Properties.MessageExpiry = &expiry
|
|
// Register before publishing: accepted/completed can arrive before Publish returns.
|
|
publishCtx, stopPublish := context.WithDeadline(ctx, p.expires)
|
|
err = b.Client.Publish(publishCtx, packet)
|
|
stopPublish()
|
|
if err != nil {
|
|
// An authenticated correlated terminal ACK is execution evidence even if
|
|
// transport confirmation was lost. Never replace it with a generic timeout.
|
|
select {
|
|
case result := <-p.result:
|
|
return result.Ack, result.Err
|
|
default:
|
|
}
|
|
if !time.Now().Before(p.expires) {
|
|
b.Client.Record(events.CommandTimeout, events.Internal, packet, cmd.RequestID, "deadline elapsed during publish; execution unknown")
|
|
}
|
|
return nil, err
|
|
}
|
|
timer := time.NewTimer(time.Until(p.expires))
|
|
defer timer.Stop()
|
|
select {
|
|
case result := <-p.result:
|
|
return result.Ack, result.Err
|
|
case <-ctx.Done():
|
|
return nil, errors.New("command interrupted; execution unknown")
|
|
case <-timer.C:
|
|
b.Client.Record(events.CommandTimeout, events.Internal, packet, cmd.RequestID, "deadline elapsed; execution unknown")
|
|
return nil, errors.New("command deadline elapsed; execution unknown")
|
|
}
|
|
}
|
|
func (b *Backend) Disconnect() {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for id, p := range b.pending {
|
|
if !p.terminal {
|
|
p.terminal = true
|
|
p.result <- Result{Err: errors.New("connection lost; execution unknown")}
|
|
b.Client.Record(events.Kind("command_unknown"), events.Internal, nil, id, "connection lost; no automatic retry")
|
|
}
|
|
}
|
|
}
|
|
func (b *Backend) Handle(packet *paho.Publish) {
|
|
if b.Client.Config.Topics.ObserveAll {
|
|
b.Client.Record(events.MessageReceived, events.In, packet, "", "raw observation; contract validation disabled")
|
|
if packet.Topic == b.Client.Config.Topics.Namespace+"/scenarios/results" {
|
|
var result events.ScenarioResult
|
|
if json.Unmarshal(packet.Payload, &result) == nil && result.Feature != "" && result.Environment != "" {
|
|
b.Client.Events.Add(events.Event{Kind: "scenario_result", Direction: events.In, Topic: packet.Topic, Scenario: &result, Outcome: "reported test evidence; not independently verified hardware"})
|
|
}
|
|
}
|
|
b.enqueueReceipt(packet)
|
|
return
|
|
}
|
|
device, kind := mqtt.Identity(b.Client.Config, packet.Topic)
|
|
msg, err := contracts.Decode(kind, packet.Payload, device)
|
|
if err != nil {
|
|
b.Client.Record(events.ValidationFailed, events.In, packet, "", "invalid contract: "+err.Error())
|
|
return
|
|
}
|
|
request := ""
|
|
switch v := msg.(type) {
|
|
case *contracts.Ack:
|
|
request = v.RequestID
|
|
case *contracts.Command:
|
|
request = v.RequestID
|
|
}
|
|
b.Client.Record(events.MessageReceived, events.In, packet, request, "")
|
|
ack, ok := msg.(*contracts.Ack)
|
|
if !ok {
|
|
return
|
|
}
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
p, ok := b.pending[ack.RequestID]
|
|
if !ok || p.terminal || ack.DeviceID != p.command.DeviceID {
|
|
b.Client.Record(events.LateAck, events.In, packet, ack.RequestID, "no active own request; latency not measured")
|
|
return
|
|
}
|
|
if !time.Now().Before(p.expires) {
|
|
b.Client.Record(events.LateAck, events.In, packet, ack.RequestID, "ack arrived after deadline")
|
|
return
|
|
}
|
|
if packet.Retain || packet.Properties == nil || !bytes.Equal(packet.Properties.CorrelationData, p.correlation) || packet.Properties.ResponseTopic != "" {
|
|
b.Client.Record(events.ValidationFailed, events.In, packet, ack.RequestID, "invalid native ACK correlation or retained ACK")
|
|
return
|
|
}
|
|
if ack.Stage == "completed" && ((p.command.Command == "ping" && ack.Result.Pong == nil) || (p.command.Command == "get_version" && ack.Result.Version == "")) {
|
|
b.Client.Record(events.ValidationFailed, events.In, packet, ack.RequestID, "result does not match command")
|
|
return
|
|
}
|
|
if ack.Stage == "accepted" {
|
|
if !p.accepted {
|
|
p.accepted = true
|
|
b.Client.Record(events.CommandAccepted, events.In, packet, ack.RequestID, "")
|
|
}
|
|
return
|
|
}
|
|
p.terminal = true
|
|
elapsed := time.Since(p.started).Milliseconds()
|
|
b.Client.Events.Add(events.Event{Kind: events.CommandCompleted, Direction: events.Internal, RequestID: ack.RequestID, DeviceID: ack.DeviceID, Outcome: ack.Stage, LatencyMS: &elapsed})
|
|
if !*ack.OK {
|
|
p.result <- Result{Ack: ack, Err: errors.New("device rejected or failed command")}
|
|
} else {
|
|
p.result <- Result{Ack: ack}
|
|
}
|
|
}
|
|
func (b *Backend) Serve(ctx context.Context) error {
|
|
defer b.Disconnect()
|
|
workerCtx, cancelWorker := context.WithCancel(ctx)
|
|
done := make(chan struct{})
|
|
go func() { defer close(done); b.receiptWorker(workerCtx) }()
|
|
defer func() { cancelWorker(); <-done }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case err := <-b.Client.Fatal():
|
|
return err
|
|
case <-b.Client.Down:
|
|
b.Disconnect()
|
|
case packet := <-b.Client.Incoming:
|
|
b.Handle(packet)
|
|
}
|
|
}
|
|
}
|
|
func (b *Backend) Summary() string {
|
|
_, counts := b.Client.Events.Snapshot()
|
|
return fmt.Sprintf("received=%d; completed=%d; timeouts=%d; late_acks=%d", counts[events.MessageReceived], counts[events.CommandCompleted], counts[events.CommandTimeout], counts[events.LateAck])
|
|
}
|