From 1f2ea26932b468f7631ed440e4f52a8b6e427027 Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:06:09 -0600 Subject: [PATCH 1/4] feat: add output_drives, one row per output-device drive and its result alarm_events says a rule fired for a read. It does not say which devices were driven or whether the drive reached them, so a drive to an offline reader looks the same as one that sounded the horn. output_drives is the delivery record: device, edge, cause, ok/error with the error text, and latency. InsertAlarmEvent now returns the row id so the drives that deliver an alarm can name it. Refs: TRA-1314 Co-Authored-By: Claude Opus 5.5 --- backend/internal/geofence/engine.go | 4 +- backend/internal/geofence/engine_test.go | 6 +- backend/internal/storage/alarm_events.go | 16 +- .../storage/alarm_events_integration_test.go | 7 +- backend/internal/storage/output_drives.go | 135 +++++++++++++++++ .../storage/output_drives_integration_test.go | 143 ++++++++++++++++++ .../internal/storage/output_drives_test.go | 21 +++ .../migrations/000055_output_drives.down.sql | 4 + .../migrations/000055_output_drives.up.sql | 58 +++++++ backend/migrations/checksums.txt | 2 + 10 files changed, 382 insertions(+), 14 deletions(-) create mode 100644 backend/internal/storage/output_drives.go create mode 100644 backend/internal/storage/output_drives_integration_test.go create mode 100644 backend/internal/storage/output_drives_test.go create mode 100644 backend/migrations/000055_output_drives.down.sql create mode 100644 backend/migrations/000055_output_drives.up.sql diff --git a/backend/internal/geofence/engine.go b/backend/internal/geofence/engine.go index d90522242..faf718a9d 100644 --- a/backend/internal/geofence/engine.go +++ b/backend/internal/geofence/engine.go @@ -28,7 +28,7 @@ import ( // engineStore is the storage surface the engine needs; *storage.Storage // satisfies it. Narrowed so unit tests can inject a fake. type engineStore interface { - InsertAlarmEvent(ctx context.Context, orgID int, ev storage.AlarmEventRow) error + InsertAlarmEvent(ctx context.Context, orgID int, ev storage.AlarmEventRow) (int64, error) ListOutputDevicesForLocation(ctx context.Context, orgID, locationID int) ([]outputdevice.OutputDevice, error) GetOrgGeofenceDefaults(ctx context.Context, orgID int) (organization.GeofenceDefaults, error) } @@ -246,7 +246,7 @@ func (e *Engine) drive(ctx context.Context, orgID int, dev outputdevice.OutputDe // one device on, and logs/metrics the fire. Best-effort: a write error never // blocks ingestion. func (e *Engine) recordFire(ctx context.Context, orgID int, tagScanID int64, receivedAt time.Time, rd storage.ResolvedRead) { - if err := e.store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ + if _, err := e.store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ AssetID: rd.AssetID, ScanPointID: rd.ScanPointID, LocationID: rd.LocationID, diff --git a/backend/internal/geofence/engine_test.go b/backend/internal/geofence/engine_test.go index 1cfbd24a6..484d91919 100644 --- a/backend/internal/geofence/engine_test.go +++ b/backend/internal/geofence/engine_test.go @@ -33,12 +33,12 @@ func (s *fakeStore) GetOrgGeofenceDefaults(_ context.Context, _ int) (organizati return s.orgDefaults, nil } -func (s *fakeStore) InsertAlarmEvent(_ context.Context, _ int, ev storage.AlarmEventRow) error { +func (s *fakeStore) InsertAlarmEvent(_ context.Context, _ int, ev storage.AlarmEventRow) (int64, error) { if s.insertErr != nil { - return s.insertErr + return 0, s.insertErr } s.rows = append(s.rows, ev) - return nil + return int64(len(s.rows)), nil } func (s *fakeStore) ListOutputDevicesForLocation(_ context.Context, _, _ int) ([]outputdevice.OutputDevice, error) { diff --git a/backend/internal/storage/alarm_events.go b/backend/internal/storage/alarm_events.go index 9e19fd314..d5c99ab5f 100644 --- a/backend/internal/storage/alarm_events.go +++ b/backend/internal/storage/alarm_events.go @@ -21,20 +21,24 @@ type AlarmEventRow struct { } // InsertAlarmEvent appends a geofence alarm to trakrf.alarm_events under org -// context (RLS). It is called best-effort from the geofence engine: a failure +// context (RLS) and returns its id, which the drives that deliver it carry +// (output_drives.alarm_event_id). It is called best-effort from the geofence engine: a failure // here is logged by the caller and never blocks ingestion or the asset_scans // write, which is the authoritative record. -func (s *Storage) InsertAlarmEvent(ctx context.Context, orgID int, ev AlarmEventRow) error { - return s.WithOrgTx(ctx, orgID, func(tx pgx.Tx) error { - _, err := tx.Exec(ctx, +func (s *Storage) InsertAlarmEvent(ctx context.Context, orgID int, ev AlarmEventRow) (int64, error) { + var id int64 + err := s.WithOrgTx(ctx, orgID, func(tx pgx.Tx) error { + err := tx.QueryRow(ctx, `INSERT INTO trakrf.alarm_events (org_id, asset_id, scan_point_id, location_id, epc, rssi, tag_scan_id, fired_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + RETURNING id`, orgID, ev.AssetID, ev.ScanPointID, ev.LocationID, ev.EPC, ev.RSSI, ev.TagScanID, ev.FiredAt, - ) + ).Scan(&id) if err != nil { return fmt.Errorf("insert alarm_event: %w", err) } return nil }) + return id, err } diff --git a/backend/internal/storage/alarm_events_integration_test.go b/backend/internal/storage/alarm_events_integration_test.go index 1e6d3d3ba..4c24eef2f 100644 --- a/backend/internal/storage/alarm_events_integration_test.go +++ b/backend/internal/storage/alarm_events_integration_test.go @@ -34,7 +34,7 @@ func TestInsertAlarmEvent_WritesUnderOrgContext(t *testing.T) { spID := scanPointID(t, db, orgID, dev.ID) firedAt := time.Now() - err := db.Store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ + _, err := db.Store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ AssetID: asset.ID, ScanPointID: spID, EPC: testEPC, @@ -64,9 +64,10 @@ func TestInsertAlarmEvent_RLSIsolatesByOrg(t *testing.T) { asset := testutil.CreateTestAsset(t, db.AdminPool, orgA, "alarm-asset") spID := scanPointID(t, db, orgA, dev.ID) - require.NoError(t, db.Store.InsertAlarmEvent(ctx, orgA, storage.AlarmEventRow{ + _, err := db.Store.InsertAlarmEvent(ctx, orgA, storage.AlarmEventRow{ AssetID: asset.ID, ScanPointID: spID, EPC: testEPC, RSSI: -50, FiredAt: time.Now(), - })) + }) + require.NoError(t, err) // Under org B's RLS context the store pool must not see org A's alarm row. var visibleToB, visibleToA int diff --git a/backend/internal/storage/output_drives.go b/backend/internal/storage/output_drives.go new file mode 100644 index 000000000..70b9481b9 --- /dev/null +++ b/backend/internal/storage/output_drives.go @@ -0,0 +1,135 @@ +package storage + +import ( + "context" + "fmt" + "time" + "unicode/utf8" + + "github.com/jackc/pgx/v5" +) + +// DriveCause says what drove an output device. The values match the +// output_drives.cause check. +type DriveCause string + +const ( + DriveCauseAlarm DriveCause = "alarm" // the ON a geofence fire sent + DriveCauseRefresh DriveCause = "refresh" // a held presence output's re-ON + DriveCauseAgeOut DriveCause = "age_out" // a presence output's OFF on age-out + DriveCauseTest DriveCause = "test" // the test action's ON and OFF + DriveCauseReset DriveCause = "reset" // the reset action's OFF + DriveCauseDelete DriveCause = "delete" // the OFF sent before a delete +) + +// maxDriveErrorBytes caps the stored error text, so a transport that returns +// a whole response body cannot bloat the table. +const maxDriveErrorBytes = 1000 + +// recordDriveTimeout bounds the write of one drive row. The write is detached +// from the caller's context, so this is the only bound it has. +const recordDriveTimeout = 5 * time.Second + +// OutputDriveRow is one drive to record: Err nil means it worked. +type OutputDriveRow struct { + OutputDeviceID int + AlarmEventID *int64 // the fire this ON delivered; nil for every other cause + On bool + Cause DriveCause + Err error + Latency time.Duration + DrivenAt time.Time +} + +// OutputDrive is one recorded drive, as read back. +type OutputDrive struct { + ID int64 + OutputDeviceID int + AlarmEventID *int64 + Edge string // on | off + Cause string + Result string // ok | error + Error *string + LatencyMs int + DrivenAt time.Time +} + +// RecordOutputDrive writes one drive row under org context (RLS). Callers +// treat it as best-effort: a failure here is logged and never changes what +// the drive did. The write is detached from ctx's cancellation, because the +// drive has already happened: an HTTP client hanging up after a test fire +// must not erase the record of it. +func (s *Storage) RecordOutputDrive(ctx context.Context, orgID int, row OutputDriveRow) error { + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recordDriveTimeout) + defer cancel() + + edge := "off" + if row.On { + edge = "on" + } + result := "ok" + var errText *string + if row.Err != nil { + result = "error" + t := truncateDriveError(row.Err.Error()) + errText = &t + } + latencyMs := max(0, row.Latency.Milliseconds()) + + return s.WithOrgTx(ctx, orgID, func(tx pgx.Tx) error { + _, err := tx.Exec(ctx, + `INSERT INTO trakrf.output_drives + (org_id, output_device_id, alarm_event_id, edge, cause, result, error, latency_ms, driven_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`, + orgID, row.OutputDeviceID, row.AlarmEventID, edge, string(row.Cause), result, errText, latencyMs, row.DrivenAt, + ) + if err != nil { + return fmt.Errorf("insert output_drive: %w", err) + } + return nil + }) +} + +// ListOutputDrives reads one output device's most recent drives, newest +// first: the per-output activity log. +func (s *Storage) ListOutputDrives(ctx context.Context, orgID, outputDeviceID, limit int) ([]OutputDrive, error) { + var out []OutputDrive + err := s.WithOrgTx(ctx, orgID, func(tx pgx.Tx) error { + rows, err := tx.Query(ctx, + `SELECT id, output_device_id, alarm_event_id, edge, cause, result, error, latency_ms, driven_at + FROM trakrf.output_drives + WHERE org_id = $1 AND output_device_id = $2 + ORDER BY driven_at DESC, id DESC + LIMIT $3`, + orgID, outputDeviceID, limit) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var d OutputDrive + if err := rows.Scan(&d.ID, &d.OutputDeviceID, &d.AlarmEventID, &d.Edge, &d.Cause, + &d.Result, &d.Error, &d.LatencyMs, &d.DrivenAt); err != nil { + return err + } + out = append(out, d) + } + return rows.Err() + }) + if err != nil { + return nil, fmt.Errorf("list output drives: %w", err) + } + return out, nil +} + +// truncateDriveError caps s at maxDriveErrorBytes without splitting a rune. +func truncateDriveError(s string) string { + if len(s) <= maxDriveErrorBytes { + return s + } + cut := maxDriveErrorBytes + for cut > 0 && !utf8.RuneStart(s[cut]) { + cut-- + } + return s[:cut] +} diff --git a/backend/internal/storage/output_drives_integration_test.go b/backend/internal/storage/output_drives_integration_test.go new file mode 100644 index 000000000..65299e109 --- /dev/null +++ b/backend/internal/storage/output_drives_integration_test.go @@ -0,0 +1,143 @@ +//go:build integration + +package storage_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/trakrf/platform/backend/internal/models/outputdevice" + "github.com/trakrf/platform/backend/internal/storage" + "github.com/trakrf/platform/backend/internal/testutil" +) + +func createOutput(t *testing.T, db *testutil.TestDB, orgID int, name string) int { + t.Helper() + dev, err := db.Store.CreateOutputDevice(context.Background(), orgID, outputdevice.CreateOutputDeviceRequest{ + Name: name, BaseURL: "http://192.0.2.10", + }) + require.NoError(t, err) + return dev.ID +} + +func TestInsertAlarmEvent_ReturnsTheRowID(t *testing.T) { + db := testutil.SetupTestDBFull(t) + ctx := context.Background() + orgID := testutil.CreateTestAccount(t, db.AdminPool) + dev := registerDevice(t, db, orgID, "cs463-214") + asset := testutil.CreateTestAsset(t, db.AdminPool, orgID, "alarm-asset") + + id, err := db.Store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ + AssetID: asset.ID, ScanPointID: scanPointID(t, db, orgID, dev.ID), EPC: testEPC, RSSI: -50, FiredAt: time.Now(), + }) + require.NoError(t, err) + + var epc string + require.NoError(t, db.AdminPool.QueryRow(ctx, `SELECT epc FROM trakrf.alarm_events WHERE id = $1`, id).Scan(&epc)) + assert.Equal(t, testEPC, epc) +} + +func TestRecordOutputDrive_ListsEachDeviceNewestFirst(t *testing.T) { + db := testutil.SetupTestDBFull(t) + ctx := context.Background() + orgID := testutil.CreateTestAccount(t, db.AdminPool) + horn := createOutput(t, db, orgID, "horn") + strobe := createOutput(t, db, orgID, "strobe") + dev := registerDevice(t, db, orgID, "cs463-214") + asset := testutil.CreateTestAsset(t, db.AdminPool, orgID, "alarm-asset") + alarmID, err := db.Store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ + AssetID: asset.ID, ScanPointID: scanPointID(t, db, orgID, dev.ID), EPC: testEPC, RSSI: -50, FiredAt: time.Now(), + }) + require.NoError(t, err) + + t0 := time.Now().Truncate(time.Millisecond) + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: horn, AlarmEventID: &alarmID, On: true, Cause: storage.DriveCauseAlarm, + Err: errors.New("reader rpc: timeout after 5s"), Latency: 5003 * time.Millisecond, DrivenAt: t0, + })) + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: horn, On: false, Cause: storage.DriveCauseAgeOut, Latency: 12 * time.Millisecond, DrivenAt: t0.Add(time.Minute), + })) + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: strobe, AlarmEventID: &alarmID, On: true, Cause: storage.DriveCauseAlarm, Latency: 40 * time.Millisecond, DrivenAt: t0, + })) + + got, err := db.Store.ListOutputDrives(ctx, orgID, horn, 10) + require.NoError(t, err) + require.Len(t, got, 2, "only the horn's drives") + + assert.Equal(t, "off", got[0].Edge) + assert.Equal(t, "age_out", got[0].Cause) + assert.Equal(t, "ok", got[0].Result) + assert.Nil(t, got[0].Error) + assert.Nil(t, got[0].AlarmEventID) + assert.Equal(t, 12, got[0].LatencyMs) + + assert.Equal(t, "on", got[1].Edge) + assert.Equal(t, "alarm", got[1].Cause) + assert.Equal(t, "error", got[1].Result) + require.NotNil(t, got[1].Error) + assert.Equal(t, "reader rpc: timeout after 5s", *got[1].Error) + require.NotNil(t, got[1].AlarmEventID) + assert.Equal(t, alarmID, *got[1].AlarmEventID) + assert.Equal(t, 5003, got[1].LatencyMs) + assert.True(t, got[1].DrivenAt.Equal(t0)) +} + +func TestRecordOutputDrive_SurvivesACancelledContext(t *testing.T) { + db := testutil.SetupTestDBFull(t) + orgID := testutil.CreateTestAccount(t, db.AdminPool) + horn := createOutput(t, db, orgID, "horn") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // the HTTP client hung up before the drive returned + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: horn, On: true, Cause: storage.DriveCauseTest, DrivenAt: time.Now(), + })) + + got, err := db.Store.ListOutputDrives(context.Background(), orgID, horn, 10) + require.NoError(t, err) + assert.Len(t, got, 1) +} + +func TestOutputDrives_RLSIsolatesByOrg(t *testing.T) { + db := testutil.SetupTestDBFull(t) + ctx := context.Background() + orgA := testutil.CreateTestAccount(t, db.AdminPool) + var orgB int + require.NoError(t, db.AdminPool.QueryRow(ctx, + `INSERT INTO trakrf.organizations (name, identifier, is_active) + VALUES ('Test Organization B', 'test-org-b', true) RETURNING id`).Scan(&orgB)) + horn := createOutput(t, db, orgA, "horn") + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgA, storage.OutputDriveRow{ + OutputDeviceID: horn, On: true, Cause: storage.DriveCauseTest, DrivenAt: time.Now(), + })) + + var visibleToB int + require.NoError(t, db.Store.WithOrgTx(ctx, orgB, func(tx pgx.Tx) error { + return tx.QueryRow(ctx, `SELECT count(*) FROM trakrf.output_drives`).Scan(&visibleToB) + })) + assert.Equal(t, 0, visibleToB) + got, err := db.Store.ListOutputDrives(ctx, orgA, horn, 10) + require.NoError(t, err) + assert.Len(t, got, 1) +} + +func TestOutputDrives_OKRowCannotCarryErrorText(t *testing.T) { + db := testutil.SetupTestDBFull(t) + ctx := context.Background() + orgID := testutil.CreateTestAccount(t, db.AdminPool) + horn := createOutput(t, db, orgID, "horn") + + _, err := db.AdminPool.Exec(ctx, ` + INSERT INTO trakrf.output_drives (org_id, output_device_id, edge, cause, result, error, latency_ms, driven_at) + VALUES ($1, $2, 'on', 'test', 'ok', 'boom', 1, now())`, orgID, horn) + require.Error(t, err) + assert.Contains(t, err.Error(), "output_drives_error_iff_failed") +} diff --git a/backend/internal/storage/output_drives_test.go b/backend/internal/storage/output_drives_test.go new file mode 100644 index 000000000..770cf0bcf --- /dev/null +++ b/backend/internal/storage/output_drives_test.go @@ -0,0 +1,21 @@ +package storage + +import ( + "strings" + "testing" + "unicode/utf8" + + "github.com/stretchr/testify/assert" +) + +func TestTruncateDriveError_KeepsShortText(t *testing.T) { + assert.Equal(t, "rpc timeout", truncateDriveError("rpc timeout")) +} + +func TestTruncateDriveError_CapsLongTextOnARuneBoundary(t *testing.T) { + long := strings.Repeat("é", maxDriveErrorBytes) // 2 bytes per rune + got := truncateDriveError(long) + assert.LessOrEqual(t, len(got), maxDriveErrorBytes) + assert.True(t, utf8.ValidString(got), "a cut must not split a rune") + assert.Greater(t, len(got), maxDriveErrorBytes-4) +} diff --git a/backend/migrations/000055_output_drives.down.sql b/backend/migrations/000055_output_drives.down.sql new file mode 100644 index 000000000..ddd4f7681 --- /dev/null +++ b/backend/migrations/000055_output_drives.down.sql @@ -0,0 +1,4 @@ +-- Drop the output drive log. Policy and indexes drop with the table. +SET search_path = trakrf, public; + +DROP TABLE IF EXISTS output_drives; diff --git a/backend/migrations/000055_output_drives.up.sql b/backend/migrations/000055_output_drives.up.sql new file mode 100644 index 000000000..4bac8f9fe --- /dev/null +++ b/backend/migrations/000055_output_drives.up.sql @@ -0,0 +1,58 @@ +-- One row per output-device drive: which device, which edge, why it was +-- driven, and whether the drive worked. +-- +-- alarm_events records that a rule fired for a read (detection). It does not +-- say which devices were driven or whether the drive reached them: one fire +-- drives every device at the location, each suppressed or driven on its own, +-- and a drive to an offline reader fails with only a log line to show for it. +-- These rows are the delivery record. An alarm whose drive rows are all +-- result = 'error' (or that has none) was detected but not delivered. +-- +-- cause says what drove it: +-- alarm the ON a fire sent; alarm_event_id names the fire, unless the +-- alarm row itself failed to write +-- refresh a presence output's periodic re-ON while a tag stays in view +-- age_out a presence output's OFF when no member has been read for its +-- age-out +-- test the output-device test action (ON, then OFF) +-- reset the output-device reset action (OFF) +-- delete the OFF sent before an output device is deleted +-- +-- output_device_id is the device as it was at drive time. Output devices are +-- soft-deleted, so the FK never blocks a delete. +-- +-- Regular table, not a hypertable, like alarm_events. Volume is bounded by +-- fires plus one refresh per held presence output every 10 s. +SET search_path = trakrf, public; + +CREATE TABLE output_drives ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + org_id BIGINT NOT NULL REFERENCES organizations(id), + output_device_id BIGINT NOT NULL REFERENCES output_devices(id), + alarm_event_id BIGINT REFERENCES alarm_events(id), + edge TEXT NOT NULL CHECK (edge IN ('on', 'off')), + cause TEXT NOT NULL + CHECK (cause IN ('alarm', 'refresh', 'age_out', 'test', 'reset', 'delete')), + result TEXT NOT NULL CHECK (result IN ('ok', 'error')), + error TEXT, + latency_ms INT NOT NULL CHECK (latency_ms >= 0), + driven_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT output_drives_error_iff_failed CHECK ((result = 'error') = (error IS NOT NULL)), + CONSTRAINT output_drives_alarm_only_for_alarm_cause CHECK (alarm_event_id IS NULL OR cause = 'alarm') +); + +-- The per-output activity log: one device's drives, newest first. +CREATE INDEX idx_output_drives_device_time ON output_drives(org_id, output_device_id, driven_at DESC); +-- An alarm's delivery: its drive rows. +CREATE INDEX idx_output_drives_alarm ON output_drives(alarm_event_id) WHERE alarm_event_id IS NOT NULL; + +COMMENT ON TABLE output_drives IS 'One row per output-device drive and its result. The delivery record for alarm_events.'; +COMMENT ON COLUMN output_drives.error IS 'Transport error text when result = error (RPC timeout, reader error code), capped at 1000 bytes.'; +COMMENT ON COLUMN output_drives.latency_ms IS 'Time the drive call took, success or failure.'; + +-- RLS: org isolation, USING-only, identical shape to alarm_events. +ALTER TABLE output_drives ENABLE ROW LEVEL SECURITY; + +CREATE POLICY org_isolation_output_drives ON output_drives + USING (org_id = current_setting('app.current_org_id')::BIGINT); diff --git a/backend/migrations/checksums.txt b/backend/migrations/checksums.txt index af7c0c1a2..c5122cdbc 100644 --- a/backend/migrations/checksums.txt +++ b/backend/migrations/checksums.txt @@ -113,3 +113,5 @@ f435d3a3de09b6e6ed032f8a1ef56d1c25218aa57c1c5e4f855f2a0803190ac5 000053_suppres e63f15be38e76b1189c299b952ddb19511b655838b8fe32c169f8e64cd838f16 000053_suppression_source_callback_fks.up.sql fb64d05971920ea00448d7f7804802382d133148a85ac0db33584129e8baa469 000054_notification_event_sources.down.sql ea8b874ac0efc8108144343d62a8a3b350d5a3d936263707589989319f605ee2 000054_notification_event_sources.up.sql +b2a68d3cec860c34ece04760fe71ca5fca77777fa4de8aeb2851ce8eef4b02ab 000055_output_drives.down.sql +eef009d88b90f10fc668b0b5f7b373da17f4e4cce63c8a6de2930ce8174201da 000055_output_drives.up.sql From d5337f2d9ec4968defa57e58673108f7294f8db6 Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:08:26 -0600 Subject: [PATCH 2/4] feat: record every geofence drive, alarm row first Evaluate now writes the alarm_events row before driving, so each ON names the fire it delivered, then records one output_drives row per drive: the fire's ONs (cause alarm), a held presence output's refreshes (refresh) and the age-out OFF (age_out), which had no record at all. A fire whose every drive failed is logged as a warning and counted in geofence_alarms_undelivered_total, so a drive to an offline reader no longer reads as a working alarm. A failed alarm or drive-row write never stops or delays a drive. Refs: TRA-1314 Co-Authored-By: Claude Opus 5.5 --- backend/internal/geofence/engine.go | 134 +++++++++++---- .../internal/geofence/engine_drives_test.go | 162 ++++++++++++++++++ backend/internal/geofence/engine_test.go | 31 +++- backend/internal/geofence/metrics.go | 15 ++ backend/internal/geofence/presence.go | 26 +-- backend/internal/geofence/presence_test.go | 12 +- 6 files changed, 333 insertions(+), 47 deletions(-) create mode 100644 backend/internal/geofence/engine_drives_test.go diff --git a/backend/internal/geofence/engine.go b/backend/internal/geofence/engine.go index faf718a9d..3dfcca6c6 100644 --- a/backend/internal/geofence/engine.go +++ b/backend/internal/geofence/engine.go @@ -8,8 +8,10 @@ // presence), age_out_seconds, rssi_threshold, auto_off_seconds. The engine // resolves location -> outputs and keys its dedup/presence state per (output, // epc). Membership is implicit — only membership-passing reads ever reach -// Evaluate. Firing writes an alarm_events row and drives the device; both are -// best-effort and never block ingestion or the authoritative asset_scans write. +// Evaluate. Firing writes an alarm_events row (the detection), then drives each +// device and writes one output_drives row per drive (the delivery, TRA-1314). +// All of it is best-effort and never blocks ingestion or the authoritative +// asset_scans write. package geofence import ( @@ -29,6 +31,7 @@ import ( // satisfies it. Narrowed so unit tests can inject a fake. type engineStore interface { InsertAlarmEvent(ctx context.Context, orgID int, ev storage.AlarmEventRow) (int64, error) + RecordOutputDrive(ctx context.Context, orgID int, row storage.OutputDriveRow) error ListOutputDevicesForLocation(ctx context.Context, orgID, locationID int) ([]outputdevice.OutputDevice, error) GetOrgGeofenceDefaults(ctx context.Context, orgID int) (organization.GeofenceDefaults, error) } @@ -62,16 +65,19 @@ type Engine struct { func NewEngine(cfg Config, store *storage.Storage, driver outputDriver, log *zerolog.Logger) *Engine { l := log.With().Str("component", "geofence").Logger() clk := RealClock{} - return &Engine{ + e := &Engine{ cfg: cfg, store: store, driver: driver, latch: newLatch(cfg.SweepInterval, clk), pulses: newPulseGuard(), - presence: newPresence(driver, l), startupGrace: cfg.StartupGrace, log: l, } + // The presence tracker drives its age-out OFF through the engine, so that + // OFF is recorded like every other drive. + e.presence = newPresence(e.drive, l) + return e } // Start is a no-op today (the sweepers start in their constructors) but keeps the @@ -153,7 +159,9 @@ func (e *Engine) Evaluate(ctx context.Context, orgID int, tagScanID int64, recei continue } - firedAny := false + // The devices this read fires, collected first: the alarm row is written + // before any drive, so each ON can name the fire it delivers. + var fires []pendingFire for _, dev := range devices { // Collapse the three tuning tiers (system/code -> org default -> // per-output override) for this device (TRA-955). @@ -179,15 +187,13 @@ func (e *Engine) Evaluate(ctx context.Context, orgID int, tagScanID int64, recei metricSuppressed.WithLabelValues("startup_grace").Inc() continue } - firedAny = true - ok := e.drive(ctx, orgID, dev, true, presenceOnSeconds) - e.presence.driven(orgID, dev.ID, receivedAt, ok) + fires = append(fires, pendingFire{dev: dev, offAfter: presenceOnSeconds, presence: true}) continue } if e.presence.refreshDue(orgID, dev.ID, receivedAt) { // A refresh, or the retry of a failed ON. Not a fire: no // alarm_events row. - ok := e.drive(ctx, orgID, dev, true, presenceOnSeconds) + ok := e.drive(ctx, orgID, dev, true, presenceOnSeconds, receivedAt, storage.DriveCauseRefresh, nil) e.presence.driven(orgID, dev.ID, receivedAt, ok) } continue @@ -213,40 +219,88 @@ func (e *Engine) Evaluate(ctx context.Context, orgID int, tagScanID int64, recei metricSuppressed.WithLabelValues("pulse_active").Inc() continue } - firedAny = true - if !e.drive(ctx, orgID, dev, true, tuning.AutoOffSeconds) { - // A failed ON must not hold the output busy for the whole window. - e.pulses.release(orgID, dev.ID, receivedAt, pulse) - } + fires = append(fires, pendingFire{dev: dev, offAfter: tuning.AutoOffSeconds, pulse: pulse}) } - if firedAny { - e.recordFire(ctx, orgID, tagScanID, receivedAt, rd) + if len(fires) == 0 { + continue + } + alarmID := e.recordFire(ctx, orgID, tagScanID, receivedAt, rd) + delivered := 0 + for _, f := range fires { + ok := e.drive(ctx, orgID, f.dev, true, f.offAfter, receivedAt, storage.DriveCauseAlarm, alarmID) + if ok { + delivered++ + } + if f.presence { + e.presence.driven(orgID, f.dev.ID, receivedAt, ok) + } else if !ok { + // A failed ON must not hold the output busy for the whole window. + e.pulses.release(orgID, f.dev.ID, receivedAt, f.pulse) + } } + e.logFire(orgID, rd, delivered, len(fires)) } } +// pendingFire is one device a read fires: the ON to send once the alarm row +// exists. pulse is the egress re-entrancy window to release if the ON fails. +type pendingFire struct { + dev outputdevice.OutputDevice + offAfter int + presence bool + pulse time.Duration +} + // presenceOnSeconds is the on-time every presence ON carries: the reader's // default on-time, sent explicitly so the backend and the daemon cannot drift // apart. A Shelly gets it as toggle_after, a CS463 GPO as pulse_ms. var presenceOnSeconds = int(readerrpc.GpoDefaultOn / time.Second) -// drive turns a device on/off, folding any error into the best-effort metric. -// It reports whether the drive succeeded. -func (e *Engine) drive(ctx context.Context, orgID int, dev outputdevice.OutputDevice, on bool, offAfter int) bool { - if err := e.driver.Set(ctx, dev, on, offAfter); err != nil { - e.log.Error().Err(err).Int("org_id", orgID).Int("output_device_id", dev.ID).Bool("on", on).Msg("output device drive failed (best-effort)") +// drive turns a device on/off and records the drive as one output_drives row +// (TRA-1314). It reports whether the drive succeeded. A drive error is folded +// into the best-effort metric; a record error is logged and never changes the +// result. at is the drive's driven_at: the read's receive time for a read-driven +// drive, the timer's fire time for an age-out OFF. alarmID names the fire an +// ON delivers, nil for every other cause. +func (e *Engine) drive(ctx context.Context, orgID int, dev outputdevice.OutputDevice, on bool, offAfter int, at time.Time, cause storage.DriveCause, alarmID *int64) bool { + start := time.Now() + err := e.driver.Set(ctx, dev, on, offAfter) + latency := time.Since(start) + + edge, result := "off", "ok" + if on { + edge = "on" + } + if err != nil { + result = "error" + e.log.Error().Err(err).Int("org_id", orgID).Int("output_device_id", dev.ID).Bool("on", on). + Str("cause", string(cause)).Msg("output device drive failed (best-effort)") metricFireErrors.Inc() - return false } - return true + metricDrives.WithLabelValues(edge, result).Inc() + + if rerr := e.store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: dev.ID, + AlarmEventID: alarmID, + On: on, + Cause: cause, + Err: err, + Latency: latency, + DrivenAt: at, + }); rerr != nil { + e.log.Error().Err(rerr).Int("org_id", orgID).Int("output_device_id", dev.ID).Msg("output_drives write failed") + metricDriveWriteErrors.Inc() + } + return err == nil } -// recordFire writes the durable alarm_events row for a read that drove at least -// one device on, and logs/metrics the fire. Best-effort: a write error never -// blocks ingestion. -func (e *Engine) recordFire(ctx context.Context, orgID int, tagScanID int64, receivedAt time.Time, rd storage.ResolvedRead) { - if _, err := e.store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ +// recordFire writes the durable alarm_events row for a read that fires at +// least one device, before any of them is driven. It returns the row's id, or +// nil when the write failed: the drives still go, and their rows carry no +// alarm id. Best-effort: a write error never blocks ingestion or a drive. +func (e *Engine) recordFire(ctx context.Context, orgID int, tagScanID int64, receivedAt time.Time, rd storage.ResolvedRead) *int64 { + id, err := e.store.InsertAlarmEvent(ctx, orgID, storage.AlarmEventRow{ AssetID: rd.AssetID, ScanPointID: rd.ScanPointID, LocationID: rd.LocationID, @@ -254,17 +308,33 @@ func (e *Engine) recordFire(ctx context.Context, orgID int, tagScanID int64, rec RSSI: rd.RSSI, TagScanID: tagScanID, FiredAt: receivedAt, - }); err != nil { + }) + if err != nil { e.log.Error().Err(err).Int("org_id", orgID).Str("epc", rd.EPC).Msg("alarm_events write failed") metricEventWriteErrors.Inc() + return nil } + return &id +} +// logFire logs and metrics a fire once its drives are done. A fire none of +// whose devices was reached is undelivered: it is logged as a warning and +// counted, so it never reads as a working alarm. +func (e *Engine) logFire(orgID int, rd storage.ResolvedRead, delivered, devices int) { metricFired.Inc() - e.log.Info(). - Int("org_id", orgID). + ev := e.log.Info() + msg := "geofence rule fired" + if delivered == 0 { + metricUndelivered.Inc() + ev = e.log.Warn() + msg = "geofence rule fired but no output device was reached" + } + ev.Int("org_id", orgID). Int("asset_id", rd.AssetID). Int("scan_point_id", rd.ScanPointID). Str("epc", rd.EPC). Int("rssi", rd.RSSI). - Msg("geofence rule fired") + Int("devices", devices). + Int("delivered", delivered). + Msg(msg) } diff --git a/backend/internal/geofence/engine_drives_test.go b/backend/internal/geofence/engine_drives_test.go new file mode 100644 index 000000000..0e4d2dadb --- /dev/null +++ b/backend/internal/geofence/engine_drives_test.go @@ -0,0 +1,162 @@ +package geofence + +import ( + "context" + "errors" + "testing" + "time" + + dto "github.com/prometheus/client_model/go" + + "github.com/trakrf/platform/backend/internal/models/outputdevice" + "github.com/trakrf/platform/backend/internal/storage" +) + +// TRA-1314: every drive leaves an output_drives row naming the device and the +// result, and the alarm row is written before the drives so each ON can name +// the fire it delivered. + +// undelivered reads the global undelivered-alarm counter; tests assert deltas. +func undelivered(t *testing.T) float64 { + t.Helper() + m := &dto.Metric{} + if err := metricUndelivered.Write(m); err != nil { + t.Fatalf("read undelivered counter: %v", err) + } + return m.GetCounter().GetValue() +} + +func TestEvaluate_RecordsEachDeviceDriveWithAlarmID(t *testing.T) { + s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil), dev(2, nil)}} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, &fakeDriver{}) + defer e.Stop() + at := time.Unix(1000, 0) + + e.Evaluate(context.Background(), 42, 9, at, []storage.ResolvedRead{read("EPC1", -50)}) + + drives := s.recordedDrives() + if len(drives) != 2 { + t.Fatalf("drive rows = %d, want one per device (2)", len(drives)) + } + for i, d := range drives { + if d.OutputDeviceID != i+1 || !d.On || d.Cause != storage.DriveCauseAlarm || d.Err != nil { + t.Errorf("drive %d = %+v, want device %d, ON, cause alarm, ok", i, d, i+1) + } + if d.AlarmEventID == nil || *d.AlarmEventID != 1 { + t.Errorf("drive %d alarm id = %v, want the fire's id 1", i, d.AlarmEventID) + } + if !d.DrivenAt.Equal(at) { + t.Errorf("drive %d driven_at = %v, want the read time %v", i, d.DrivenAt, at) + } + } +} + +func TestEvaluate_FailedDriveRecordsErrorAndCountsUndelivered(t *testing.T) { + s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil)}} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, &fakeDriver{err: errors.New("reader rpc: timeout after 5s")}) + defer e.Stop() + before := undelivered(t) + + e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) + + drives := s.recordedDrives() + if len(drives) != 1 || drives[0].Err == nil || drives[0].Err.Error() != "reader rpc: timeout after 5s" { + t.Fatalf("drives = %+v, want one row carrying the timeout", drives) + } + if got := undelivered(t) - before; got != 1 { + t.Errorf("undelivered alarms += %v, want 1", got) + } + if len(s.rows) != 1 { + t.Errorf("alarm rows = %d, want 1: the detection is still recorded", len(s.rows)) + } +} + +func TestEvaluate_OneDeliveredDeviceIsNotUndelivered(t *testing.T) { + s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil), dev(2, nil)}} + d := &flakyDriver{fail: map[int]bool{0: true}} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, d) + defer e.Stop() + before := undelivered(t) + + e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) + + if got := undelivered(t) - before; got != 0 { + t.Errorf("undelivered alarms += %v, want 0 when one of two devices worked", got) + } +} + +func TestEvaluate_AlarmInsertFailureStillDrivesAndRecords(t *testing.T) { + s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil)}, insertErr: errors.New("db down")} + d := &fakeDriver{} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, d) + defer e.Stop() + + e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) + + if d.onCount() != 1 { + t.Fatalf("ON drives = %d, want 1: a failed alarm write must not stop the horn", d.onCount()) + } + drives := s.recordedDrives() + if len(drives) != 1 || drives[0].AlarmEventID != nil || drives[0].Cause != storage.DriveCauseAlarm { + t.Fatalf("drives = %+v, want one alarm-cause row with no alarm id", drives) + } +} + +func TestEvaluate_DriveRecordFailureDoesNotStopTheNextDevice(t *testing.T) { + s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil), dev(2, nil)}, driveErr: errors.New("db down")} + d := &fakeDriver{} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, d) + defer e.Stop() + + e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) + + if d.onCount() != 2 { + t.Fatalf("ON drives = %d, want 2", d.onCount()) + } +} + +func TestEvaluate_PresenceRefreshRecordsCauseRefresh(t *testing.T) { + s := presenceStore() + e := presenceEngine(t, &fakeDriver{}, s) + t0 := time.Unix(1000, 0) + + evalAt(e, 1, t0) + evalAt(e, 2, t0.Add(presenceRefreshEvery)) + + drives := s.recordedDrives() + if len(drives) != 2 { + t.Fatalf("drive rows = %d, want the edge ON and one refresh", len(drives)) + } + if drives[0].Cause != storage.DriveCauseAlarm || drives[0].AlarmEventID == nil { + t.Errorf("first drive = %+v, want the alarm ON naming its fire", drives[0]) + } + if drives[1].Cause != storage.DriveCauseRefresh || drives[1].AlarmEventID != nil || !drives[1].On { + t.Errorf("second drive = %+v, want a refresh ON with no alarm id", drives[1]) + } +} + +func TestEvaluate_PresenceAgeOutRecordsOffRow(t *testing.T) { + s := presenceStore() + e := presenceEngine(t, &fakeDriver{}, s) + var timers []*recordedTimer + e.presence.afterFunc = func(d time.Duration, f func()) timerHandle { + rt := &recordedTimer{d: d, f: f} + timers = append(timers, rt) + return rt + } + + evalAt(e, 1, time.Unix(1000, 0)) + timers[len(timers)-1].fire() + + drives := s.recordedDrives() + if len(drives) != 2 { + t.Fatalf("drive rows = %d, want ON then the age-out OFF", len(drives)) + } + off := drives[1] + if off.On || off.Cause != storage.DriveCauseAgeOut || off.AlarmEventID != nil || off.OutputDeviceID != 11 || off.Err != nil { + t.Errorf("age-out drive = %+v, want device 11 OFF, cause age_out, no alarm id, ok", off) + } + if s.driveOrgs[1] != 42 { + t.Errorf("age-out drive org = %d, want 42", s.driveOrgs[1]) + } +} diff --git a/backend/internal/geofence/engine_test.go b/backend/internal/geofence/engine_test.go index 484d91919..8f236e717 100644 --- a/backend/internal/geofence/engine_test.go +++ b/backend/internal/geofence/engine_test.go @@ -15,8 +15,8 @@ import ( "github.com/trakrf/platform/backend/internal/storage" ) -// fakeStore records alarm_events writes and serves a fixed device list; either -// call can be made to fail. +// fakeStore records alarm_events and output_drives writes and serves a fixed +// device list; any call can be made to fail. type fakeStore struct { rows []storage.AlarmEventRow devices []outputdevice.OutputDevice @@ -24,6 +24,28 @@ type fakeStore struct { insertErr error listErr error defaultsErr error + driveErr error + + mu sync.Mutex // drives are recorded from the presence timer goroutine too + drives []storage.OutputDriveRow + driveOrgs []int +} + +func (s *fakeStore) RecordOutputDrive(_ context.Context, orgID int, row storage.OutputDriveRow) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.driveErr != nil { + return s.driveErr + } + s.drives = append(s.drives, row) + s.driveOrgs = append(s.driveOrgs, orgID) + return nil +} + +func (s *fakeStore) recordedDrives() []storage.OutputDriveRow { + s.mu.Lock() + defer s.mu.Unlock() + return append([]storage.OutputDriveRow(nil), s.drives...) } func (s *fakeStore) GetOrgGeofenceDefaults(_ context.Context, _ int) (organization.GeofenceDefaults, error) { @@ -96,16 +118,17 @@ func (d *fakeDriver) first() setCall { func newTestEngine(cfg Config, s engineStore, d outputDriver) *Engine { log := zerolog.New(io.Discard) clk := NewFakeClock(time.Unix(0, 0)) - return &Engine{ + e := &Engine{ cfg: cfg, store: s, driver: d, latch: newLatch(0, clk), pulses: newPulseGuard(), - presence: newPresence(d, log), startupGrace: cfg.StartupGrace, log: log, } + e.presence = newPresence(e.drive, log) + return e } func ptr(i int) *int { return &i } diff --git a/backend/internal/geofence/metrics.go b/backend/internal/geofence/metrics.go index c4666caf7..a5a603382 100644 --- a/backend/internal/geofence/metrics.go +++ b/backend/internal/geofence/metrics.go @@ -28,6 +28,21 @@ var ( Help: "Errors driving output devices or looking them up (best-effort; do not block ingestion).", }) + metricDrives = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "geofence_output_drives_total", + Help: "Output-device drives sent by the geofence engine, by edge and result.", + }, []string{"edge", "result"}) // edge: on|off; result: ok|error + + metricUndelivered = promauto.NewCounter(prometheus.CounterOpts{ + Name: "geofence_alarms_undelivered_total", + Help: "Alarms whose every output-device drive failed: detected, but no device was reached.", + }) + + metricDriveWriteErrors = promauto.NewCounter(prometheus.CounterOpts{ + Name: "geofence_drive_write_errors_total", + Help: "Errors writing output_drives rows (best-effort; never delays a drive).", + }) + metricEventWriteErrors = promauto.NewCounter(prometheus.CounterOpts{ Name: "geofence_event_write_errors_total", Help: "Errors writing alarm_events rows (best-effort; do not block ingestion).", diff --git a/backend/internal/geofence/presence.go b/backend/internal/geofence/presence.go index 32f929a4f..349db314c 100644 --- a/backend/internal/geofence/presence.go +++ b/backend/internal/geofence/presence.go @@ -9,8 +9,13 @@ import ( "github.com/rs/zerolog" "github.com/trakrf/platform/backend/internal/models/outputdevice" + "github.com/trakrf/platform/backend/internal/storage" ) +// driveFunc drives one output and records the drive; Engine.drive satisfies +// it. It reports whether the drive succeeded. +type driveFunc func(ctx context.Context, orgID int, dev outputdevice.OutputDevice, on bool, offAfter int, at time.Time, cause storage.DriveCause, alarmID *int64) bool + // presence drives presence-mode outputs (TRA-943/TRA-950). Each output, while a // member tag is present, holds a single per-output timer set to the output's // age-out and reset on every qualifying read. The ON edge fires on the 0->1 @@ -27,8 +32,8 @@ type presence struct { outputs map[string]*presenceOutput // "org:outputID" -> state closed bool - driver outputDriver - log zerolog.Logger + drive driveFunc + log zerolog.Logger // afterFunc schedules f to run after d; *time.Timer satisfies the return. // Injectable so tests fire timers deterministically instead of sleeping. @@ -41,6 +46,7 @@ type timerHandle interface { } type presenceOutput struct { + orgID int // so the age-out OFF is recorded under its org dev outputdevice.OutputDevice // snapshot, so the timer can drive OFF later timer timerHandle // gen increments on every observe; the timer callback captures the gen it was @@ -88,10 +94,10 @@ func (p *presence) refreshDue(orgID, outputID int, at time.Time) bool { return po != nil && po.wanted && at.Sub(po.onAt) >= presenceRefreshEvery } -func newPresence(driver outputDriver, log zerolog.Logger) *presence { +func newPresence(drive driveFunc, log zerolog.Logger) *presence { return &presence{ outputs: map[string]*presenceOutput{}, - driver: driver, + drive: drive, log: log, afterFunc: func(d time.Duration, f func()) timerHandle { return time.AfterFunc(d, f) @@ -119,6 +125,7 @@ func (p *presence) observe(orgID int, dev outputdevice.OutputDevice, ttl time.Du po = &presenceOutput{} p.outputs[k] = po } + po.orgID = orgID po.dev = dev po.gen++ gen := po.gen @@ -130,8 +137,8 @@ func (p *presence) observe(orgID int, dev outputdevice.OutputDevice, ttl time.Du } // expire drives an output off when its age-out timer fires, unless a newer read -// has rescheduled it (stale gen) or it has already been cleared. The Set happens -// outside the lock (best-effort). +// has rescheduled it (stale gen) or it has already been cleared. The drive +// happens outside the lock (best-effort) and is recorded with cause age_out. func (p *presence) expire(key string, gen uint64) { p.mu.Lock() po := p.outputs[key] @@ -139,13 +146,12 @@ func (p *presence) expire(key string, gen uint64) { p.mu.Unlock() return } - dev := po.dev + orgID, dev := po.orgID, po.dev delete(p.outputs, key) p.mu.Unlock() - if err := p.driver.Set(context.Background(), dev, false, 0); err != nil { - p.log.Error().Err(err).Int("output_device_id", dev.ID).Msg("presence off failed (best-effort)") - } + // drive logs and metrics a failure itself. + p.drive(context.Background(), orgID, dev, false, 0, time.Now(), storage.DriveCauseAgeOut, nil) } // Close stops all outstanding timers and blocks further observes. Idempotent. diff --git a/backend/internal/geofence/presence_test.go b/backend/internal/geofence/presence_test.go index 0b421f322..833009707 100644 --- a/backend/internal/geofence/presence_test.go +++ b/backend/internal/geofence/presence_test.go @@ -1,6 +1,7 @@ package geofence import ( + "context" "io" "testing" "time" @@ -8,6 +9,7 @@ import ( "github.com/rs/zerolog" "github.com/trakrf/platform/backend/internal/models/outputdevice" + "github.com/trakrf/platform/backend/internal/storage" ) // recordedTimer is a fake timer the test fires manually. Stop() reports whether @@ -34,10 +36,18 @@ func (t *recordedTimer) fire() { } } +// plainDrive adapts a bare driver to the presence tracker's driveFunc, with no +// recording, for tests of the tracker alone. +func plainDrive(d outputDriver) driveFunc { + return func(ctx context.Context, _ int, dev outputdevice.OutputDevice, on bool, offAfter int, _ time.Time, _ storage.DriveCause, _ *int64) bool { + return d.Set(ctx, dev, on, offAfter) == nil + } +} + // newTestPresence builds a presence tracker whose timers are captured for manual // firing instead of running on the wall clock. func newTestPresence(driver outputDriver) (*presence, *[]*recordedTimer) { - p := newPresence(driver, zerolog.New(io.Discard)) + p := newPresence(plainDrive(driver), zerolog.New(io.Discard)) timers := &[]*recordedTimer{} p.afterFunc = func(d time.Duration, f func()) timerHandle { rt := &recordedTimer{d: d, f: f} From cce3f45aa46cb6aa0325b25c8c5b3a7f210b7960 Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:10:25 -0600 Subject: [PATCH 3/4] feat: record the test, reset and delete drives of an output device The manual actions drive the same devices a geofence fire does, and commissioning proves a horn works with the test action. Each now leaves an output_drives row (cause test, reset or delete), failures included, so that proof no longer needs SSH to the reader's own log. Refs: TRA-1314 Co-Authored-By: Claude Opus 5.5 --- .../internal/handlers/outputdevices/drives.go | 31 +++++++ .../handlers/outputdevices/outputdevices.go | 8 +- .../outputdevices_drives_integration_test.go | 85 +++++++++++++++++++ 3 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 backend/internal/handlers/outputdevices/drives.go create mode 100644 backend/internal/handlers/outputdevices/outputdevices_drives_integration_test.go diff --git a/backend/internal/handlers/outputdevices/drives.go b/backend/internal/handlers/outputdevices/drives.go new file mode 100644 index 000000000..57050761b --- /dev/null +++ b/backend/internal/handlers/outputdevices/drives.go @@ -0,0 +1,31 @@ +package outputdevices + +import ( + "context" + "time" + + "github.com/trakrf/platform/backend/internal/logger" + "github.com/trakrf/platform/backend/internal/models/outputdevice" + "github.com/trakrf/platform/backend/internal/storage" +) + +// set drives dev and records the drive as one output_drives row (TRA-1314), +// so a test, reset or delete leaves the same record a geofence fire does. It +// returns the drive's own error. The record is best-effort: a failed write is +// logged and never changes the response. +func (h *Handler) set(ctx context.Context, orgID int, dev outputdevice.OutputDevice, on bool, offAfterSec int, cause storage.DriveCause) error { + drivenAt := time.Now() + err := h.actuator.Set(ctx, dev, on, offAfterSec) + if rerr := h.storage.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: dev.ID, + On: on, + Cause: cause, + Err: err, + Latency: time.Since(drivenAt), + DrivenAt: drivenAt, + }); rerr != nil { + logger.Get().Error().Err(rerr).Int("org_id", orgID).Int("output_device_id", dev.ID). + Str("cause", string(cause)).Msg("output_drives write failed") + } + return err +} diff --git a/backend/internal/handlers/outputdevices/outputdevices.go b/backend/internal/handlers/outputdevices/outputdevices.go index b7c3fe1d3..f2e00b9f0 100644 --- a/backend/internal/handlers/outputdevices/outputdevices.go +++ b/backend/internal/handlers/outputdevices/outputdevices.go @@ -341,7 +341,7 @@ func (h *Handler) Delete(w http.ResponseWriter, r *http.Request) { if !found { return } - if err := h.actuator.Set(r.Context(), *device, false, 0); err != nil && r.URL.Query().Get("force") != "true" { + if err := h.set(r.Context(), orgID, *device, false, 0, storage.DriveCauseDelete); err != nil && r.URL.Query().Get("force") != "true" { httputil.WriteJSONError(w, r, http.StatusBadGateway, modelerrors.ErrInternal, "could not turn the output off before deleting it, so it was kept: "+err.Error(), reqID) return @@ -382,7 +382,7 @@ func (h *Handler) Test(w http.ResponseWriter, r *http.Request) { // cannot leave the output latched on (TRA-1273): the OFF is a second // message, and alarm outputs fail toward off. offAfter := max(1, int(math.Ceil(h.testPulse.Seconds()))) - if err := h.actuator.Set(ctx, *device, true, offAfter); err != nil { + if err := h.set(ctx, orgID, *device, true, offAfter, storage.DriveCauseTest); err != nil { httputil.WriteJSONError(w, r, http.StatusBadGateway, modelerrors.ErrInternal, "output device unreachable: "+err.Error(), reqID) return } @@ -390,7 +390,7 @@ func (h *Handler) Test(w http.ResponseWriter, r *http.Request) { time.Sleep(h.testPulse) } // Best-effort off; the operator can still use reset if this fails. - _ = h.actuator.Set(ctx, *device, false, 0) + _ = h.set(ctx, orgID, *device, false, 0, storage.DriveCauseTest) httputil.WriteJSON(w, http.StatusOK, map[string]any{"status": "ok"}) } @@ -412,7 +412,7 @@ func (h *Handler) Reset(w http.ResponseWriter, r *http.Request) { if !ok { return } - if err := h.actuator.Set(r.Context(), *device, false, 0); err != nil { + if err := h.set(r.Context(), orgID, *device, false, 0, storage.DriveCauseReset); err != nil { httputil.WriteJSONError(w, r, http.StatusBadGateway, modelerrors.ErrInternal, "output device unreachable: "+err.Error(), reqID) return } diff --git a/backend/internal/handlers/outputdevices/outputdevices_drives_integration_test.go b/backend/internal/handlers/outputdevices/outputdevices_drives_integration_test.go new file mode 100644 index 000000000..359a45f2b --- /dev/null +++ b/backend/internal/handlers/outputdevices/outputdevices_drives_integration_test.go @@ -0,0 +1,85 @@ +//go:build integration + +package outputdevices_test + +import ( + "context" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/trakrf/platform/backend/internal/testutil" +) + +// TRA-1314: the manual actions are drives too, and leave the same record as a +// geofence fire. Commissioning proves a horn works with the test action, and +// that proof must not need SSH to the reader. + +func TestDrives_TestActionRecordsOnThenOff(t *testing.T) { + db := testutil.SetupTestDBFull(t) + r, orgID := newTestServerOn(t, db, &fakeDriver{}) + id := createDevice(t, r, orgID, "http://192.168.50.66") + + rec := doReq(t, r, orgID, http.MethodPost, "/api/v1/output-devices/"+itoa(id)+"/test", nil) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + + drives, err := db.Store.ListOutputDrives(context.Background(), orgID, id, 10) + require.NoError(t, err) + require.Len(t, drives, 2) + edges := map[string]string{} + for _, d := range drives { + edges[d.Edge] = d.Cause + "/" + d.Result + } + assert.Equal(t, map[string]string{"on": "test/ok", "off": "test/ok"}, edges) +} + +func TestDrives_FailedTestActionRecordsTheError(t *testing.T) { + db := testutil.SetupTestDBFull(t) + r, orgID := newTestServerOn(t, db, &fakeDriver{failURL: "http://192.168.50.66"}) + id := createDevice(t, r, orgID, "http://192.168.50.66") + + rec := doReq(t, r, orgID, http.MethodPost, "/api/v1/output-devices/"+itoa(id)+"/test", nil) + require.Equal(t, http.StatusBadGateway, rec.Code, rec.Body.String()) + + drives, err := db.Store.ListOutputDrives(context.Background(), orgID, id, 10) + require.NoError(t, err) + require.Len(t, drives, 1, "the ON failed, so no OFF was sent") + assert.Equal(t, "on", drives[0].Edge) + assert.Equal(t, "error", drives[0].Result) + require.NotNil(t, drives[0].Error) + assert.Equal(t, "device unreachable", *drives[0].Error) +} + +func TestDrives_ResetRecordsOff(t *testing.T) { + db := testutil.SetupTestDBFull(t) + r, orgID := newTestServerOn(t, db, &fakeDriver{}) + id := createDevice(t, r, orgID, "http://192.168.50.66") + + rec := doReq(t, r, orgID, http.MethodPost, "/api/v1/output-devices/"+itoa(id)+"/reset", nil) + require.Equal(t, http.StatusOK, rec.Code, rec.Body.String()) + + drives, err := db.Store.ListOutputDrives(context.Background(), orgID, id, 10) + require.NoError(t, err) + require.Len(t, drives, 1) + assert.Equal(t, "off", drives[0].Edge) + assert.Equal(t, "reset", drives[0].Cause) + assert.Equal(t, "ok", drives[0].Result) +} + +func TestDrives_DeleteRecordsItsOffEvenWhenItFails(t *testing.T) { + db := testutil.SetupTestDBFull(t) + r, orgID := newTestServerOn(t, db, &fakeDriver{failURL: "http://192.168.50.66"}) + id := createDevice(t, r, orgID, "http://192.168.50.66") + + rec := doReq(t, r, orgID, http.MethodDelete, "/api/v1/output-devices/"+itoa(id), nil) + require.Equal(t, http.StatusBadGateway, rec.Code, rec.Body.String()) + + drives, err := db.Store.ListOutputDrives(context.Background(), orgID, id, 10) + require.NoError(t, err) + require.Len(t, drives, 1) + assert.Equal(t, "off", drives[0].Edge) + assert.Equal(t, "delete", drives[0].Cause) + assert.Equal(t, "error", drives[0].Result) +} From 7feb36836c99b502ac7d675763b2768551d81d6a Mon Sep 17 00:00:00 2001 From: Mike Stankavich Date: Sun, 8 Nov 2026 22:27:51 -0600 Subject: [PATCH 4/4] fix: record a broker-only drive as sent, not ok A CS463 GPO drive goes through readercontrol's fire-and-forget Notify, and a Shelly over MQTT is a bare publish: success means the broker took the message, not that the device acted. A powered-off reader therefore recorded result=ok. An mqtt-transport drive that publishes cleanly is now result=sent; only an http device's drive can be ok. geofence_alarms_fired_total's help now says detected, not sounded. Refs: TRA-1314 Co-Authored-By: Claude Opus 5.5 --- backend/internal/geofence/engine.go | 4 +++ .../internal/geofence/engine_drives_test.go | 32 +++++++++++++++++-- backend/internal/geofence/metrics.go | 6 ++-- .../internal/handlers/outputdevices/drives.go | 1 + .../models/outputdevice/outputdevice.go | 7 ++++ .../models/outputdevice/outputdevice_test.go | 15 +++++++++ backend/internal/storage/output_drives.go | 13 ++++++-- .../storage/output_drives_integration_test.go | 28 ++++++++++++++-- .../migrations/000055_output_drives.up.sql | 17 +++++++--- backend/migrations/checksums.txt | 2 +- 10 files changed, 110 insertions(+), 15 deletions(-) diff --git a/backend/internal/geofence/engine.go b/backend/internal/geofence/engine.go index 3dfcca6c6..20a3f962a 100644 --- a/backend/internal/geofence/engine.go +++ b/backend/internal/geofence/engine.go @@ -272,6 +272,9 @@ func (e *Engine) drive(ctx context.Context, orgID int, dev outputdevice.OutputDe if on { edge = "on" } + if !dev.ConfirmsDrives() { + result = "sent" + } if err != nil { result = "error" e.log.Error().Err(err).Int("org_id", orgID).Int("output_device_id", dev.ID).Bool("on", on). @@ -286,6 +289,7 @@ func (e *Engine) drive(ctx context.Context, orgID int, dev outputdevice.OutputDe On: on, Cause: cause, Err: err, + Unconfirmed: !dev.ConfirmsDrives(), Latency: latency, DrivenAt: at, }); rerr != nil { diff --git a/backend/internal/geofence/engine_drives_test.go b/backend/internal/geofence/engine_drives_test.go index 0e4d2dadb..1b2970c1f 100644 --- a/backend/internal/geofence/engine_drives_test.go +++ b/backend/internal/geofence/engine_drives_test.go @@ -53,14 +53,14 @@ func TestEvaluate_RecordsEachDeviceDriveWithAlarmID(t *testing.T) { func TestEvaluate_FailedDriveRecordsErrorAndCountsUndelivered(t *testing.T) { s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil)}} - e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, &fakeDriver{err: errors.New("reader rpc: timeout after 5s")}) + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, &fakeDriver{err: errors.New("device unreachable: dial tcp 192.0.2.10:80: i/o timeout")}) defer e.Stop() before := undelivered(t) e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) drives := s.recordedDrives() - if len(drives) != 1 || drives[0].Err == nil || drives[0].Err.Error() != "reader rpc: timeout after 5s" { + if len(drives) != 1 || drives[0].Err == nil || drives[0].Err.Error() != "device unreachable: dial tcp 192.0.2.10:80: i/o timeout" { t.Fatalf("drives = %+v, want one row carrying the timeout", drives) } if got := undelivered(t) - before; got != 1 { @@ -160,3 +160,31 @@ func TestEvaluate_PresenceAgeOutRecordsOffRow(t *testing.T) { t.Errorf("age-out drive org = %d, want 42", s.driveOrgs[1]) } } + +// A drive over the broker is only handed off: a powered-off CS463 still lets +// the publish succeed. Its row must not claim the device was reached. +func TestEvaluate_BrokerDriveIsRecordedUnconfirmed(t *testing.T) { + gpo := outputdevice.OutputDevice{ID: 1, Transport: outputdevice.TransportMQTT, Type: outputdevice.TypeCSLGPO} + s := &fakeStore{devices: []outputdevice.OutputDevice{gpo}} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, &fakeDriver{}) + defer e.Stop() + + e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) + + drives := s.recordedDrives() + if len(drives) != 1 || !drives[0].Unconfirmed || drives[0].Err != nil { + t.Fatalf("drives = %+v, want one unconfirmed row with no error", drives) + } +} + +func TestEvaluate_HTTPDriveIsRecordedConfirmed(t *testing.T) { + s := &fakeStore{devices: []outputdevice.OutputDevice{dev(1, nil)}} + e := newTestEngine(Config{RSSIThreshold: -65, LatchTTL: time.Minute}, s, &fakeDriver{}) + defer e.Stop() + + e.Evaluate(context.Background(), 42, 9, time.Unix(1000, 0), []storage.ResolvedRead{read("EPC1", -50)}) + + if drives := s.recordedDrives(); len(drives) != 1 || drives[0].Unconfirmed { + t.Fatalf("drives = %+v, want one confirmed row", drives) + } +} diff --git a/backend/internal/geofence/metrics.go b/backend/internal/geofence/metrics.go index a5a603382..3c88af4f8 100644 --- a/backend/internal/geofence/metrics.go +++ b/backend/internal/geofence/metrics.go @@ -15,7 +15,7 @@ var ( metricFired = promauto.NewCounter(prometheus.CounterOpts{ Name: "geofence_alarms_fired_total", - Help: "Geofence boundary alarms fired (post-dedup).", + Help: "Geofence boundary alarms detected (post-dedup), whether or not a device was reached. See geofence_alarms_undelivered_total.", }) metricSuppressed = promauto.NewCounterVec(prometheus.CounterOpts{ @@ -31,11 +31,11 @@ var ( metricDrives = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "geofence_output_drives_total", Help: "Output-device drives sent by the geofence engine, by edge and result.", - }, []string{"edge", "result"}) // edge: on|off; result: ok|error + }, []string{"edge", "result"}) // edge: on|off; result: ok|sent|error metricUndelivered = promauto.NewCounter(prometheus.CounterOpts{ Name: "geofence_alarms_undelivered_total", - Help: "Alarms whose every output-device drive failed: detected, but no device was reached.", + Help: "Alarms whose every output-device drive returned an error. A drive over the broker that published cleanly is not counted: the reader does not confirm it.", }) metricDriveWriteErrors = promauto.NewCounter(prometheus.CounterOpts{ diff --git a/backend/internal/handlers/outputdevices/drives.go b/backend/internal/handlers/outputdevices/drives.go index 57050761b..e3f86ee3e 100644 --- a/backend/internal/handlers/outputdevices/drives.go +++ b/backend/internal/handlers/outputdevices/drives.go @@ -21,6 +21,7 @@ func (h *Handler) set(ctx context.Context, orgID int, dev outputdevice.OutputDev On: on, Cause: cause, Err: err, + Unconfirmed: !dev.ConfirmsDrives(), Latency: time.Since(drivenAt), DrivenAt: drivenAt, }); rerr != nil { diff --git a/backend/internal/models/outputdevice/outputdevice.go b/backend/internal/models/outputdevice/outputdevice.go index 53418d2de..ec9b5ba11 100644 --- a/backend/internal/models/outputdevice/outputdevice.go +++ b/backend/internal/models/outputdevice/outputdevice.go @@ -227,3 +227,10 @@ type UpdateOutputDeviceRequest struct { type OutputDeviceResponse struct { Data OutputDevice `json:"data"` } + +// ConfirmsDrives reports whether a drive's success means the device acted. +// An http device answers each call. An mqtt device is only published to: the +// broker's ack is all the caller gets (a cs463 gpo rides readercontrol's +// fire-and-forget Notify, a shelly a bare publish), so a powered-off reader +// still "succeeds". Its drives are recorded as sent, not ok (TRA-1314). +func (d OutputDevice) ConfirmsDrives() bool { return d.Transport != TransportMQTT } diff --git a/backend/internal/models/outputdevice/outputdevice_test.go b/backend/internal/models/outputdevice/outputdevice_test.go index b6418c984..ece1cfd55 100644 --- a/backend/internal/models/outputdevice/outputdevice_test.go +++ b/backend/internal/models/outputdevice/outputdevice_test.go @@ -119,3 +119,18 @@ func TestOutputDevice_ScanDeviceIDAndReaderBaseTopicJSON(t *testing.T) { t.Errorf("ReaderBaseTopic should not decode from JSON, got %q", back.ReaderBaseTopic) } } + +func TestConfirmsDrives(t *testing.T) { + if !(OutputDevice{Transport: TransportHTTP}).ConfirmsDrives() { + t.Error("an http device answers each drive") + } + if !(OutputDevice{}).ConfirmsDrives() { + t.Error("an unset transport is http, the default") + } + if (OutputDevice{Transport: TransportMQTT, Type: TypeCSLGPO}).ConfirmsDrives() { + t.Error("a cs463 gpo drive is fire-and-forget: the broker's ack is not the reader's") + } + if (OutputDevice{Transport: TransportMQTT, Type: TypeShellyGen4}).ConfirmsDrives() { + t.Error("a shelly over mqtt is published, not answered") + } +} diff --git a/backend/internal/storage/output_drives.go b/backend/internal/storage/output_drives.go index 70b9481b9..503cd9433 100644 --- a/backend/internal/storage/output_drives.go +++ b/backend/internal/storage/output_drives.go @@ -37,8 +37,12 @@ type OutputDriveRow struct { On bool Cause DriveCause Err error - Latency time.Duration - DrivenAt time.Time + // Unconfirmed marks a transport that only hands the drive off (the MQTT + // broker) and never hears from the device: a drive that sent cleanly is + // recorded as "sent", not "ok". A failure is an error either way. + Unconfirmed bool + Latency time.Duration + DrivenAt time.Time } // OutputDrive is one recorded drive, as read back. @@ -48,7 +52,7 @@ type OutputDrive struct { AlarmEventID *int64 Edge string // on | off Cause string - Result string // ok | error + Result string // ok | sent | error Error *string LatencyMs int DrivenAt time.Time @@ -68,6 +72,9 @@ func (s *Storage) RecordOutputDrive(ctx context.Context, orgID int, row OutputDr edge = "on" } result := "ok" + if row.Unconfirmed { + result = "sent" + } var errText *string if row.Err != nil { result = "error" diff --git a/backend/internal/storage/output_drives_integration_test.go b/backend/internal/storage/output_drives_integration_test.go index 65299e109..586906481 100644 --- a/backend/internal/storage/output_drives_integration_test.go +++ b/backend/internal/storage/output_drives_integration_test.go @@ -59,7 +59,7 @@ func TestRecordOutputDrive_ListsEachDeviceNewestFirst(t *testing.T) { t0 := time.Now().Truncate(time.Millisecond) require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ OutputDeviceID: horn, AlarmEventID: &alarmID, On: true, Cause: storage.DriveCauseAlarm, - Err: errors.New("reader rpc: timeout after 5s"), Latency: 5003 * time.Millisecond, DrivenAt: t0, + Err: errors.New("device unreachable: dial tcp 192.0.2.10:80: i/o timeout"), Latency: 5003 * time.Millisecond, DrivenAt: t0, })) require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ OutputDeviceID: horn, On: false, Cause: storage.DriveCauseAgeOut, Latency: 12 * time.Millisecond, DrivenAt: t0.Add(time.Minute), @@ -83,7 +83,7 @@ func TestRecordOutputDrive_ListsEachDeviceNewestFirst(t *testing.T) { assert.Equal(t, "alarm", got[1].Cause) assert.Equal(t, "error", got[1].Result) require.NotNil(t, got[1].Error) - assert.Equal(t, "reader rpc: timeout after 5s", *got[1].Error) + assert.Equal(t, "device unreachable: dial tcp 192.0.2.10:80: i/o timeout", *got[1].Error) require.NotNil(t, got[1].AlarmEventID) assert.Equal(t, alarmID, *got[1].AlarmEventID) assert.Equal(t, 5003, got[1].LatencyMs) @@ -141,3 +141,27 @@ func TestOutputDrives_OKRowCannotCarryErrorText(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "output_drives_error_iff_failed") } + +// A broker-only drive that published cleanly is "sent": handed off, never +// answered. It must not read as "ok". +func TestRecordOutputDrive_UnconfirmedSuccessIsSent(t *testing.T) { + db := testutil.SetupTestDBFull(t) + ctx := context.Background() + orgID := testutil.CreateTestAccount(t, db.AdminPool) + horn := createOutput(t, db, orgID, "horn") + + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: horn, On: true, Cause: storage.DriveCauseTest, Unconfirmed: true, DrivenAt: time.Now(), + })) + require.NoError(t, db.Store.RecordOutputDrive(ctx, orgID, storage.OutputDriveRow{ + OutputDeviceID: horn, On: false, Cause: storage.DriveCauseTest, Unconfirmed: true, + Err: errors.New("broker unreachable"), DrivenAt: time.Now().Add(time.Second), + })) + + got, err := db.Store.ListOutputDrives(ctx, orgID, horn, 10) + require.NoError(t, err) + require.Len(t, got, 2) + assert.Equal(t, "error", got[0].Result, "a failed publish is still an error") + assert.Equal(t, "sent", got[1].Result) + assert.Nil(t, got[1].Error) +} diff --git a/backend/migrations/000055_output_drives.up.sql b/backend/migrations/000055_output_drives.up.sql index 4bac8f9fe..c735feb20 100644 --- a/backend/migrations/000055_output_drives.up.sql +++ b/backend/migrations/000055_output_drives.up.sql @@ -5,8 +5,16 @@ -- say which devices were driven or whether the drive reached them: one fire -- drives every device at the location, each suppressed or driven on its own, -- and a drive to an offline reader fails with only a log line to show for it. --- These rows are the delivery record. An alarm whose drive rows are all --- result = 'error' (or that has none) was detected but not delivered. +-- These rows are the delivery record: an alarm's drive rows say which devices +-- were reached. They are best-effort writes, so a missing row is a lost +-- record, not a missed drive. +-- +-- result: +-- ok the device answered (http transport) +-- sent handed to the MQTT broker, which is all an mqtt transport hears: a +-- cs463 gpo drive is fire-and-forget, and a powered-off reader still +-- lets the publish succeed. Not proof the output moved. +-- error the drive failed; error says how -- -- cause says what drove it: -- alarm the ON a fire sent; alarm_event_id names the fire, unless the @@ -33,7 +41,7 @@ CREATE TABLE output_drives ( edge TEXT NOT NULL CHECK (edge IN ('on', 'off')), cause TEXT NOT NULL CHECK (cause IN ('alarm', 'refresh', 'age_out', 'test', 'reset', 'delete')), - result TEXT NOT NULL CHECK (result IN ('ok', 'error')), + result TEXT NOT NULL CHECK (result IN ('ok', 'sent', 'error')), error TEXT, latency_ms INT NOT NULL CHECK (latency_ms >= 0), driven_at TIMESTAMPTZ NOT NULL, @@ -48,7 +56,8 @@ CREATE INDEX idx_output_drives_device_time ON output_drives(org_id, output_devic CREATE INDEX idx_output_drives_alarm ON output_drives(alarm_event_id) WHERE alarm_event_id IS NOT NULL; COMMENT ON TABLE output_drives IS 'One row per output-device drive and its result. The delivery record for alarm_events.'; -COMMENT ON COLUMN output_drives.error IS 'Transport error text when result = error (RPC timeout, reader error code), capped at 1000 bytes.'; +COMMENT ON COLUMN output_drives.error IS 'Transport error text when result = error (HTTP failure, broker publish failure, device config error), capped at 1000 bytes.'; +COMMENT ON COLUMN output_drives.result IS 'ok = the device answered; sent = handed to the MQTT broker, unconfirmed; error = failed.'; COMMENT ON COLUMN output_drives.latency_ms IS 'Time the drive call took, success or failure.'; -- RLS: org isolation, USING-only, identical shape to alarm_events. diff --git a/backend/migrations/checksums.txt b/backend/migrations/checksums.txt index c5122cdbc..33b9b90ff 100644 --- a/backend/migrations/checksums.txt +++ b/backend/migrations/checksums.txt @@ -114,4 +114,4 @@ e63f15be38e76b1189c299b952ddb19511b655838b8fe32c169f8e64cd838f16 000053_suppres fb64d05971920ea00448d7f7804802382d133148a85ac0db33584129e8baa469 000054_notification_event_sources.down.sql ea8b874ac0efc8108144343d62a8a3b350d5a3d936263707589989319f605ee2 000054_notification_event_sources.up.sql b2a68d3cec860c34ece04760fe71ca5fca77777fa4de8aeb2851ce8eef4b02ab 000055_output_drives.down.sql -eef009d88b90f10fc668b0b5f7b373da17f4e4cce63c8a6de2930ce8174201da 000055_output_drives.up.sql +75d4bdad89a9c7df8164b37ac408f3ce739d3ab20bc976ad547600e15ccfc2de 000055_output_drives.up.sql