Publish typed values with bounded repeat timing

The shared publisher sends protocol publications through a small caller
transport. Retained values suppress equal payloads, while command
correlation and causation keep their evidence. Changed telemetry stays
non-retained at QoS 0 and supplies the latest five-minute or reconnect
snapshot at QoS 1. Repeats preserve observation times with fresh IDs.

All sends use the caller context and a five-second limit. The publisher
returns background failures to the runner without its own logging.

Issue: 40b314f

diff --git a/doc/driver.rst b/doc/driver.rst
index 33086f1068dc95d0a2ccaba5665506aef68019c7..00b062ab63ce449a51636648e6d2a5e6a3d62cdd 100644 (file)
--- a/doc/driver.rst
+++ b/doc/driver.rst
@@ -14,3 +14,35 @@ SIGTERM and finishes cleanup before exit. No integration or broker runs.
 
 Tests use fake drivers. They do not verify a broker or a physical
 device.
 
 Tests use fake drivers. They do not verify a broker or a physical
 device.

+
+******************
+ Typed publishing
+******************
+
+``Publisher`` takes a ``Transport`` with one ``Publish`` operation on
+``protocol.Publication``. No broker client or topic operation belongs in
+the framework. Typed methods cover descriptor, availability, reported
+values, telemetry, events, and results. Missing message IDs and
+observation times use fresh values. Supplied metadata remains subject to
+protocol validation.
+
+Equal retained payloads are suppressed. Correlation and causation bypass
+suppression. A changed telemetry sample uses QoS 0 without retention.
+Its latest complete value supplies a QoS 1 retained reported snapshot at
+the five-minute repeat or reconnect, not on every change. Repeats keep
+observation times, use fresh IDs, and omit command correlation and
+causation. JSON object key order and whitespace do not change equality.
+Measurement topics are independent. Boolean reported state does not
+repeat unless telemetry marks that point as a measurement. Online
+availability repeats every five minutes. Offline availability,
+descriptors, and Boolean state repeat on reconnect only. Expiring
+values, events, and results are not suppressed or cached.
+
+Run one publisher alongside the integration drivers. Its methods are
+safe for concurrent use and copy caller-owned payloads and timestamp
+pointers. Each send has a five-second timeout derived from the caller
+context. A background send error returns to the runner rather than being
+logged by the publisher. Reconnect refresh requests are nonblocking and
+coalesce while pending. The cache is memory-only. Tests use a fake
+transport and simulated time. They do not verify MQTT acknowledgement or
+a broker reconnect.

diff --git a/driver/publication_value.go b/driver/publication_value.go
new file mode 100644 (file)
index 0000000..2965ce5
--- /dev/null
+++ b/driver/publication_value.go
@@ -0,0 +1,153 @@
+package driver
+
+import (
+       "bytes"
+       "encoding/json"
+       "fmt"
+       "time"
+
+       "onehouse/protocol"
+)
+
+type publicationValue struct {
+       publication protocol.Publication
+       address     protocol.Address
+       metadata    protocol.Metadata
+       payload     any
+       comparison  []byte
+       measurement bool
+       online      bool
+}
+
+func newPublicationValue[T any](address protocol.Address, metadata protocol.Metadata, payload T) (publicationValue, error) {
+       metadata, err := newMetadata(metadata)
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("create publication metadata: %w", err)
+       }
+
+       value := publicationValue{
+               address:  address,
+               metadata: metadata,
+               payload:  payload,
+       }
+       value.publication, err = value.build()
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("build typed publication: %w", err)
+       }
+
+       data, err := json.Marshal(payload)
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("encode comparison payload: %w", err)
+       }
+
+       var normalized any
+
+       decoder := json.NewDecoder(bytes.NewReader(data))
+       decoder.UseNumber()
+
+       err = decoder.Decode(&normalized)
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("decode comparison payload: %w", err)
+       }
+
+       comparison, err := json.Marshal(normalized)
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("normalize comparison payload: %w", err)
+       }
+
+       var copied T
+
+       err = json.Unmarshal(comparison, &copied)
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("copy publication payload: %w", err)
+       }
+
+       value.payload = copied
+       value.comparison = comparison
+
+       switch typed := any(copied).(type) {
+       case protocol.ValuePayload:
+               value.measurement = address.Kind == protocol.Telemetry || string(typed.Value) != "true" && string(typed.Value) != "false"
+       case protocol.AvailabilityPayload:
+               value.online = typed.Status == "online"
+       }
+
+       return value, nil
+}
+
+func newMetadata(metadata protocol.Metadata) (protocol.Metadata, error) {
+       now := time.Now().UTC()
+       if metadata.MessageID == "" {
+               id, err := protocol.NewID(now)
+               if err != nil {
+                       return protocol.Metadata{}, fmt.Errorf("create message ID: %w", err)
+               }
+
+               metadata.MessageID = id
+       }
+
+       if metadata.SentAt == nil {
+               metadata.SentAt = &now
+       } else {
+               timestamp := *metadata.SentAt
+               metadata.SentAt = &timestamp
+       }
+
+       if metadata.ExpiresAt != nil {
+               timestamp := *metadata.ExpiresAt
+               metadata.ExpiresAt = &timestamp
+       }
+
+       return metadata, nil
+}
+
+func (v publicationValue) build() (protocol.Publication, error) {
+       var (
+               publication protocol.Publication
+               err         error
+       )
+
+       switch payload := v.payload.(type) {
+       case protocol.DescriptorPayload:
+               publication, err = protocol.BuildDescriptor(v.address.ID, v.metadata, payload)
+       case protocol.AvailabilityPayload:
+               publication, err = protocol.BuildAvailability(v.address.ID, v.metadata, payload)
+       case protocol.ValuePayload:
+               if v.address.Kind == protocol.Telemetry {
+                       publication, err = protocol.BuildTelemetry(v.address.ID, v.address.Target, v.metadata, payload)
+               } else {
+                       publication, err = protocol.BuildReported(v.address.ID, v.address.Target, v.metadata, payload)
+               }
+       case protocol.EventPayload:
+               publication, err = protocol.BuildEvent(v.address.ID, v.address.Target, v.metadata, payload)
+       case protocol.ResultPayload:
+               publication, err = protocol.BuildResult(v.address.ID, v.address.Target, v.metadata, payload)
+       default:
+               return protocol.Publication{}, fmt.Errorf("unsupported publication payload")
+       }
+
+       if err != nil {
+               return protocol.Publication{}, fmt.Errorf("build %s: %w", v.address.Kind, err)
+       }
+
+       return publication, nil
+}
+
+func (v publicationValue) repeated() (publicationValue, error) {
+       metadata, err := newMetadata(protocol.Metadata{SentAt: v.metadata.SentAt})
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("create repeat metadata: %w", err)
+       }
+
+       v.metadata = metadata
+       if v.address.Kind == protocol.Telemetry {
+               v.address.Kind = protocol.Reported
+       }
+
+       v.publication, err = v.build()
+       if err != nil {
+               return publicationValue{}, fmt.Errorf("build repeated publication: %w", err)
+       }
+
+       return v, nil
+}

diff --git a/driver/publisher.go b/driver/publisher.go
new file mode 100644 (file)
index 0000000..7bd4ce2
--- /dev/null
+++ b/driver/publisher.go
@@ -0,0 +1,284 @@
+package driver
+
+import (
+       "bytes"
+       "context"
+       "fmt"
+       "sync"
+       "time"
+
+       "onehouse/protocol"
+)
+
+const publishTimeout = 5 * time.Second
+
+// Publisher sends typed device messages and owns retained-value suppression.
+// Its methods are safe for concurrent use. Run owns repeats and Refresh
+// requests reconnect snapshots. The caller supplies a valid transport and runs
+// this publisher once alongside the integration drivers.
+type Publisher struct {
+       transport Transport
+       mu        sync.Mutex
+       retained  map[protocol.Address]*retainedValue
+       refresh   chan struct{}
+}
+
+// NewPublisher creates a publisher with no cached state. The caller retains
+// ownership of transport and must keep it available until Run stops.
+func NewPublisher(transport Transport) *Publisher {
+       return &Publisher{
+               transport: transport,
+               retained:  make(map[protocol.Address]*retainedValue),
+               refresh:   make(chan struct{}, 1),
+       }
+}
+
+// Run repeats measurements and online evidence at the contract interval. A send
+// failure returns to the runner. Observation times stay unchanged and repeated
+// messages have fresh IDs without command correlation or causation.
+func (p *Publisher) Run(ctx context.Context) error {
+       ticker := time.NewTicker(250 * time.Millisecond)
+       defer ticker.Stop()
+
+       for {
+               force := false
+
+               select {
+               case <-ctx.Done():
+                       return ctx.Err()
+               case <-p.refresh:
+                       force = true
+               case <-ticker.C:
+               }
+
+               err := p.repeat(ctx, force)
+               if err != nil {
+                       return fmt.Errorf("repeat snapshots: %w", err)
+               }
+       }
+}
+
+// Refresh coalesces a nonblocking request to resend all cached snapshots.
+func (p *Publisher) Refresh() {
+       select {
+       case p.refresh <- struct{}{}:
+       default:
+       }
+}
+
+// Descriptor sends retained metadata unless its value equals the last success.
+// Missing message ID and observation time are filled by the publisher. A
+// supplied correlation or causation always bypasses value suppression.
+func (p *Publisher) Descriptor(ctx context.Context, deviceID string, metadata protocol.Metadata, payload protocol.DescriptorPayload) error {
+       return publishTyped(ctx, p, protocol.Address{
+               Kind: protocol.Descriptor,
+               ID:   deviceID,
+       }, metadata, payload)
+}
+
+// Availability sends reachability and repeats online evidence during Run.
+// Metadata defaults and command correlation follow Descriptor.
+func (p *Publisher) Availability(ctx context.Context, deviceID string, metadata protocol.Metadata, payload protocol.AvailabilityPayload) error {
+       return publishTyped(ctx, p, protocol.Address{
+               Kind: protocol.Availability,
+               ID:   deviceID,
+       }, metadata, payload)
+}
+
+// Reported sends a complete retained value. Non-Boolean values repeat during
+// Run. A Boolean point also repeats after Telemetry marks it as a measurement.
+// Metadata defaults and command correlation follow Descriptor.
+func (p *Publisher) Reported(ctx context.Context, deviceID string, capability string, metadata protocol.Metadata, payload protocol.ValuePayload) error {
+       return publishTyped(ctx, p, protocol.Address{
+               Kind:   protocol.Reported,
+               ID:     deviceID,
+               Target: capability,
+       }, metadata, payload)
+}
+
+// Telemetry sends changed measurements at QoS 0 without retention. The latest
+// observation supplies the retained reported snapshot on repeat or reconnect,
+// not on every telemetry change. Equal samples are suppressed until repeat.
+// Metadata defaults and command correlation follow Descriptor.
+func (p *Publisher) Telemetry(ctx context.Context, deviceID string, capability string, metadata protocol.Metadata, payload protocol.ValuePayload) error {
+       return publishTyped(ctx, p, protocol.Address{
+               Kind:   protocol.Telemetry,
+               ID:     deviceID,
+               Target: capability,
+       }, metadata, payload)
+}
+
+// Event sends each occurrence without suppression. Metadata defaults follow
+// Descriptor. The caller owns event-specific observation and safety rules.
+func (p *Publisher) Event(ctx context.Context, deviceID string, event string, metadata protocol.Metadata, payload protocol.EventPayload) error {
+       return publishTyped(ctx, p, protocol.Address{
+               Kind:   protocol.Event,
+               ID:     deviceID,
+               Target: event,
+       }, metadata, payload)
+}
+
+// Result sends an outcome without suppression. Correlation must equal the
+// command ID. Reuse ledger metadata to replay the same result ID and time.
+// Missing message ID and observation time follow Descriptor defaults.
+func (p *Publisher) Result(ctx context.Context, deviceID string, commandID string, metadata protocol.Metadata, payload protocol.ResultPayload) error {
+       return publishTyped(ctx, p, protocol.Address{
+               Kind:   protocol.Result,
+               ID:     deviceID,
+               Target: commandID,
+       }, metadata, payload)
+}
+
+func (p *Publisher) publish(ctx context.Context, value publicationValue) error {
+       err := ctx.Err()
+       if err != nil {
+               return err
+       }
+
+       if value.metadata.ExpiresAt != nil && !time.Now().Before(*value.metadata.ExpiresAt) {
+               return fmt.Errorf("publication expired")
+       }
+
+       publication := value.publication
+
+       if !publication.Retain && value.address.Kind != protocol.Telemetry || value.metadata.ExpiresAt != nil {
+               err = p.send(ctx, publication)
+               if err != nil {
+                       return fmt.Errorf("send direct publication: %w", err)
+               }
+
+               return nil
+       }
+
+       key := value.address
+       if key.Kind == protocol.Telemetry {
+               key.Kind = protocol.Reported
+       }
+
+       p.mu.Lock()
+
+       entry := p.retained[key]
+       if entry == nil {
+               entry = &retainedValue{retainedAt: time.Now()}
+               p.retained[key] = entry
+       }
+
+       p.mu.Unlock()
+       entry.mu.Lock()
+       defer entry.mu.Unlock()
+
+       entry.latest = value
+       entry.measurement = entry.measurement || value.measurement
+       now := time.Now()
+       equal := bytes.Equal(entry.successful, value.comparison)
+       correlated := value.metadata.CorrelationID != "" || value.metadata.CausationID != ""
+       due := entry.measurement && !now.Before(entry.retainedAt.Add(protocol.MeasurementRepeatInterval)) || value.online && !now.Before(entry.publishedAt.Add(protocol.OnlineRepeatInterval))
+       if equal && !correlated && !due {
+               return nil
+       }
+
+       if entry.measurement && equal && due && !correlated {
+               value, err = value.repeated()
+               if err != nil {
+                       return fmt.Errorf("prepare reported repeat: %w", err)
+               }
+
+               publication = value.publication
+       }
+
+       err = p.send(ctx, publication)
+       if err != nil {
+               return fmt.Errorf("send publication: %w", err)
+       }
+
+       entry.successful = value.comparison
+       entry.publishedAt = now
+       if publication.Retain {
+               entry.retainedAt = now
+       }
+
+       return nil
+}
+
+func (p *Publisher) send(ctx context.Context, publication protocol.Publication) error {
+       err := ctx.Err()
+       if err != nil {
+               return err
+       }
+
+       if publication.ExpiresAt != nil && !time.Now().Before(*publication.ExpiresAt) {
+               return fmt.Errorf("publication expired")
+       }
+
+       ctx, cancel := context.WithTimeout(ctx, publishTimeout)
+       defer cancel()
+
+       err = p.transport.Publish(ctx, publication)
+       if err != nil {
+               return fmt.Errorf("send transport publication: %w", err)
+       }
+
+       return nil
+}
+
+func (p *Publisher) repeat(ctx context.Context, force bool) error {
+       p.mu.Lock()
+
+       entries := make([]*retainedValue, 0, len(p.retained))
+
+       for _, entry := range p.retained {
+               entries = append(entries, entry)
+       }
+
+       p.mu.Unlock()
+
+       for _, entry := range entries {
+               err := p.repeatEntry(ctx, entry, force)
+               if err != nil {
+                       return fmt.Errorf("repeat retained value: %w", err)
+               }
+       }
+
+       return nil
+}
+
+func (p *Publisher) repeatEntry(ctx context.Context, entry *retainedValue, force bool) error {
+       entry.mu.Lock()
+       defer entry.mu.Unlock()
+
+       now := time.Now()
+       due := entry.measurement && (!now.Before(entry.retainedAt.Add(protocol.MeasurementRepeatInterval)) || !bytes.Equal(entry.successful, entry.latest.comparison)) || entry.latest.online && !now.Before(entry.publishedAt.Add(protocol.OnlineRepeatInterval))
+       if !force && !due {
+               return nil
+       }
+
+       value, err := entry.latest.repeated()
+       if err != nil {
+               return fmt.Errorf("prepare snapshot: %w", err)
+       }
+
+       err = p.send(ctx, value.publication)
+       if err != nil {
+               return fmt.Errorf("send snapshot: %w", err)
+       }
+
+       entry.successful = value.comparison
+       entry.publishedAt = now
+       entry.retainedAt = now
+
+       return nil
+}
+
+func publishTyped[T any](ctx context.Context, publisher *Publisher, address protocol.Address, metadata protocol.Metadata, payload T) error {
+       value, err := newPublicationValue(address, metadata, payload)
+       if err != nil {
+               return fmt.Errorf("prepare typed publication: %w", err)
+       }
+
+       err = publisher.publish(ctx, value)
+       if err != nil {
+               return fmt.Errorf("publish %s: %w", address.Kind, err)
+       }
+
+       return nil
+}

diff --git a/driver/publisher_test.go b/driver/publisher_test.go
new file mode 100644 (file)
index 0000000..d1135b5
--- /dev/null
+++ b/driver/publisher_test.go
@@ -0,0 +1,351 @@
+package driver
+
+import (
+       "context"
+       "encoding/json"
+       "errors"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/protocol"
+)
+
+func TestPublisherSuppressesEqualRetainedValuesAndKeepsCommandEvidence(t *testing.T) {
+       transport := &recordingTransport{}
+       publisher := NewPublisher(transport)
+       metadata := protocol.Metadata{}
+       descriptor := protocol.DescriptorPayload{
+               Manufacturer: "example",
+               Model:        "switch",
+               Firmware:     "1",
+               Capabilities: []protocol.Capability{{
+                       ID:           "relay",
+                       Type:         "switch",
+                       ReportsState: true,
+               }},
+       }
+
+       for range 2 {
+               require.NoError(t, publisher.Descriptor(t.Context(), testDeviceID, metadata, descriptor))
+               require.NoError(t, publisher.Availability(t.Context(), testDeviceID, metadata, protocol.AvailabilityPayload{Status: "offline"}))
+               require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", metadata, protocol.ValuePayload{Value: json.RawMessage("false")}))
+       }
+
+       require.Len(t, transport.messages(), 3)
+       require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
+       require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CausationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
+       require.Len(t, transport.messages(), 5)
+
+       for _, publication := range transport.messages() {
+               require.Equal(t, byte(1), publication.QoS)
+               require.True(t, publication.Retain)
+       }
+}
+
+func TestPublisherRepeatsLatestMeasurementWithoutMovingDeadline(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               transport := &recordingTransport{}
+               publisher := NewPublisher(transport)
+               finished := make(chan error, 1)
+
+               require.NoError(t, publisher.Reported(ctx, testDeviceID, "weather", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)}))
+
+               first := transport.messages()[0]
+               go func() {
+                       finished <- publisher.Run(ctx)
+               }()
+
+               for range 299 {
+                       time.Sleep(time.Second)
+                       require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "weather", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage(`{ "humidity":50,"temperature":20 }`)}))
+                       synctest.Wait()
+                       require.Len(t, transport.messages(), 1)
+               }
+
+               time.Sleep(time.Second)
+               synctest.Wait()
+
+               packets := transport.messages()
+               require.Len(t, packets, 2)
+
+               repeated, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+               require.NoError(t, err)
+
+               original, err := protocol.ParseReported(first.Topic, first.Data, first.Retain)
+               require.NoError(t, err)
+               require.NotEqual(t, original.Metadata.MessageID, repeated.Metadata.MessageID)
+               require.Equal(t, time.Now().Add(-time.Second).UTC(), *repeated.Metadata.SentAt)
+               require.Empty(t, repeated.Metadata.CorrelationID)
+               require.Empty(t, repeated.Metadata.CausationID)
+               require.JSONEq(t, `{"temperature":20,"humidity":50}`, string(repeated.Payload.Value))
+               cancel()
+               require.ErrorIs(t, <-finished, context.Canceled)
+       })
+}
+
+func TestPublisherTelemetryKeepsSnapshotOnScheduleAndReconnect(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               transport := &recordingTransport{}
+               publisher := NewPublisher(transport)
+               finished := make(chan error, 1)
+
+               require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("10")}))
+
+               go func() {
+                       finished <- publisher.Run(ctx)
+               }()
+
+               time.Sleep(4 * time.Minute)
+               require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
+               require.NoError(t, publisher.Telemetry(ctx, testOtherDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("30")}))
+               synctest.Wait()
+
+               packets := transport.messages()
+               require.Len(t, packets, 3)
+
+               for _, packet := range packets {
+                       require.Equal(t, byte(0), packet.QoS)
+                       require.False(t, packet.Retain)
+               }
+
+               time.Sleep(time.Minute)
+               synctest.Wait()
+
+               packets = transport.messages()
+               require.Len(t, packets, 4)
+
+               snapshot, err := protocol.ParseReported(packets[3].Topic, packets[3].Data, packets[3].Retain)
+               require.NoError(t, err)
+               require.JSONEq(t, "20", string(snapshot.Payload.Value))
+               require.Equal(t, byte(1), packets[3].QoS)
+               require.Equal(t, time.Now().Add(-time.Minute).UTC(), *snapshot.Metadata.SentAt)
+               publisher.Refresh()
+               synctest.Wait()
+               require.Len(t, transport.messages(), 6)
+               cancel()
+               require.ErrorIs(t, <-finished, context.Canceled)
+       })
+}
+
+func TestPublisherRepeatsOnlineButNotBooleanOrOffline(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               transport := &recordingTransport{}
+               publisher := NewPublisher(transport)
+               finished := make(chan error, 1)
+
+               require.NoError(t, publisher.Availability(ctx, testDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "online"}))
+               require.NoError(t, publisher.Availability(ctx, testOtherDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"}))
+               require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+
+               go func() {
+                       finished <- publisher.Run(ctx)
+               }()
+
+               time.Sleep(5 * time.Minute)
+               synctest.Wait()
+
+               packets := transport.messages()
+               require.Len(t, packets, 4)
+
+               online, err := protocol.ParseAvailability(packets[3].Topic, packets[3].Data, packets[3].Retain)
+               require.NoError(t, err)
+               require.Equal(t, "online", online.Payload.Status)
+               publisher.Refresh()
+               synctest.Wait()
+               require.Len(t, transport.messages(), 7)
+               cancel()
+               require.ErrorIs(t, <-finished, context.Canceled)
+       })
+}
+
+func TestPublisherDoesNotSuppressEventsResultsOrExpiringValues(t *testing.T) {
+       transport := &recordingTransport{}
+       publisher := NewPublisher(transport)
+       expiry := time.Now().UTC().Add(time.Minute)
+
+       for range 2 {
+               require.NoError(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{"action": json.RawMessage(`"pressed"`)}))
+               require.NoError(t, publisher.Result(t.Context(), testDeviceID, testCommandID, protocol.Metadata{CorrelationID: testCommandID}, protocol.ResultPayload{Status: "completed"}))
+               require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{ExpiresAt: &expiry}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+       }
+
+       require.Len(t, transport.messages(), 6)
+       require.NoError(t, publisher.repeat(t.Context(), true))
+       require.Len(t, transport.messages(), 6)
+}
+
+func TestPublisherKeepsFailedLatestSampleAndReturnsRepeatFailures(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               transport := &recordingTransport{}
+               publisher := NewPublisher(transport)
+               require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
+
+               failure := errors.New("broker unavailable")
+               transport.setError(failure)
+               require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("21")}), failure)
+               require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("22")}), failure)
+               publisher.Refresh()
+               require.ErrorIs(t, publisher.Run(t.Context()), failure)
+               transport.setError(nil)
+               require.NoError(t, publisher.repeat(t.Context(), false))
+
+               packets := transport.messages()
+               last := packets[len(packets)-1]
+               snapshot, err := protocol.ParseReported(last.Topic, last.Data, last.Retain)
+               require.NoError(t, err)
+               require.JSONEq(t, "22", string(snapshot.Payload.Value))
+       })
+}
+
+func TestPublisherValidatesBeforeSendingAndHonorsCancellation(t *testing.T) {
+       transport := &recordingTransport{}
+       publisher := NewPublisher(transport)
+       require.Error(t, publisher.Reported(t.Context(), "invalid", "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
+       require.Error(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("invalid")}))
+
+       past := time.Now().UTC().Add(-time.Second)
+       require.ErrorContains(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{ExpiresAt: &past}, protocol.EventPayload{}), "expired")
+
+       ctx, cancel := context.WithCancel(t.Context())
+       defer cancel()
+
+       cancel()
+       require.ErrorIs(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}), context.Canceled)
+       require.Empty(t, transport.messages())
+}
+
+func TestPublisherSendTimeoutAndCallerCancellation(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               publisher := NewPublisher(&waitingTransport{})
+               err := publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
+               require.ErrorIs(t, err, context.DeadlineExceeded)
+
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               finished := make(chan error, 1)
+               go func() {
+                       finished <- publisher.Event(ctx, testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
+               }()
+
+               synctest.Wait()
+               cancel()
+               require.ErrorIs(t, <-finished, context.Canceled)
+       })
+}
+
+func TestPublisherCopiesSnapshotsAndAcceptsTypedNull(t *testing.T) {
+       transport := &recordingTransport{}
+       publisher := NewPublisher(transport)
+       now := time.Now().UTC()
+       metadata := protocol.Metadata{SentAt: &now}
+       value := json.RawMessage(`{"power":10}`)
+       require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{Value: value}))
+
+       observation := now
+       now = now.Add(time.Hour)
+
+       copy(value, `{"power":99}`)
+       publisher.Refresh()
+       publisher.Refresh()
+       require.NoError(t, publisher.repeat(t.Context(), true))
+
+       packets := transport.messages()
+       require.Len(t, packets, 2)
+
+       snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+       require.NoError(t, err)
+       require.JSONEq(t, `{"power":10}`, string(snapshot.Payload.Value))
+       require.Equal(t, observation, *snapshot.Metadata.SentAt)
+       require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{}))
+}
+
+func TestPublisherEqualTelemetryAtRepeatTimeSendsRetainedSnapshot(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               transport := &recordingTransport{}
+               publisher := NewPublisher(transport)
+               value := protocol.ValuePayload{Value: json.RawMessage("null")}
+               require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
+               time.Sleep(protocol.MeasurementRepeatInterval)
+               require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
+
+               packets := transport.messages()
+               require.Len(t, packets, 2)
+               require.True(t, packets[1].Retain)
+
+               snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+               require.NoError(t, err)
+               require.JSONEq(t, "null", string(snapshot.Payload.Value))
+       })
+}
+
+func TestPublisherPreservesDirectMeasurementEvidenceAtRepeatDeadline(t *testing.T) {
+       testCases := []struct {
+               name          string
+               correlationID string
+               causationID   string
+       }{
+               {
+                       name:          "correlation keeps caller identity",
+                       correlationID: testCommandID,
+               },
+               {
+                       name:        "causation keeps caller identity",
+                       causationID: testCommandID,
+               },
+               {
+                       name:          "correlation and causation keep caller identity",
+                       correlationID: testCommandID,
+                       causationID:   testOtherDeviceID,
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               transport := &recordingTransport{}
+                               publisher := NewPublisher(transport)
+                               value := protocol.ValuePayload{Value: json.RawMessage("20")}
+                               require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, value))
+                               time.Sleep(protocol.MeasurementRepeatInterval)
+
+                               observedAt := time.Now().UTC()
+                               metadata := protocol.Metadata{
+                                       MessageID:     testOtherDeviceID,
+                                       SentAt:        &observedAt,
+                                       CorrelationID: tc.correlationID,
+                                       CausationID:   tc.causationID,
+                               }
+                               require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", metadata, value))
+
+                               packets := transport.messages()
+                               require.Len(t, packets, 2)
+
+                               message, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
+                               require.NoError(t, err)
+                               require.Equal(t, metadata, message.Metadata)
+                               require.Equal(t, value, message.Payload)
+                       })
+               })
+       }
+}
+
+type waitingTransport struct{}
+
+func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {
+       <-ctx.Done()
+
+       return ctx.Err()
+}

diff --git a/driver/retained_value.go b/driver/retained_value.go
new file mode 100644 (file)
index 0000000..3b1ce25
--- /dev/null
+++ b/driver/retained_value.go
@@ -0,0 +1,15 @@
+package driver
+
+import (
+       "sync"
+       "time"
+)
+
+type retainedValue struct {
+       mu          sync.Mutex
+       latest      publicationValue
+       successful  []byte
+       publishedAt time.Time
+       retainedAt  time.Time
+       measurement bool
+}

diff --git a/driver/runner.go b/driver/runner.go
index e918416a7a4e02b585c84c3c82fade7b9ad8a00b..287c333cb75cb7706cf6a82720bc23fd24ab5dc9 100644 (file)
--- a/driver/runner.go
+++ b/driver/runner.go
@@ -31,7 +31,10 @@ func Run(ctx context.Context, drivers []Driver, connected <-chan struct{}) error
 
        for index, integration := range drivers {
                workers.Go(func() {
 
        for index, integration := range drivers {
                workers.Go(func() {

-                       results <- driverResult{index: index, err: integration.Run(ctx)}
+                       results <- driverResult{
+                               index: index,
+                               err:   integration.Run(ctx),
+                       }

                })
        }
 
                })
        }
 

diff --git a/driver/runner_test.go b/driver/runner_test.go
index 2ece6cd98fbeddd7fcc6b0047cde627ff36a6e90..c1c32aa1d97872cf54a613328a8f60acec843328 100644 (file)
--- a/driver/runner_test.go
+++ b/driver/runner_test.go
@@ -20,8 +20,14 @@ func TestRunRefreshesDriversAndWaitsForCleanup(t *testing.T) {
                ctx, cancel := context.WithCancel(t.Context())
                defer cancel()
 
                ctx, cancel := context.WithCancel(t.Context())
                defer cancel()
 

-               first := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
-               second := &fakeDriver{stopped: make(chan struct{}), refreshed: make(chan struct{}, 1)}
+               first := &fakeDriver{
+                       stopped:   make(chan struct{}),
+                       refreshed: make(chan struct{}, 1),
+               }
+               second := &fakeDriver{
+                       stopped:   make(chan struct{}),
+                       refreshed: make(chan struct{}, 1),
+               }

                connected := make(chan struct{}, 1)
                finished := make(chan error, 1)
                go func() {
                connected := make(chan struct{}, 1)
                finished := make(chan error, 1)
                go func() {


@@ -53,18 +59,67 @@ func TestRunClassifiesStopsAndJoinedFailures(t *testing.T) {
                canceled bool
                failure  bool
        }{
                canceled bool
                failure  bool
        }{

-               {name: "normal completion", err: nil},
-               {name: "cancellation is a stop", err: context.Canceled, canceled: true},
-               {name: "closed transport is a stop", err: net.ErrClosed, canceled: true},
-               {name: "closed pipe is a stop", err: io.ErrClosedPipe, canceled: true},
-               {name: "broken pipe is a stop", err: syscall.EPIPE, canceled: true},
-               {name: "reset connection is a stop", err: syscall.ECONNRESET, canceled: true},
-               {name: "wrapped stop stays a stop", err: fmt.Errorf("read: %w", net.ErrClosed), canceled: true},
-               {name: "joined stops stay stops", err: errors.Join(context.Canceled, net.ErrClosed), canceled: true},
-               {name: "joined failure stays a failure", err: errors.Join(context.Canceled, failure), canceled: true, failure: true},
-               {name: "request timeout stays a failure", err: context.DeadlineExceeded, failure: true},
-               {name: "active transport close is a failure", err: net.ErrClosed, failure: true},
-               {name: "independent failure stays a failure", err: failure, canceled: true, failure: true},
+               {
+                       name: "normal completion",
+                       err:  nil,
+               },
+               {
+                       name:     "cancellation is a stop",
+                       err:      context.Canceled,
+                       canceled: true,
+               },
+               {
+                       name:     "closed transport is a stop",
+                       err:      net.ErrClosed,
+                       canceled: true,
+               },
+               {
+                       name:     "closed pipe is a stop",
+                       err:      io.ErrClosedPipe,
+                       canceled: true,
+               },
+               {
+                       name:     "broken pipe is a stop",
+                       err:      syscall.EPIPE,
+                       canceled: true,
+               },
+               {
+                       name:     "reset connection is a stop",
+                       err:      syscall.ECONNRESET,
+                       canceled: true,
+               },
+               {
+                       name:     "wrapped stop stays a stop",
+                       err:      fmt.Errorf("read: %w", net.ErrClosed),
+                       canceled: true,
+               },
+               {
+                       name:     "joined stops stay stops",
+                       err:      errors.Join(context.Canceled, net.ErrClosed),
+                       canceled: true,
+               },
+               {
+                       name:     "joined failure stays a failure",
+                       err:      errors.Join(context.Canceled, failure),
+                       canceled: true,
+                       failure:  true,
+               },
+               {
+                       name:    "request timeout stays a failure",
+                       err:     context.DeadlineExceeded,
+                       failure: true,
+               },
+               {
+                       name:    "active transport close is a failure",
+                       err:     net.ErrClosed,
+                       failure: true,
+               },
+               {
+                       name:     "independent failure stays a failure",
+                       err:      failure,
+                       canceled: true,
+                       failure:  true,
+               },

        }
 
        for _, tc := range testCases {
        }
 
        for _, tc := range testCases {

diff --git a/driver/transport.go b/driver/transport.go
new file mode 100644 (file)
index 0000000..34e1ea4
--- /dev/null
+++ b/driver/transport.go
@@ -0,0 +1,13 @@
+package driver
+
+import (
+       "context"
+
+       "onehouse/protocol"
+)
+
+// Transport sends a complete protocol publication. The caller owns the
+// connection. Publish must stop when ctx stops and must not mutate publication.
+type Transport interface {
+       Publish(ctx context.Context, publication protocol.Publication) error
+}

diff --git a/driver/transport_helper_test.go b/driver/transport_helper_test.go
new file mode 100644 (file)
index 0000000..92f43e5
--- /dev/null
+++ b/driver/transport_helper_test.go
@@ -0,0 +1,43 @@
+package driver
+
+import (
+       "context"
+       "sync"
+
+       "onehouse/protocol"
+)
+
+const (
+       testDeviceID      = "01K2K7M5TZ3J7W4Q9X8N6P2RVA"
+       testOtherDeviceID = "01K2K7M5TZ3J7W4Q9X8N6P2RVB"
+       testCommandID     = "01K2K7M5TZ3J7W4Q9X8N6P2RVF"
+)
+
+type recordingTransport struct {
+       mu           sync.Mutex
+       publications []protocol.Publication
+       err          error
+}
+
+func (p *recordingTransport) Publish(_ context.Context, publication protocol.Publication) error {
+       p.mu.Lock()
+       defer p.mu.Unlock()
+
+       p.publications = append(p.publications, publication)
+
+       return p.err
+}
+
+func (p *recordingTransport) messages() []protocol.Publication {
+       p.mu.Lock()
+       defer p.mu.Unlock()
+
+       return append([]protocol.Publication(nil), p.publications...)
+}
+
+func (p *recordingTransport) setError(err error) {
+       p.mu.Lock()
+       defer p.mu.Unlock()
+
+       p.err = err
+}