From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 22:21:22 +0000 (-0300) Subject: Publish typed values with bounded repeat timing X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/a5486293282c47e6bca115c05ddde6e39ba4930c?ds=inline;hp=2015860819c6228bf5ec708a48e5ca9b7c04e27d 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 33086f1..00b062a 100644 --- 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. + +****************** + 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 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 = ×tamp + } + + if metadata.ExpiresAt != nil { + timestamp := *metadata.ExpiresAt + metadata.ExpiresAt = ×tamp + } + + 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 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 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 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 e918416..287c333 100644 --- 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() { - 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 2ece6cd..c1c32aa 100644 --- 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() - 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() { @@ -53,18 +59,67 @@ func TestRunClassifiesStopsAndJoinedFailures(t *testing.T) { 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 { diff --git a/driver/transport.go b/driver/transport.go new file mode 100644 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 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 +}