diff --git a/.changeset/send-on-arrival.md b/.changeset/send-on-arrival.md new file mode 100644 index 000000000..2d9ecd760 --- /dev/null +++ b/.changeset/send-on-arrival.md @@ -0,0 +1,5 @@ +--- +"ftw": minor +--- + +A phone whose session runs through FTW Cloud now gets a frame as each driver poll lands, instead of on the box's next one-second pulse, with a beat each second when no poll comes. A frame goes on every poll whether or not a value changed, so its timing shows the poll rhythm and never which reading moved. While a phone watches, the readable copy's `live` also goes when a reading lands, still at most once a second. Sessions over the relay keep their fixed cadence. Lane 0's invariant changes from "fixed length and cadence" to "fixed length; fixed cadence over the relay". diff --git a/AGENTS.md b/AGENTS.md index 1b2e6938c..110dbadd9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -68,8 +68,12 @@ Read [docs/architecture.md](docs/architecture.md) for the system map and - Every clamp protects a quantified hardware or control risk. - Persistent device state is keyed by stable hardware identity, not a YAML name. - Database queries stay in [`go/internal/state`](go/internal/state). -- Lane 0 frames have a fixed length and cadence; the padding is a privacy - control. The proposed FTW Cloud uplink is not padded, so it sends lane 1 +- Lane 0 frames have a fixed length; the padding is a privacy control. Over + the relay they keep a fixed cadence. Through FTW Cloud a frame goes on + every arrival, each finished driver poll whether or not a value moved, + plus a beat when none has gone for the subscribed period, so the timing + shows the poll rhythm and never which reading moved (Fredrik, 16 October + 2026). The proposed FTW Cloud uplink is not padded, so it sends lane 1 frames without the bulk buckets and never lane 0. A phone's own session rides inside it as `tun`, with its lane 0 padding intact. - FTW Cloud may hold a readable copy of what the owner enabled. It carries the diff --git a/docs/adr/0008-ftw-cloud.md b/docs/adr/0008-ftw-cloud.md index 8ff1cfa13..a601c2d82 100644 --- a/docs/adr/0008-ftw-cloud.md +++ b/docs/adr/0008-ftw-cloud.md @@ -288,6 +288,7 @@ rules, the app asks again. A public launch needs C2–C4 and the legal work. | Composing state costs too much CPU on a Pi | Bench in C2 | | Signatures cost data on a 4G box, about 150 MB a month | Measure in C2 | | The network sees message sizes and timing | The privacy notice | +| With the copy off, the service sees from the tunnel's timing when the box reads its devices while a phone watches | The privacy notice; a frame goes on every poll whether or not a value moved, frames keep their size, and a hidden session keeps its 5 s | | Apple sees the home id, event codes and Live Activity state | The privacy notice; no names in arguments; the M7 spike | | Claims or defaults drift back | Claim tests; a missing grant reads as off | | A paid app meets App Store purchase rules | Decide the payment path before charging | diff --git a/docs/protocol-cloud.md b/docs/protocol-cloud.md index cad168358..ce4502877 100644 --- a/docs/protocol-cloud.md +++ b/docs/protocol-cloud.md @@ -615,6 +615,10 @@ back `box.unreachable` until the box returns or an hour passes. - Two messages skip the period. The box looks at its source table every second, and a `live` whose table changed goes then, with `sources` and `fieldSources`. An `hb` goes when it is due. +- While a `watch` is in force, a driver poll that changed `live` sends it at + once if the last `live` left a period ago or more, and otherwise when that + period ends. So the copy costs what it did, and a reading from a device + read once a second waits for nothing. - `plan` goes only when `decisionId` changes, `prices` only when its hash changes, `tiles` when a bucket closes, and `event` at once. - Every 15 minutes, and after a `resync`, the box sends a full `live` with @@ -826,9 +830,37 @@ Noise session, in the app protocol's `hello_ok.cloud`. The phone's own session with its box, `ftw.session.v1` (Noise IK to the box's static key, as the app protocol specifies), rides inside the app -session. Nothing in it changes: lane 0 keeps its buckets and its cadence, so -the service sees when a message goes, not what it says or how large a command -is. +session. Its frames and buckets do not change, so the service sees when a +message goes, not what it says or how large a command is. Its timing does: +lane 0 goes as readings land, not on the relay's one-second pulse. + +### Lane 0 through the tunnel + +- When a driver finishes a poll, the box sends each visible session one + frame: a `delta` of what changed, or a `tick` when nothing did. It waits + 5 ms for polls that finish with it, so they go in one frame, and sends at + most five such frames a second on average, three at once. +- When nothing has gone for the subscribed period, 1 s while the phone shows + the stream and 5 s while it is hidden, the box sends a `tick`, or a `delta` + if something changed. A hidden session gets only these. +- A source table too large for the bucket goes in a snapshot on bulk, sent + in line with lane 0 so no reading sent after it reaches the phone first. + Every lane 0 frame keeps its bucket. +- Past 60 % of the tunnel's 24-hour budget a session gets the beat alone, + and past 90 % no readings, so handshakes, command answers and ends still + go. +- Over the relay nothing changes: one frame per pulse, every 1 s or 5 s. + +**What the service learns.** It cannot read a frame, and every lane 0 frame +is the same size. While a phone watches, a frame goes on every finished +poll, changed or not, so their timing shows when the box reads its devices +and never which reading moved, by how much or which way. A snapshot is +larger, and goes when the control revision or the source table changes. +With the copy off, the poll rhythm is new to the service, for the minutes +someone watches. With the copy on, the service reads in `live` the values of +the fields the grant lists, and the rhythm adds when the box read them. +While no phone has a session nothing goes, and a hidden session sends every +5 s whatever happens. ### `tun` @@ -1360,6 +1392,8 @@ constants, so `srcfl/ftw-cloud` need not spell them by hand. | Largest message from a phone | 4 KiB; a `tun` message 16,664 bytes and Noise's 24 | | `tun` per home and direction | 240 messages, then 60 a second; 4 MiB, then 512 KiB a second | | `tun` queues and budget on the box | 64 lane 0, 256 bulk paced to 90 % of the limits; 250 MB per 24 hours of uptime | +| Lane 0 through the tunnel | On each finished poll, after 5 ms; at most 5 frames a second, 3 at once; a beat after 1 s quiet, 5 s while hidden; pushes stop at 60 % of the tunnel budget, beats at 90 % | +| `live` while watched | On each poll that changed it, at most once per period | | Box handshakes through the tunnel | 12 a minute per device, 240 in all | | Pending pairing entry | One QR code, for its own ten minutes; admits one handshake | | Box session up after TLS | One round trip: its message 1 rides in `now.tun` | diff --git a/go/cmd/ftw/app_link.go b/go/cmd/ftw/app_link.go index a64f0e9b9..a5335ef98 100644 --- a/go/cmd/ftw/app_link.go +++ b/go/cmd/ftw/app_link.go @@ -824,6 +824,7 @@ func startAppLink( Deadman: deadman, Handler: handlerFor(false), ExchangeHandler: handlerFor(true), + Arrivals: arrivalsFor(ctx, tel), }) if err != nil { return enroll, nil, enabled, err @@ -843,6 +844,17 @@ func startAppLink( return enroll, uplink, true, nil } +// arrivalsFor hears each finished driver poll until ctx ends, or nothing +// without a store. +func arrivalsFor(ctx context.Context, tel *telemetry.Store) <-chan struct{} { + if tel == nil { + return nil + } + arrivals, stop := tel.Arrivals() + context.AfterFunc(ctx, stop) + return arrivals +} + func siteMeterName(ctrl *control.State, ctrlMu *sync.Mutex) string { ctrlMu.Lock() defer ctrlMu.Unlock() diff --git a/go/cmd/ftw/cloud_link.go b/go/cmd/ftw/cloud_link.go index 5d475265d..1c676c98b 100644 --- a/go/cmd/ftw/cloud_link.go +++ b/go/cmd/ftw/cloud_link.go @@ -320,6 +320,7 @@ func startCloudLink( Started: processStarted, ClockSynced: cloudlink.ClockSynced, Notify: notices.notify, + Arrivals: arrivalsFor(ctx, tel), } // Set only with a table: a typed nil would read as a tunnel. if tunnel != nil { diff --git a/go/internal/appproto/handler.go b/go/internal/appproto/handler.go index dd1ba6f39..f089df5b6 100644 --- a/go/internal/appproto/handler.go +++ b/go/internal/appproto/handler.go @@ -176,7 +176,12 @@ type Handler struct { // one hertz while visible, 0.2 hertz while hidden. cadenceEvery uint8 cadenceLeft uint8 - seq uint64 + // lastFrameMs is the box uptime when this session's last reading + // left: a snapshot, a delta or a tick. Beat measures silence from it. + lastFrameMs int64 + // resync says a snapshot never left: the next frame sends one. + resync bool + seq uint64 // lastSent is the value the client was last told for each field. Only // fields actually put on the wire are recorded, so anything dropped to // fit a bucket is simply still different next tick and goes then. @@ -410,7 +415,16 @@ func (h *Handler) Handle(ctx context.Context, frame []byte) error { h.log.Warn("undecodable frame dropped", "err", err) return nil } - return h.dispatch(ctx, f.Envelope) + return dropped(h.dispatch(ctx, f.Envelope)) +} + +// dropped is err, unless it says only that the carrier lost one frame: the +// session goes on after that. +func dropped(err error) error { + if errors.Is(err, ErrDropped) { + return nil + } + return err } func (h *Handler) dispatch(ctx context.Context, env Envelope) error { @@ -647,9 +661,10 @@ func (h *Handler) applySub(sub Sub) error { h.mu.Unlock() // Keep subscribed false until the full snapshot is on the wire. Otherwise - // the one-second ticker can race ahead and send a delta before the snapshot - // that gives it meaning. - if err := h.sendSnapshot(); err != nil { + // the pulse or an arrival can race ahead and send a delta before the + // snapshot that gives it meaning. A snapshot the carrier dropped is sent + // again with the first frame. + if err := h.sendSnapshot(); err != nil && !errors.Is(err, ErrDropped) { return err } h.mu.Lock() @@ -658,8 +673,8 @@ func (h *Handler) applySub(sub Sub) error { return nil } -// sendSnapshot puts the full site state on the bulk lane and records what the -// client now knows. +// sendSnapshot puts the full site state on the bulk lane, in line with lane +// 0, and records what the client now knows once it has left. func (h *Handler) sendSnapshot() error { uptimeMs := h.cfg.Clock.UptimeMs() snap := h.cfg.Site.Snapshot() @@ -670,7 +685,25 @@ func (h *Handler) sendSnapshot() error { blocked = []string{} } + f, err := h.bulkFrame(MsgSnap, nil, Snap{ + UptimeMs: uptimeMs, + ControlRev: snap.ControlRev, + Dict: h.dict, + Fields: fields, + Sources: sources, + DispatchBlockedBy: blocked, + }) + if err == nil { + err = h.sendInLine(f) + } + h.mu.Lock() + defer h.mu.Unlock() + if err != nil { + h.resync = true + return err + } + h.resync = false h.lastSent = make(map[string]int64, len(fields)) for k, v := range fields { h.lastSent[k] = v @@ -678,22 +711,17 @@ func (h *Handler) sendSnapshot() error { h.lastSources = sources h.lastBlocked = blocked h.lastRev = snap.ControlRev - h.mu.Unlock() - - return h.sendBulk(MsgSnap, nil, Snap{ - UptimeMs: uptimeMs, - ControlRev: snap.ControlRev, - Dict: h.dict, - Fields: fields, - Sources: sources, - DispatchBlockedBy: blocked, - }) + h.lastFrameMs = uptimeMs + return nil } // Tick consumes one global one-second pulse. The handler gates that pulse to // this subscription's 1 Hz or 0.2 Hz cadence. On each due pulse it emits a // frame even when nothing changed, so the relay cannot distinguish an idle // house from an active one within the chosen cadence. +// +// Only a session over the relay ticks. A session that FTW Cloud carries is +// streamed with Push and Beat instead (docs/protocol-cloud.md, "Tunnel"). func (h *Handler) Tick() error { if h.isAgent() { if err := h.agentTick(); err != nil { @@ -711,6 +739,53 @@ func (h *Handler) Tick() error { return nil } h.cadenceLeft = h.cadenceEvery + h.mu.Unlock() + return h.stream() +} + +// Visible says Push would send: the session streams and someone sees it. +func (h *Handler) Visible() bool { + h.mu.Lock() + defer h.mu.Unlock() + return h.subscribed && h.cadenceEvery == 1 +} + +// Push sends a frame now: a driver has just finished a poll. A delta of +// what changed, or a tick when nothing did, so a frame's timing shows when +// the box read its devices and never which reading moved. Nothing while +// hidden, where the session asked for one frame in five seconds and Beat +// sends it. +func (h *Handler) Push() error { + if !h.Visible() { + return nil + } + return h.stream() +} + +// Beat sends a frame when none has left for the subscribed period, so the +// phone hears the session is alive while nothing in the house moves, and +// the hidden session gets its one frame in five seconds. Call it more often +// than the period; it decides. +func (h *Handler) Beat() error { + h.mu.Lock() + if !h.subscribed { + h.mu.Unlock() + return nil + } + period := int64(h.cadenceEvery) * 1000 + quiet := h.cfg.Clock.UptimeMs() - h.lastFrameMs + h.mu.Unlock() + if quiet < period { + return nil + } + return h.stream() +} + +// stream sends one lane 0 frame: a delta of what changed since the client +// was last told, or a tick when nothing did. A frame the carrier dropped is +// not recorded as sent, so the next one carries the same news. +func (h *Handler) stream() error { + h.mu.Lock() bucket := h.bucket h.mu.Unlock() @@ -732,11 +807,11 @@ func (h *Handler) Tick() error { // whole set instead. fields := fieldValues(snap, h.modes) h.mu.Lock() - stale := snap.ControlRev != h.lastRev || anyGone(h.lastSent, fields) + stale := h.resync || snap.ControlRev != h.lastRev || anyGone(h.lastSent, fields) h.mu.Unlock() if stale { if err := h.sendSnapshot(); err != nil { - return err + return dropped(err) } } @@ -759,7 +834,7 @@ func (h *Handler) Tick() error { h.mu.Unlock() if len(changed) == 0 && !sourcesMoved { - return h.sendControl(MsgTick, nil, Tick{Seq: seq, UptimeMs: uptimeMs}, 0, bucket) + return h.sendTick(seq, uptimeMs, bucket) } body := Delta{Seq: seq, UptimeMs: uptimeMs, Fields: changed} @@ -776,9 +851,19 @@ func (h *Handler) Tick() error { if err != nil { return err } + if sourcesMoved && body.Sources == nil { + // The source table did not fit the bucket, and a source going + // stale is the news the freshness model exists to show. The + // snapshot carries it, with every reading; lane 0 still gets its + // frame. + if err := h.sendSnapshot(); err != nil { + return dropped(err) + } + return h.sendTick(seq, uptimeMs, bucket) + } if err := h.sendControl(MsgDelta, nil, body, flags, bucket); err != nil { - return err + return dropped(err) } h.mu.Lock() @@ -789,6 +874,18 @@ func (h *Handler) Tick() error { h.lastSources = sources h.lastBlocked = blocked } + h.lastFrameMs = uptimeMs + h.mu.Unlock() + return nil +} + +// sendTick sends lane 0's frame when nothing changed. +func (h *Handler) sendTick(seq uint64, uptimeMs int64, bucket int) error { + if err := h.sendControl(MsgTick, nil, Tick{Seq: seq, UptimeMs: uptimeMs}, 0, bucket); err != nil { + return dropped(err) + } + h.mu.Lock() + h.lastFrameMs = uptimeMs h.mu.Unlock() return nil } @@ -799,7 +896,8 @@ func (h *Handler) Tick() error { // The bucket never grows and the frame never fragments, because both would // make frame size a function of what happened in the house — which is the one // thing lane 0's constancy exists to hide. Dropped fields are simply not -// recorded as sent, so they are still different on the next tick and go then. +// recorded as sent, so they are still different on the next frame and go +// then. A source table that does not fit goes in a snapshot (stream). // Field ids ascend by importance: 1-9 are the frozen core view. func (h *Handler) fitDelta(body *Delta, bucket int) (map[string]int64, uint8, error) { const headerBytes = 6 @@ -1092,12 +1190,14 @@ func (h *Handler) acceptCmd(cmd Cmd, uptimeMs int64) (took bool, err error) { if rec, fresh := h.cfg.Commands.remember(root, cmd.CmdID, ack, uptimeMs); !fresh { return false, h.replayCmd(rec) } - if err := h.sendCmdAck(ack); err != nil { + if err := h.sendCmdAck(ack); err != nil && !errors.Is(err, ErrDropped) { // The box has not acted, and the app never heard the ack. Keeping // the id would answer its retry with an ack and no result. h.cfg.Commands.forget(root, cmd.CmdID) return false, err } + // An ack the carrier dropped is one lost frame: the box acts, and the + // result tells the app. return true, nil } @@ -1290,7 +1390,7 @@ func (h *Handler) Terminate(reason string) error { h.subscribed = false bucket := h.bucket h.mu.Unlock() - return h.sendControl(MsgSessionTerminate, nil, SessionTerminate{Reason: reason}, 0, bucket) + return dropped(h.sendControl(MsgSessionTerminate, nil, SessionTerminate{Reason: reason}, 0, bucket)) } // -------------------------------------------------------------------------- @@ -1341,19 +1441,27 @@ func (h *Handler) sendControl(t string, id *uint32, body any, flags uint8, bucke // sendBulk sizes the bucket from the encoded body rather than a guess, so a // long mode catalogue or a wide source table cannot silently overrun. func (h *Handler) sendBulk(t string, id *uint32, body any) error { - env, err := newEnvelope(t, id, body) + f, err := h.bulkFrame(t, id, body) if err != nil { return err } + return h.send(f) +} + +func (h *Handler) bulkFrame(t string, id *uint32, body any) (Frame, error) { + env, err := newEnvelope(t, id, body) + if err != nil { + return Frame{}, err + } raw, err := EncodeEnvelope(env) if err != nil { - return err + return Frame{}, err } bucket := bulkBucketFor(len(raw)) if bucket == 0 { - return fmt.Errorf("appproto: %s is %d bytes, larger than the biggest bulk bucket", t, len(raw)) + return Frame{}, fmt.Errorf("appproto: %s is %d bytes, larger than the biggest bulk bucket", t, len(raw)) } - return h.send(Frame{Lane: LaneBulk, Bucket: bucket, Envelope: env}) + return Frame{Lane: LaneBulk, Bucket: bucket, Envelope: env}, nil } func (h *Handler) send(f Frame) error { @@ -1363,3 +1471,17 @@ func (h *Handler) send(f Frame) error { } return h.cfg.Sender.Send(bytes) } + +// sendInLine sends a bulk frame in line with lane 0 where the carrier can +// (InLineSender), and as any other frame where it cannot. +func (h *Handler) sendInLine(f Frame) error { + s, ok := h.cfg.Sender.(InLineSender) + if !ok { + return h.send(f) + } + bytes, err := h.cfg.Codec.EncodeFrame(f) + if err != nil { + return fmt.Errorf("appproto: encode %s: %w", f.Envelope.T, err) + } + return s.SendInLine(bytes) +} diff --git a/go/internal/appproto/ports.go b/go/internal/appproto/ports.go index 54bf5708b..4c2ac2b69 100644 --- a/go/internal/appproto/ports.go +++ b/go/internal/appproto/ports.go @@ -2,6 +2,7 @@ package appproto import ( "context" + "errors" "time" "github.com/srcfl/ftw/go/internal/control" @@ -284,6 +285,21 @@ type PriceReader interface { } // Sender delivers an encoded frame to the carrier. +// +// Send returns ErrDropped, or an error wrapping it, when the carrier lost +// that one frame and counted it. The session goes on: what the frame carried +// is not recorded as sent, so the next frame carries it again. type Sender interface { Send(frame []byte) error } + +// ErrDropped is one frame the carrier lost and counted. See Sender. +var ErrDropped = errors.New("appproto: the carrier dropped a frame") + +// InLineSender is a Sender whose carrier sends bulk after lane 0. SendInLine +// keeps one bulk frame in line with lane 0 instead, so a reading sent after +// it never reaches the phone first. The snapshot goes this way: a delta that +// overtook it would be applied and then overwritten with older values. +type InLineSender interface { + SendInLine(frame []byte) error +} diff --git a/go/internal/appproto/push_test.go b/go/internal/appproto/push_test.go new file mode 100644 index 000000000..23c112e58 --- /dev/null +++ b/go/internal/appproto/push_test.go @@ -0,0 +1,248 @@ +package appproto + +import ( + "fmt" + "testing" +) + +// A session through FTW Cloud hears every poll the moment it lands: a delta +// of what changed, or a tick, so the frame's timing shows when the box read +// its devices and never which reading moved. +func TestPushSendsAFrameOnEveryPoll(t *testing.T) { + h, box, rec, _ := newRig(t) + subscribe(t, h, rec) + + if err := h.Push(); err != nil { + t.Fatal(err) + } + if f := rec.only(t, MsgTick); f.lane != LaneControl || f.size != 512 { + t.Fatalf("a poll that changed nothing sent lane %d, %d bytes; want a 512-byte tick", f.lane, f.size) + } + + rec.reset() + box.snap.GridW = 1450 + if err := h.Push(); err != nil { + t.Fatal(err) + } + f := rec.only(t, MsgDelta) + if f.lane != LaneControl || f.size != 512 { + t.Fatalf("pushed delta: lane %d, %d bytes; want lane 0 in its 512-byte bucket", f.lane, f.size) + } + d := body[Delta](t, f) + if len(d.Fields) != 1 || d.Fields[fidKey(FidGridW)] != 1450 { + t.Fatalf("pushed delta carries %v, want only grid_w 1450", d.Fields) + } + + rec.reset() + if err := h.Push(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[tick]" { + t.Fatalf("the same reading again sent %s, want a tick", got) + } +} + +// Hidden, the phone asked for one frame in five seconds. A poll does not +// break that; the beat carries what changed when the five seconds are up. +func TestPushWaitsForTheBeatWhileHidden(t *testing.T) { + h, box, rec, clock := newRig(t) + subscribe(t, h, rec) + deliver(t, h, MsgSub, nil, Sub{Bucket: 512, Hz: 0.2}) + + box.snap.GridW = 1450 + if err := h.Push(); err != nil { + t.Fatal(err) + } + clock.uptimeMs += 4_999 + if err := h.Beat(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[]" { + t.Fatalf("hidden session sent %s within its five seconds", got) + } + clock.uptimeMs++ + if err := h.Beat(); err != nil { + t.Fatal(err) + } + if d := body[Delta](t, rec.only(t, MsgDelta)); d.Fields[fidKey(FidGridW)] != 1450 { + t.Fatalf("the hidden beat carries %v, want grid_w 1450", d.Fields) + } +} + +// Silence on lane 0 means the session is gone, so a quiet house still gets +// a tick once per subscribed period, counted from the last frame of any kind. +func TestBeatTicksOnlyAfterAQuietPeriod(t *testing.T) { + h, box, rec, clock := newRig(t) + subscribe(t, h, rec) + + clock.uptimeMs += 999 + if err := h.Beat(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[]" { + t.Fatalf("beat %d ms after the snapshot sent %s", 999, got) + } + clock.uptimeMs++ + if err := h.Beat(); err != nil { + t.Fatal(err) + } + tick := rec.only(t, MsgTick) + if tick.size != 512 { + t.Fatalf("beat tick is %d bytes, want the 512-byte bucket", tick.size) + } + + // A pushed delta resets the quiet period. + rec.reset() + clock.uptimeMs += 500 + box.snap.GridW = 1450 + if err := h.Push(); err != nil { + t.Fatal(err) + } + clock.uptimeMs += 999 + if err := h.Beat(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[delta]" { + t.Fatalf("got %s; a beat within a second of the delta must send nothing", got) + } + clock.uptimeMs++ + if err := h.Beat(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[delta tick]" { + t.Fatalf("got %s, want the delta and then one tick", got) + } + + // One sequence across pushes and beats, with no gap for a poll that + // changed nothing. + frames := rec.snapshot() + if d, k := body[Delta](t, frames[0]), body[Tick](t, frames[1]); d.Seq != 2 || k.Seq != 3 { + t.Fatalf("seq %d then %d, want 2 then 3", d.Seq, k.Seq) + } +} + +// A frame the carrier lost is not news the phone has: the next frame +// carries it again, and the quiet period runs from the last frame that left. +func TestADroppedFrameIsSentAgain(t *testing.T) { + h, box, rec, clock := newRig(t) + subscribe(t, h, rec) + + rec.fail = fmt.Errorf("tun: %w", ErrDropped) + box.snap.GridW = 1450 + if err := h.Push(); err != nil { + t.Fatalf("a dropped frame failed the session: %v", err) + } + rec.fail = nil + if err := h.Push(); err != nil { + t.Fatal(err) + } + if d := body[Delta](t, rec.only(t, MsgDelta)); d.Fields[fidKey(FidGridW)] != 1450 { + t.Fatalf("the next frame carries %v, want grid_w 1450 again", d.Fields) + } + + rec.reset() + clock.uptimeMs += 1000 + rec.fail = ErrDropped + if err := h.Beat(); err != nil { + t.Fatal(err) + } + rec.fail = nil + if err := h.Beat(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[tick]" { + t.Fatalf("after a dropped beat the next beat sent %s, want a tick", got) + } +} + +// A snapshot the carrier lost goes again with the first frame, before it. +func TestADroppedSnapshotGoesAgainFirst(t *testing.T) { + h, _, rec, _ := newRig(t) + deliver(t, h, MsgHello, nil, Hello{Proto: ProtoRange{Min: ProtoMin, Max: ProtoMax}}) + rec.reset() + rec.fail = ErrDropped + deliver(t, h, MsgSub, nil, Sub{Bucket: 512, Hz: 1}) + if !h.Subscribed() { + t.Fatal("a dropped snapshot left the session unsubscribed") + } + rec.fail = nil + if err := h.Push(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[snap tick]" { + t.Fatalf("got %s, want the snapshot again and then lane 0's frame", got) + } +} + +// inLine records which frames went in line with lane 0. +type inLine struct { + *recorder + types []string +} + +func (s *inLine) SendInLine(frame []byte) error { + f, err := testCodec{}.DecodeFrame(frame) + if err != nil { + return err + } + s.types = append(s.types, f.Envelope.T) + return s.Send(frame) +} + +// Through FTW Cloud bulk waits behind lane 0. The snapshot must not: a delta +// that overtook it would be applied, then overwritten with older values. +func TestTheSnapshotGoesInLineWithLaneZero(t *testing.T) { + h, box, rec, _ := newRig(t) + sender := &inLine{recorder: rec} + h.cfg.Sender = sender + subscribe(t, h, rec) + box.snap.ControlRev++ + if err := h.Push(); err != nil { + t.Fatal(err) + } + if fmt.Sprint(sender.types) != "[snap snap]" { + t.Fatalf("in line: %v; want both snapshots and nothing else", sender.types) + } +} + +// Six drivers' source table does not fit a 512-byte lane 0 frame. When one +// goes stale the phone must still hear it, so the table goes in a snapshot, +// once, and the frame after it is a tick. +func TestSixDriversSourcesThatDoNotFitGoInASnapshot(t *testing.T) { + h, box, rec, clock := newRig(t) + box.snap.Sources = nil + for _, name := range []string{"Sungrow SH10RT hybrid inverter", "Sungrow SBR battery stack", + "Easee Home charger in the carport", "P1 meter in the hall cupboard", "Ferroamp EnergyHub XL", + "Shelly Pro 3EM on the heat pump"} { + box.snap.Sources = append(box.snap.Sources, Source{ + ID: "driver." + name[:6], Kind: "driver", Name: name, LastOkUptimeMs: 60_000, StaleAfterMs: 5_000, + }) + } + subscribe(t, h, rec) + + // One driver stops answering: its state moves to lagging. + box.snap.Sources[3].LastOkUptimeMs = 54_000 + clock.uptimeMs += 1_000 + probe := Delta{Fields: map[string]int64{}, Sources: wireSources(box.snap, clock.uptimeMs), DispatchBlockedBy: []string{}} + if _, flags, _ := h.fitDelta(&probe, 512); flags&FlagTrunc == 0 { + t.Fatal("six drivers' sources fit a 512-byte delta; the test needs a table that does not") + } + if err := h.Push(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[snap tick]" { + t.Fatalf("got %s, want the table in a snapshot and lane 0's tick", got) + } + snap := body[Snap](t, rec.only(t, MsgSnap)) + if src := snap.Sources["driver.P1 met"]; src.State != SourceLagging { + t.Fatalf("the snapshot says the quiet meter is %q, want lagging", src.State) + } + + rec.reset() + if err := h.Push(); err != nil { + t.Fatal(err) + } + if got := rec.types(); got != "[tick]" { + t.Fatalf("the next poll sent %s; the table was already told", got) + } +} diff --git a/go/internal/appproto/telemetry_test.go b/go/internal/appproto/telemetry_test.go index 3b907f84b..67afd0e5e 100644 --- a/go/internal/appproto/telemetry_test.go +++ b/go/internal/appproto/telemetry_test.go @@ -327,7 +327,7 @@ func TestUnchangedSourcesAreNotResent(t *testing.T) { // The bucket never grows and the frame never fragments, because both would // make frame size a function of what happened in the house. -func TestAnOversizedDeltaTruncatesInsteadOfGrowing(t *testing.T) { +func TestAnOversizedDeltaGoesAsASnapshotNotALargerFrame(t *testing.T) { h, box, rec, clock := newRig(t) // A site with many devices, so the source table alone overruns lane 0. @@ -349,50 +349,16 @@ func TestAnOversizedDeltaTruncatesInsteadOfGrowing(t *testing.T) { t.Fatal(err) } - f := rec.only(t, MsgDelta) - if f.size != 512 { - t.Fatalf("truncated delta is %d bytes, want the 512-byte bucket", f.size) - } - if f.flags&FlagTrunc == 0 { - t.Fatal("a delta that did not fit was sent without TRUNC") + // Lane 0 keeps its bucket; the table and the reading go in a snapshot. + if got := rec.types(); got != "[snap tick]" { + t.Fatalf("got %s, want a snapshot and lane 0's tick", got) } - d := body[Delta](t, f) - if d.Sources != nil { - t.Fatal("the source table survived truncation; readings should have won the space") + if f := rec.only(t, MsgTick); f.size != 512 { + t.Fatalf("lane 0's frame is %d bytes, want the 512-byte bucket", f.size) } - if d.Fields[fidKey(FidGridW)] != 4321 { - t.Fatalf("the reading that changed did not survive truncation: %v", d.Fields) - } -} - -// A field dropped to fit the bucket was never sent, so it is still different -// on the next tick and goes then. -func TestTruncatedSourcesAreRetriedOnTheNextTick(t *testing.T) { - h, box, rec, clock := newRig(t) - box.snap.Sources = nil - for i := 0; i < 24; i++ { - box.snap.Sources = append(box.snap.Sources, Source{ - ID: "driver.long.identifier." + string(rune('a'+i)), - Kind: "inverter", - Name: "A device with a comfortably long display name", - LastOkUptimeMs: 60_000, - StaleAfterMs: 15_000, - }) - } - subscribe(t, h, rec) - - clock.uptimeMs += 60_000 - if err := h.Tick(); err != nil { - t.Fatal(err) - } - rec.reset() - - if err := h.Tick(); err != nil { - t.Fatal(err) - } - f := rec.last(t, MsgDelta) - if f.flags&FlagTrunc == 0 { - t.Fatal("the retry did not report itself truncated either") + snap := body[Snap](t, rec.only(t, MsgSnap)) + if snap.Fields[fidKey(FidGridW)] != 4321 || len(snap.Sources) != 24 { + t.Fatalf("the snapshot carries grid_w %d and %d sources", snap.Fields[fidKey(FidGridW)], len(snap.Sources)) } } diff --git a/go/internal/appuplink/agents_test.go b/go/internal/appuplink/agents_test.go index eb1a292f3..fa39474ac 100644 --- a/go/internal/appuplink/agents_test.go +++ b/go/internal/appuplink/agents_test.go @@ -38,7 +38,7 @@ func agentTunRig(t *testing.T, now *time.Time) (*tunRig, *AgentAccess) { SrcGrid: "meter", SrcPV: "meter", SrcBattery: "meter", Logger: slog.New(slog.DiscardHandler), }) } - r.tunnel = newTunnel(build, enroll, r.slots, slog.New(slog.DiscardHandler)) + r.tunnel = newTunnel(build, enroll, r.slots, nil, slog.New(slog.DiscardHandler)) r.tunnel.Attach(r.out) ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) diff --git a/go/internal/appuplink/arrival_test.go b/go/internal/appuplink/arrival_test.go new file mode 100644 index 000000000..f4f2a5257 --- /dev/null +++ b/go/internal/appuplink/arrival_test.go @@ -0,0 +1,489 @@ +package appuplink + +import ( + "context" + "log/slog" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw/go/internal/apiauth" + "github.com/srcfl/ftw/go/internal/appenroll" + "github.com/srcfl/ftw/go/internal/appproto" + "github.com/srcfl/ftw/go/internal/cloudlink" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// movingSite is stubSite with a grid reading the test moves. +type movingSite struct{ gridW atomic.Int64 } + +func (m *movingSite) Snapshot() appproto.Snapshot { + s := stubSite{}.Snapshot() + s.GridW = float64(m.gridW.Load()) + return s +} + +// arrivalRig is a tunnel table that hears arrivals, with one phone +// subscribed through it. +type arrivalRig struct { + *tunRig + site *movingSite + arrivals chan struct{} + app *appClient +} + +func newArrivalRig(t *testing.T) *arrivalRig { + t.Helper() + enroll, err := appenroll.LoadOrCreate(filepath.Join(t.TempDir(), "nova.key")) + if err != nil { + t.Fatal(err) + } + r := &arrivalRig{site: &movingSite{}, arrivals: make(chan struct{}, 1), + tunRig: &tunRig{enroll: enroll, slots: &slots{}, out: &capturedTun{ch: make(chan cloudwire.Tun, 64)}, modes: &countingModes{}}} + r.site.gridW.Store(1200) + cmds := appproto.NewCmdLog() + build := func(s appproto.Sender, caller apiauth.Caller, grants appproto.GrantReader) (*appproto.Handler, error) { + return appproto.New(appproto.Config{ + Clock: appproto.SystemClock{StartedAt: time.Now(), Source: "ntp"}, + Site: r.site, Info: stubInfo{}, Modes: r.modes, Plans: stubPlans{}, Codec: codec{}, + Sender: s, Caller: caller, Grants: grants, Commands: cmds, + SrcGrid: "meter", SrcPV: "meter", SrcBattery: "meter", Logger: slog.New(slog.DiscardHandler), + }) + } + r.tunnel = newTunnel(build, enroll, r.slots, r.arrivals, slog.New(slog.DiscardHandler)) + r.tunnel.Attach(r.out) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go r.tunnel.Run(ctx) + + r.app = r.pair(t) + dev := deviceIDOf(r.app.staticPublic()) + r.open(t, r.app, dev, 7, nil) + r.hello(t, r.app, dev, 7) + r.send(t, r.app, dev, 7, appproto.MsgSub, appproto.Sub{Bucket: 512, Hz: 1}) + r.receive(t, r.app, appproto.MsgSnap) + return r +} + +// laneZero is the next lane 0 reading the phone gets, when it got it, and +// its padded size. +func (r *arrivalRig) laneZero(t *testing.T, within time.Duration) (appwire.Frame, time.Time, int) { + t.Helper() + deadline := time.Now().Add(within) + for { + m := r.next(t, time.Until(deadline)) + at := time.Now() + plain, err := r.app.decrypt(m.D) + if err != nil { + t.Fatal(err) + } + f, err := appwire.DecodeFrame(plain) + if err != nil { + t.Fatal(err) + } + if f.Envelope.T == appproto.MsgDelta || f.Envelope.T == appproto.MsgTick { + return f, at, len(plain) + } + } +} + +// Through FTW Cloud a reading leaves the box as it lands, not on the next +// pulse of a one-second clock, and still in its fixed-size bucket. Five +// readings at different phases of the second all go within 200 ms; with +// the pulse, all five doing so would be a one-in-3,125 fluke (0.2 to the +// fifth). +func TestAReadingLeavesThroughTheTunnelAsItArrives(t *testing.T) { + r := newArrivalRig(t) + for i := range 5 { + time.Sleep(time.Duration(130+97*i) * time.Millisecond) + want := int64(2000 + i) + r.site.gridW.Store(want) + at := time.Now() + r.arrivals <- struct{}{} + for { + f, got, size := r.laneZero(t, 2*time.Second) + if size != 512 { + t.Fatalf("a %s left in %d bytes, want the 512-byte bucket", f.Envelope.T, size) + } + var d appproto.Delta + if f.Envelope.T != appproto.MsgDelta || appproto.Unmarshal(f.Envelope.B, &d) != nil || + d.Fields["2"] != want { + continue // a beat that crossed the arrival + } + if took := got.Sub(at); took > 200*time.Millisecond { + t.Fatalf("reading %d took %s to leave the box", i, took) + } + break + } + } +} + +// With nothing arriving, the session still hears from the box once a +// second, so the phone never reads a quiet house as a lost carrier. +func TestAQuietTunnelSessionStillBeats(t *testing.T) { + r := newArrivalRig(t) + _, first, _ := r.laneZero(t, 2*time.Second) + f, second, size := r.laneZero(t, 2*time.Second) + if f.Envelope.T != appproto.MsgTick || size != 512 { + t.Fatalf("a quiet beat sent a %s in %d bytes, want a 512-byte tick", f.Envelope.T, size) + } + if gap := second.Sub(first); gap < 900*time.Millisecond || gap > 1500*time.Millisecond { + t.Fatalf("beats %s apart, want about a second", gap) + } +} + +// Drivers that each read once a second never wait for the pace; only a +// stream faster than five frames a second does. +func TestThePaceTakesThreeAtOnceAndFiveASecond(t *testing.T) { + t0 := time.Unix(1_760_000_000, 0) + p := pace{tokens: tunBurst, at: t0} + for i := range 3 { + if w := p.wait(t0); w != 0 { + t.Fatalf("frame %d of a burst waits %s", i, w) + } + p.take(t0) + } + if w := p.wait(t0); w != 200*time.Millisecond { + t.Fatalf("a fourth frame at once waits %s, want 200ms", w) + } + + // A poll every 100 ms for ten seconds gets five frames a second. + sent := 0 + next := t0 + for at := t0; at.Before(t0.Add(10 * time.Second)); at = at.Add(100 * time.Millisecond) { + if at.Before(next) { + continue + } + if w := p.wait(at); w > 0 { + next = at.Add(w) + continue + } + p.take(at) + sent++ + } + if sent < 49 || sent > 52 { + t.Fatalf("%d frames in ten seconds of polls every 100 ms, want about 50", sent) + } +} + +// gather collects for d what the box sends the phone app under dev, lane 0 +// frame types only, in order. +func (r *arrivalRig) gather(t *testing.T, app *appClient, dev string, d time.Duration) []string { + t.Helper() + var got []string + deadline := time.After(d) + for { + select { + case m := <-r.out.ch: + if m.Dev != dev || m.Ended() { + continue + } + plain, err := app.decrypt(m.D) + if err != nil { + t.Fatal(err) + } + f, err := appwire.DecodeFrame(plain) + if err != nil { + t.Fatal(err) + } + if f.Envelope.T == appproto.MsgDelta || f.Envelope.T == appproto.MsgTick { + got = append(got, f.Envelope.T) + } + case <-deadline: + return got + } + } +} + +// arrive signals a poll every gap for d. +func (r *arrivalRig) arrive(gap, d time.Duration) { + for end := time.Now().Add(d); time.Now().Before(end); time.Sleep(gap) { + select { + case r.arrivals <- struct{}{}: + default: + } + } +} + +// A driver reading fifty times a second costs the phone five frames a +// second and the three of a burst, no more. +func TestFastPollsArePacedToFiveFramesASecond(t *testing.T) { + r := newArrivalRig(t) + dev := deviceIDOf(r.app.staticPublic()) + r.gather(t, r.app, dev, 300*time.Millisecond) + go r.arrive(20*time.Millisecond, 2*time.Second) + if n := len(r.gather(t, r.app, dev, 2*time.Second)); n < 10 || n > 16 { + t.Fatalf("%d frames in two seconds of polls every 20 ms, want about 13", n) + } +} + +// Every phone that streams hears each poll. +func TestEveryLiveSessionHearsAPoll(t *testing.T) { + r := newArrivalRig(t) + other := r.pair(t) + dev2 := deviceIDOf(other.staticPublic()) + r.open(t, other, dev2, 8, nil) + r.send(t, other, dev2, 8, appproto.MsgHello, appproto.Hello{ + Proto: appproto.ProtoRange{Min: 0, Max: appproto.ProtoMax}, + App: appproto.AppInfo{Build: "test", UA: "ios"}, + }) + r.send(t, other, dev2, 8, appproto.MsgSub, appproto.Sub{Bucket: 512, Hz: 1}) + time.Sleep(300 * time.Millisecond) + + r.site.gridW.Store(3333) + r.arrivals <- struct{}{} + seen := map[string]bool{} + deadline := time.After(2 * time.Second) + for len(seen) < 2 { + select { + case m := <-r.out.ch: + app := r.app + if m.Dev == dev2 { + app = other + } + plain, err := app.decrypt(m.D) + if err != nil { + t.Fatal(err) + } + f, _ := appwire.DecodeFrame(plain) + var d appproto.Delta + if f.Envelope.T == appproto.MsgDelta && appproto.Unmarshal(f.Envelope.B, &d) == nil && d.Fields["2"] == 3333 { + seen[m.Dev] = true + } + case <-deadline: + t.Fatalf("the poll reached %d of 2 phones", len(seen)) + } + } +} + +// dropLaneZero loses lane 0 tun while drop is set, as a full queue would, +// and says how much of the tunnel's budget is spent. +type dropLaneZero struct { + inner *capturedTun + drop atomic.Bool + used atomic.Int64 // thousandths +} + +func (d *dropLaneZero) SendTun(m cloudwire.Tun, bulk bool) error { + if d.drop.Load() && !bulk { + return countedDrop{} + } + return d.inner.SendTun(m, bulk) +} + +func (d *dropLaneZero) TunBudgetUsed() float64 { return float64(d.used.Load()) / 1000 } + +// A reading the link dropped is not one the phone has: the next frame +// carries it, and the session goes on. +func TestAReadingTheLinkDroppedGoesWithTheNextFrame(t *testing.T) { + r := newArrivalRig(t) + dev := deviceIDOf(r.app.staticPublic()) + link := &dropLaneZero{inner: r.out} + r.tunnel.Attach(link) + r.gather(t, r.app, dev, 300*time.Millisecond) + + link.drop.Store(true) + r.site.gridW.Store(4444) + r.arrivals <- struct{}{} + time.Sleep(300 * time.Millisecond) + link.drop.Store(false) + r.arrivals <- struct{}{} + for { + f, _, _ := r.laneZero(t, 2*time.Second) + var d appproto.Delta + if f.Envelope.T == appproto.MsgDelta && appproto.Unmarshal(f.Envelope.B, &d) == nil { + if d.Fields["2"] != 4444 { + t.Fatalf("the next delta carries %v, want grid_w 4444 again", d.Fields) + } + return + } + } +} + +// Past 60 % of the tunnel's daily budget a session gets the beat alone; +// past 90 % no readings at all, so handshakes, answers and ends still go. +func TestTheBudgetStopsPushesThenBeats(t *testing.T) { + r := newArrivalRig(t) + dev := deviceIDOf(r.app.staticPublic()) + link := &dropLaneZero{inner: r.out} + r.tunnel.Attach(link) + r.gather(t, r.app, dev, 300*time.Millisecond) + + link.used.Store(700) + go r.arrive(50*time.Millisecond, 2500*time.Millisecond) + if n := len(r.gather(t, r.app, dev, 2500*time.Millisecond)); n < 1 || n > 3 { + t.Fatalf("at 70 %% of the budget %d frames in 2.5 s of polls, want the beat's 2", n) + } + + link.used.Store(950) + go r.arrive(50*time.Millisecond, 2500*time.Millisecond) + if got := r.gather(t, r.app, dev, 2500*time.Millisecond); len(got) != 0 { + t.Fatalf("at 95 %% of the budget the box still sent %v", got) + } +} + +// Hidden, a poll sends nothing; back in front, the stream picks up within +// a beat and each poll goes again. +func TestHiddenThenVisible(t *testing.T) { + r := newArrivalRig(t) + dev := deviceIDOf(r.app.staticPublic()) + r.send(t, r.app, dev, 7, appproto.MsgSub, appproto.Sub{Bucket: 512, Hz: 0.2}) + r.gather(t, r.app, dev, 300*time.Millisecond) + + go r.arrive(100*time.Millisecond, time.Second) + if got := r.gather(t, r.app, dev, time.Second); len(got) > 1 { + t.Fatalf("hidden, a second of polls sent %v", got) + } + + r.send(t, r.app, dev, 7, appproto.MsgSub, appproto.Sub{Bucket: 512, Hz: 1}) + r.laneZero(t, 1500*time.Millisecond) + r.site.gridW.Store(5555) + at := time.Now() + r.arrivals <- struct{}{} + for { + f, got, _ := r.laneZero(t, 2*time.Second) + var d appproto.Delta + if f.Envelope.T == appproto.MsgDelta && appproto.Unmarshal(f.Envelope.B, &d) == nil && d.Fields["2"] == 5555 { + if took := got.Sub(at); took > 200*time.Millisecond { + t.Fatalf("back in front, a poll took %s", took) + } + return + } + } +} + +// The relay keeps its pulse: polls arriving fast change nothing about when +// a relay session's frames go. +func TestARelaySessionKeepsItsPulseWhileArrivalsCome(t *testing.T) { + arrivals := make(chan struct{}, 1) + r := newRigOpts(t, 481234, func(o *Options) { o.Handler = newHandler; o.Arrivals = arrivals }) + app, err := newAppClient() + if err != nil { + t.Fatal(err) + } + code, _, err := r.enroll.MintPairingCode(apiauth.RoleOwner, appenroll.PairingTTL) + if err != nil { + t.Fatal(err) + } + conn := openSession(t, r, app, code) + send(t, conn, app, appproto.MsgSub, 512, appproto.Sub{Bucket: 512, Hz: 1}) + if f := receive(t, conn, app); f.Envelope.T != appproto.MsgSnap { + t.Fatalf("got %s, want the snapshot", f.Envelope.T) + } + + stop := make(chan struct{}) + defer close(stop) + go func() { + for { + select { + case <-stop: + return + case arrivals <- struct{}{}: + default: + } + time.Sleep(20 * time.Millisecond) + } + }() + frames := 0 + for start := time.Now(); time.Since(start) < 3*time.Second; { + _ = conn.SetReadDeadline(time.Now().Add(1500 * time.Millisecond)) + kind, _, err := conn.ReadMessage() + if err != nil { + break + } + if kind == websocket.BinaryMessage { + frames++ + } + } + if frames < 2 || frames > 4 { + t.Fatalf("a relay session sent %d frames in 3 s of fast polls, want its pulse's 3", frames) + } +} + +// A bulk message the link paces must not fall out of the phone's Noise +// replay window behind the lane 0 frames that overtake it. +func TestLaneZeroCannotPushABulkFrameOutOfTheReplayWindow(t *testing.T) { + wait := cloudlink.MaxBulkWait() + // Commands and their answers ride lane 0 too; a person sends a few. + const commands = 8 + if n := tunBurst + tunRate*wait.Seconds() + commands; n >= appwire.ReplayWindow { + t.Fatalf("%.0f lane 0 frames can pass a bulk frame held %s; the window is %d", n, wait, appwire.ReplayWindow) + } +} + +// What one push costs the box for one session: the site read, the delta, +// its padding, Noise and the tun frame, as Run does it. Multiply by frames a +// second and sessions for the share of a core. +func BenchmarkAPushThroughTheTunnel(b *testing.B) { + enroll, err := appenroll.LoadOrCreate(filepath.Join(b.TempDir(), "nova.key")) + if err != nil { + b.Fatal(err) + } + app, _ := newAppClient() + m1, _ := app.message1(enroll.StaticKey().Public(), nil) + hs, _ := appwire.NewResponder(appwire.HandshakeOptions{StaticKey: enroll.StaticKey(), Prologue: prologue(enroll.StaticKey().Public())}) + if _, err := hs.ReadMessage(m1); err != nil { + b.Fatal(err) + } + reply, _ := hs.WriteMessage(nil) + noise, _ := hs.Split() + if err := app.readMessage2(reply); err != nil { + b.Fatal(err) + } + site := &movingSite{} + var out []byte + send := func(msg []byte, _ bool) error { + body, err := cloudwire.Marshal(cloudwire.Tun{Dev: "AbCdEfGh", Sid: 7, D: msg}) + if err != nil { + return err + } + raw, err := cloudwire.Envelope{T: cloudwire.MsgTun, B: body}.Encode() + if err != nil { + return err + } + out, err = cloudwire.EncodeFrame(raw) + return err + } + h, err := appproto.New(appproto.Config{ + Clock: appproto.SystemClock{StartedAt: time.Now(), Source: "ntp"}, + Site: site, Info: stubInfo{}, Modes: stubModes{}, Plans: stubPlans{}, Codec: codec{}, + Sender: tunSender{transport: appwire.NewTransport(noise), send: send}, + Caller: apiauth.Caller{Subject: "app:AbCdEfGh", Kind: apiauth.KindApp, Role: apiauth.RoleOwner, + Scopes: appproto.ScopesForRole(apiauth.RoleOwner), Epoch: 1}, + Grants: ownerGrant{}, Commands: appproto.NewCmdLog(), + SrcGrid: "meter", SrcPV: "meter", SrcBattery: "meter", Logger: slog.New(slog.DiscardHandler), + }) + if err != nil { + b.Fatal(err) + } + for _, m := range []struct { + t string + body any + }{ + {appproto.MsgHello, appproto.Hello{Proto: appproto.ProtoRange{Min: 0, Max: appproto.ProtoMax}}}, + {appproto.MsgSub, appproto.Sub{Bucket: 512, Hz: 1}}, + } { + raw, _ := appproto.Marshal(m.body) + frame, _ := appwire.EncodeFrame(appwire.Frame{Lane: appwire.LaneControl, Bucket: 512, Envelope: appwire.Envelope{T: m.t, B: raw}}) + if err := h.Handle(context.Background(), frame); err != nil { + b.Fatal(err) + } + } + for b.Loop() { + site.gridW.Add(1) + if err := h.Push(); err != nil { + b.Fatal(err) + } + } + b.ReportMetric(float64(len(out)), "B/frame") +} + +// ownerGrant is a paired owner phone whose grant never moves. +type ownerGrant struct{} + +func (ownerGrant) Grant() (appproto.Enrolment, bool) { + return appproto.Enrolment{Role: apiauth.RoleOwner, Epoch: 1}, true +} diff --git a/go/internal/appuplink/client.go b/go/internal/appuplink/client.go index 771899302..225dca02d 100644 --- a/go/internal/appuplink/client.go +++ b/go/internal/appuplink/client.go @@ -110,7 +110,8 @@ const ( // refused there anyway; refusing it here keeps the memory bound local. maxFrameBytes = 65536 - // tickInterval is lane 0's cadence, fixed for the life of a session. + // tickInterval is lane 0's cadence over the relay, fixed for the life + // of a session. tickInterval = time.Second ) @@ -135,6 +136,11 @@ type Options struct { Logger *slog.Logger + // Arrivals hears each finished driver poll (telemetry.Store.Arrivals). + // Sessions through FTW Cloud send on it; nil leaves them on the beat + // alone, one frame per subscribed period. Relay sessions never use it. + Arrivals <-chan struct{} + // Deadman, when set, arms the dead man's switch: rows posted to the // relay's HTTPS side, claims spoken on every connection. Nil leaves // the uplink exactly as it was — no rows, no claims, no text frames. @@ -288,7 +294,7 @@ func New(opts Options) (*Uplink, error) { u.deadmanRefresh = deadmanRefreshInterval } u.slots = &slots{} - u.tunnel = newTunnel(opts.Handler, opts.Enroll, u.slots, u.log) + u.tunnel = newTunnel(opts.Handler, opts.Enroll, u.slots, opts.Arrivals, u.log) if opts.ExchangeHandler != nil { u.exchanges = &exchanges{ build: opts.ExchangeHandler, diff --git a/go/internal/appuplink/livebox_test.go b/go/internal/appuplink/livebox_test.go new file mode 100644 index 000000000..483b0b585 --- /dev/null +++ b/go/internal/appuplink/livebox_test.go @@ -0,0 +1,222 @@ +package appuplink + +import ( + "bufio" + "context" + "crypto/ecdsa" + "crypto/x509" + "encoding/base64" + "fmt" + "log/slog" + "math/rand/v2" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/apiauth" + "github.com/srcfl/ftw/go/internal/appenroll" + "github.com/srcfl/ftw/go/internal/appproto" + "github.com/srcfl/ftw/go/internal/cloudlink" + "github.com/srcfl/ftw/go/internal/control" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// TestLiveBoxForTheiPhone is a box for the iPhone's live tests in +// srcfl/ftw-app (LiveTunnelTests, LiveLatencyTests): the real uplink, cloud +// link, tunnel table and handler, linked to a local ftwcloudd, with the +// relay unreachable. Off unless FTW_LIVE_BOX names a directory. +// +// FTW_LIVE_BOX= holds the service's ca.pem and service.pub; gets +// pair.url, arrivals.log and tun.log +// FTW_LIVE_BOX_URL the service's /v1/box, wss://127.0.0.1:18843/v1/box +// FTW_LIVE_BOX_DRIVERS how many pretend drivers read, 1 by default +// FTW_LIVE_BOX_PERIOD_MS how often each reads, 1000 by default +// +// Each pretend driver reads at a random 0.8 to 1.2 times its period, so its +// readings land at every phase of any clock the box keeps. The grid reading is the +// wall-clock millisecond the newest arrived, so the phone can time it. arrivals.log has each +// arrival and tun.log each tun the box sent, in wall-clock microseconds. +// The box serves until /stop appears. +func TestLiveBoxForTheiPhone(t *testing.T) { + dir := os.Getenv("FTW_LIVE_BOX") + if dir == "" { + t.Skip("set FTW_LIVE_BOX") + } + read := func(name string) string { + b, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatal(err) + } + return strings.TrimSpace(string(b)) + } + svcRaw, _ := base64.RawURLEncoding.DecodeString(read("service.pub")) + svcPub, err := cloudwire.ParsePublicKey(svcRaw) + if err != nil { + t.Fatal(err) + } + pool := x509.NewCertPool() + pool.AppendCertsFromPEM([]byte(read("ca.pem"))) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})) + logs := func(name string) (func(string, ...any), func()) { + f, err := os.Create(filepath.Join(dir, name)) + if err != nil { + t.Fatal(err) + } + w := bufio.NewWriter(f) + var mu sync.Mutex + return func(format string, a ...any) { + mu.Lock() + fmt.Fprintf(w, format, a...) + mu.Unlock() + }, func() { + mu.Lock() + _ = w.Flush() + _ = f.Close() + mu.Unlock() + } + } + + enroll, err := appenroll.LoadOrCreate(filepath.Join(t.TempDir(), "nova.key")) + if err != nil { + t.Fatal(err) + } + start := time.Now() + site := &clockSite{} + site.ms.Store(start.UnixMilli()) + modes := &countingModes{} + cmds := appproto.NewCmdLog() + var bound atomic.Pointer[cloudlink.Link] + caps := append(appproto.DefaultCaps(), appproto.CapCloudLink, appproto.CapCloudTun) + build := func(s appproto.Sender, caller apiauth.Caller, grants appproto.GrantReader) (*appproto.Handler, error) { + return appproto.New(appproto.Config{ + Clock: appproto.SystemClock{StartedAt: start, Source: "ntp"}, + Site: site, Info: stubInfo{}, Modes: modes, Plans: stubPlans{}, Codec: codec{}, + Sender: s, Caller: caller, Grants: grants, Commands: cmds, Caps: caps, + Cloud: func() *appproto.CloudSession { + if l := bound.Load(); l != nil { + return l.SessionInfo() + } + return nil + }, + SrcGrid: "meter", SrcPV: "meter", SrcBattery: "meter", Logger: logger, + }) + } + arrivals := make(chan struct{}, 1) + uplink, err := New(Options{Endpoint: unreachable(t), Enroll: enroll, Handler: build, Arrivals: arrivals, Logger: logger}) + if err != nil { + t.Fatal(err) + } + link, err := cloudlink.New(cloudlink.Options{ + Site: quietSite{}, Roster: CloudRoster{Enroll: enroll, Tunnel: true}, KV: &memState{m: map[string]string{}}, + KeyPath: filepath.Join(t.TempDir(), "cloud.key"), Build: "live", Tunnel: uplink.Tunnel(), + Endpoint: os.Getenv("FTW_LIVE_BOX_URL"), ServiceKeys: []*ecdsa.PublicKey{svcPub}, RootCAs: pool, + Logger: logger, + }) + if err != nil { + t.Fatal(err) + } + tunLog, closeTun := logs("tun.log") + defer closeTun() + uplink.Tunnel().Attach(timedTun{inner: link, log: tunLog}) + bound.Store(link) + enroll.OnChange(link.RosterChanged) + enroll.OnPairingCode(link.RosterChanged) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = uplink.Run(ctx) }() + go func() { _ = link.Run(ctx) }() + + arrivalLog, closeArrivals := logs("arrivals.log") + defer closeArrivals() + drivers, _ := strconv.Atoi(os.Getenv("FTW_LIVE_BOX_DRIVERS")) + period, _ := strconv.Atoi(os.Getenv("FTW_LIVE_BOX_PERIOD_MS")) + if period <= 0 { + period = 1000 + } + var arriving sync.Mutex + for range max(drivers, 1) { + go func() { + for ctx.Err() == nil { + time.Sleep(time.Duration(period*4/5+rand.IntN(period*2/5+1)) * time.Millisecond) + arriving.Lock() + now := time.Now() + site.ms.Store(now.UnixMilli()) + site.okUpMs.Store(now.Sub(start).Milliseconds()) + arrivalLog("%d %d\n", now.UnixMicro(), now.UnixMilli()) + arriving.Unlock() + select { + case arrivals <- struct{}{}: + default: + } + } + }() + } + + code, _, err := enroll.MintPairingCode(apiauth.RoleOwner, appenroll.PairingTTL) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(15 * time.Second) + for !link.Status().Connected || link.Status().Home == "" { + if time.Now().After(deadline) { + t.Fatalf("the box never linked: %+v", link.Status()) + } + time.Sleep(20 * time.Millisecond) + } + link.RosterChanged() + time.Sleep(300 * time.Millisecond) + v2, err := enroll.EnrollmentURL(code, "") + if err != nil { + t.Fatal(err) + } + // v3 by hand while appenroll.QRNamesHome stays off. + v3 := strings.Replace(v2, "#v2.", "#v3.", 1) + "." + base64.RawURLEncoding.EncodeToString([]byte(link.Status().Home)) + if err := os.WriteFile(filepath.Join(dir, "pair.url"), []byte(v3), 0o600); err != nil { + t.Fatal(err) + } + t.Logf("box linked as %s; pairing link written", link.Status().Home) + + stop := filepath.Join(dir, "stop") + _ = os.Remove(stop) + for end := time.Now().Add(30 * time.Minute); time.Now().Before(end); time.Sleep(200 * time.Millisecond) { + if _, err := os.Stat(stop); err == nil { + break + } + } + t.Logf("done: mode set %d times", modes.count()) +} + +// clockSite is a box whose grid reading is the wall-clock millisecond its +// pretend driver last read, and whose meter last answered then. +type clockSite struct { + ms atomic.Int64 + okUpMs atomic.Int64 +} + +func (c *clockSite) Snapshot() appproto.Snapshot { + return appproto.Snapshot{ + Mode: control.ModeSelfConsumption, ControlRev: 1, + GridW: float64(c.ms.Load()), PVW: -3000, BatteryW: 800, LoadW: 1000, + BatterySoC: 0.55, BatterySoCKnown: true, + Sources: []appproto.Source{{ID: "meter", Kind: "driver", Name: "meter", + LastOkUptimeMs: c.okUpMs.Load(), StaleAfterMs: 5000}}, + } +} + +// timedTun logs each tun the box sends: when, how long and which lane. +type timedTun struct { + inner TunSender + log func(string, ...any) +} + +func (t timedTun) SendTun(m cloudwire.Tun, bulk bool) error { + at := time.Now().UnixMicro() + err := t.inner.SendTun(m, bulk) + t.log("%d %d %v\n", at, len(m.D), bulk) + return err +} diff --git a/go/internal/appuplink/tunnel.go b/go/internal/appuplink/tunnel.go index aecba5fdf..f149a0f59 100644 --- a/go/internal/appuplink/tunnel.go +++ b/go/internal/appuplink/tunnel.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "log/slog" "sync" "time" @@ -31,7 +32,12 @@ import ( // box answers it, but the replay never proves itself, so it takes no // device's place, stamps nothing and spends no code. Each new session runs // its handshake on a goroutine of its own, so a slow disk under the -// enrolment write never holds up routing or the one-second tick. +// enrolment write never holds up routing or the stream. +// +// Lane 0 goes on each finished driver poll, not on the relay's one-second +// pulse, and a beat goes when no frame has gone for the subscribed period. +// Frames keep their fixed size. What the service learns from their timing +// is written in docs/protocol-cloud.md, "Lane 0 through the tunnel". // TunSender sends one tun up the box's link to FTW Cloud. cloudlink.Link is // one. bulk says the message carries a lane 1 frame, which the link paces; @@ -65,6 +71,28 @@ const ( // tunPendingPerDevice is how many unproved sessions a device may hold, // and pendingWait (session.go) how long one waits to prove itself. tunPendingPerDevice = 4 + + // tunGather is how long an arrival waits for polls that finish with it, + // so drivers that read together go in one frame. One driver's poll is + // one arrival already. + tunGather = 5 * time.Millisecond + // tunRate and tunBurst bound the frames arrivals push: five a second on + // average, three at once. A few drivers that each read once a second + // seldom wait; one that reads ten times a second costs each phone five + // frames a second, and four phones stay well under the service's 60 tun + // a second per home. + tunRate = 5 + tunBurst = 3 + // tunPushUntil and tunBeatUntil are the shares of the tunnel's 24-hour + // budget past which readings stop: pushes at 60 %, so a display left on + // all day falls back to a frame a second, and beats at 90 %, so the + // last tenth is left for handshakes, command answers and ends. + tunPushUntil = 0.6 + tunBeatUntil = 0.9 + // tunBeat is how often each session is asked whether it has been quiet + // for its subscribed period. Silence on the phone is at most the period + // plus this. + tunBeat = 250 * time.Millisecond ) // Tunnel is the box's table of sessions that FTW Cloud carries: one live per @@ -79,6 +107,11 @@ type Tunnel struct { now func() time.Time inbox chan tunIn + // arrivals hears each finished driver poll; nil sends on Beat alone. + arrivals <-chan struct{} + // kick tells Run a session went live or ended, so it beats and hears + // arrivals only while one is live. + kick chan struct{} mu sync.Mutex sender TunSender @@ -118,11 +151,11 @@ type tunSession struct { // to deliver to. Its sessions take the same slots as the relay's. func (u *Uplink) Tunnel() *Tunnel { return u.tunnel } -func newTunnel(build HandlerBuilder, enroll *appenroll.Identity, slots *slots, log *slog.Logger) *Tunnel { +func newTunnel(build HandlerBuilder, enroll *appenroll.Identity, slots *slots, arrivals <-chan struct{}, log *slog.Logger) *Tunnel { return &Tunnel{ build: build, enroll: enroll, static: enroll.StaticKey(), slots: slots, log: log, now: time.Now, - inbox: make(chan tunIn, tunInbox), live: map[string]*tunSession{}, pending: map[tunKey]*tunSession{}, - opens: map[string][]time.Time{}, + inbox: make(chan tunIn, tunInbox), arrivals: arrivals, kick: make(chan struct{}, 1), + live: map[string]*tunSession{}, pending: map[tunKey]*tunSession{}, opens: map[string][]time.Time{}, } } @@ -165,23 +198,91 @@ func (t *Tunnel) Drop() { } // Run serves the table until ctx ends: the messages the service delivers, -// and the one-second pulse lane 0 needs. +// each arrival's frame, and the beat that keeps a quiet session heard. func (t *Tunnel) Run(ctx context.Context) { - tick := time.NewTicker(tickInterval) - defer tick.Stop() defer t.Drop() + // The beat runs, and arrivals are heard, only while a session is live: + // a box nobody watches wakes for nothing. + var beat *time.Ticker + defer func() { + if beat != nil { + beat.Stop() + } + }() + // flush is armed by an arrival and fires once its round is in and the + // pace allows a frame. Nil while nothing waits. + var flush <-chan time.Time + p := pace{tokens: tunBurst, at: time.Now()} for { + var beats <-chan time.Time + var arrivals <-chan struct{} + switch live := t.count() > 0; { + case live && beat == nil: + beat = time.NewTicker(tunBeat) + case !live && beat != nil: + beat.Stop() + beat = nil + } + if beat != nil { + beats, arrivals = beat.C, t.arrivals + } select { case <-ctx.Done(): return case m := <-t.inbox: t.route(ctx, m) - case <-tick.C: - t.tick() + case <-t.kick: + case <-arrivals: + if flush == nil { + flush = time.After(max(tunGather, p.wait(time.Now()))) + } + case <-flush: + flush = nil + // A frame spends the pace; a poll nobody sees spends nothing. + if t.budgetUsed() < tunPushUntil && t.anyVisible() { + p.take(time.Now()) + t.each("push", (*appproto.Handler).Push) + } + case <-beats: + if t.budgetUsed() < tunBeatUntil { + t.each("beat", (*appproto.Handler).Beat) + } } } } +// anyVisible says a live session would take a push. +func (t *Tunnel) anyVisible() bool { + t.mu.Lock() + defer t.mu.Unlock() + for _, s := range t.live { + if s.session.handler.Visible() { + return true + } + } + return false +} + +// budgetUsed is the share of the link's tunnel budget spent in the last 24 +// hours, or 0 from a link that does not say. +func (t *Tunnel) budgetUsed() float64 { + t.mu.Lock() + s := t.sender + t.mu.Unlock() + if b, ok := s.(interface{ TunBudgetUsed() float64 }); ok { + return b.TunBudgetUsed() + } + return 0 +} + +// kicked tells Run the live sessions changed. It never blocks. +func (t *Tunnel) kicked() { + select { + case t.kick <- struct{}{}: + default: + } +} + // count is how many live sessions the table holds. func (t *Tunnel) count() int { t.mu.Lock() @@ -381,7 +482,7 @@ func (t *Tunnel) promote(s *tunSession, transport *appwire.Transport, commit fun } caller := callerOf(grant) send := func(msg []byte, bulk bool) error { - return t.sendTun(cloudwire.Tun{Dev: s.dev, Sid: s.sid, D: msg}, bulk) + return t.sendFrame(cloudwire.Tun{Dev: s.dev, Sid: s.sid, D: msg}, bulk) } handler, err := t.build(tunSender{transport: transport, send: send}, caller, liveGrant{enroll: t.enroll, deviceID: grant.DeviceID}) @@ -409,13 +510,14 @@ func (t *Tunnel) promote(s *tunSession, transport *appwire.Transport, commit fun // The device's one slot, whichever carrier held it before. Claiming it // ends the older session, through the relay or through here. sess.setRelease(t.slots.claim(grant.DeviceID, func() { t.remove(s, true) })) + t.kicked() t.log.Info("app session open through FTW Cloud") return nil } // tunSender encrypts one session's frames for the tunnel and says which are // bulk, so the link paces those and sends lane 0 (commands, their answers -// and the ticks) first. A message the link dropped and counted is gone like +// and the readings) first. A message the link dropped and counted is gone like // any lost frame, and the session goes on; only a lost uplink fails it. type tunSender struct { transport *appwire.Transport @@ -430,6 +532,16 @@ func (s tunSender) Send(frame []byte) error { return s.send(msg, len(frame) > 1 && frame[1] == appwire.LaneBulk) } +// SendInLine sends a bulk frame with lane 0, in order: the snapshot, which a +// reading sent after it must not overtake. +func (s tunSender) SendInLine(frame []byte) error { + msg, err := s.transport.Encrypt(frame) + if err != nil { + return err + } + return s.send(msg, false) +} + // end closes the session once and frees its slot. It reports whether this // call ended it. func (s *tunSession) end() bool { @@ -450,6 +562,7 @@ func (t *Tunnel) remove(s *tunSession, tell bool) { t.mu.Lock() if t.live[s.dev] == s { delete(t.live, s.dev) + defer t.kicked() } if t.pending[tunKey{s.dev, s.sid}] == s { delete(t.pending, tunKey{s.dev, s.sid}) @@ -465,9 +578,26 @@ func (t *Tunnel) remove(s *tunSession, tell bool) { // an end. False means it did not leave. func (t *Tunnel) send(m cloudwire.Tun) bool { return t.sendTun(m, false) == nil } -// sendTun hands a session's message to the link. A counted drop is not an -// error: Noise transport survives it, and failing the session would lose a -// command id it holds. A lost uplink is. +// sendFrame hands a session's frame to the link. A frame the link dropped +// and counted comes back as appproto.ErrDropped: the session goes on, and +// its handler sends what the frame carried again. +func (t *Tunnel) sendFrame(m cloudwire.Tun, bulk bool) error { + t.mu.Lock() + s := t.sender + t.mu.Unlock() + if s == nil { + return errTunGone + } + err := s.SendTun(m, bulk) + if dropped(err) { + return fmt.Errorf("%w: %w", appproto.ErrDropped, err) + } + return err +} + +// sendTun hands the table's own message to the link: a handshake's answer +// or an end. A counted drop is not an error: the phone starts again. A lost +// uplink is. func (t *Tunnel) sendTun(m cloudwire.Tun, bulk bool) error { t.mu.Lock() s := t.sender @@ -508,8 +638,34 @@ func (t *Tunnel) allowOpenLocked(dev string) bool { return true } -// tick sends the one-second pulse to every live session. -func (t *Tunnel) tick() { +// pace is a token bucket: tunRate frames a second, tunBurst at once. +type pace struct { + tokens float64 + at time.Time +} + +func (p *pace) fill(now time.Time) { + p.tokens = min(tunBurst, p.tokens+now.Sub(p.at).Seconds()*tunRate) + p.at = now +} + +// wait is how long until a frame may go. +func (p *pace) wait(now time.Time) time.Duration { + p.fill(now) + if p.tokens >= 1 { + return 0 + } + return time.Duration((1 - p.tokens) / tunRate * float64(time.Second)) +} + +// take spends one frame. +func (p *pace) take(now time.Time) { + p.fill(now) + p.tokens-- +} + +// each runs one lane 0 step on every live session: Push or Beat. +func (t *Tunnel) each(what string, step func(*appproto.Handler) error) { t.mu.Lock() live := make([]*tunSession, 0, len(t.live)) for _, s := range t.live { @@ -517,8 +673,8 @@ func (t *Tunnel) tick() { } t.mu.Unlock() for _, s := range live { - if err := s.session.handler.Tick(); err != nil { - t.log.Warn("app session failed on a tick", "err", err) + if err := step(s.session.handler); err != nil { + t.log.Warn("app session failed on a "+what, "err", err) t.remove(s, true) } } diff --git a/go/internal/appuplink/tunnel_test.go b/go/internal/appuplink/tunnel_test.go index 3a6ee6089..3442c1ef7 100644 --- a/go/internal/appuplink/tunnel_test.go +++ b/go/internal/appuplink/tunnel_test.go @@ -46,7 +46,7 @@ func newTunRig(t *testing.T) *tunRig { } r := &tunRig{enroll: enroll, slots: &slots{}, out: &capturedTun{ch: make(chan cloudwire.Tun, 64)}, modes: &countingModes{}} - r.tunnel = newTunnel(handlerOver(r.modes, appproto.NewCmdLog()), enroll, r.slots, slog.New(slog.DiscardHandler)) + r.tunnel = newTunnel(handlerOver(r.modes, appproto.NewCmdLog()), enroll, r.slots, nil, slog.New(slog.DiscardHandler)) r.tunnel.Attach(r.out) ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) diff --git a/go/internal/appuplink/uplink_test.go b/go/internal/appuplink/uplink_test.go index 69ee9bf6a..cbc726ae0 100644 --- a/go/internal/appuplink/uplink_test.go +++ b/go/internal/appuplink/uplink_test.go @@ -120,6 +120,12 @@ func newRig(t *testing.T, epoch int64) *rig { // gateway it can hold open, so a revoke can land while a call is running. func newRigWith(t *testing.T, epoch int64, build HandlerBuilder) *rig { t.Helper() + return newRigOpts(t, epoch, func(o *Options) { o.Handler = build }) +} + +// newRigOpts is newRig with the options set changes. +func newRigOpts(t *testing.T, epoch int64, set func(*Options)) *rig { + t.Helper() relay := newFakeRelay(epoch) t.Cleanup(relay.close) @@ -130,14 +136,15 @@ func newRigWith(t *testing.T, epoch int64, build HandlerBuilder) *rig { } now := time.UnixMilli(epoch * EpochMs) - uplink, err := New(Options{ + opts := Options{ Endpoint: relay.url(), Enroll: enroll, - Handler: build, Logger: slog.New(slog.DiscardHandler), Now: func() time.Time { return now }, Random: func() float64 { return 0 }, - }) + } + set(&opts) + uplink, err := New(opts) if err != nil { t.Fatalf("New: %v", err) } diff --git a/go/internal/cloudlink/conn.go b/go/internal/cloudlink/conn.go index 1f50eae5c..5f751e007 100644 --- a/go/internal/cloudlink/conn.go +++ b/go/internal/cloudlink/conn.go @@ -223,8 +223,8 @@ func (l *Link) backoff() time.Duration { var errLinkEnded = errors.New("cloudlink: the end of the link is confirmed") // The tunnel's two queues, apart from the readable copy. Lane 0, a phone's -// commands, their answers and the one-second ticks, goes first and is never -// held. Bulk is paced under the service's limits and waits its turn. When a +// commands, their answers, its readings and its snapshot, goes first and is +// never held. Bulk is paced under the service's limits and waits its turn. When a // queue is full the newest message is dropped: Noise transport carries its // own counter, so a lost message costs the phone one frame. const ( @@ -552,6 +552,24 @@ func tunFrame(t cloudwire.Tun) ([]byte, error) { return cloudwire.EncodeFrame(raw) } +// MaxBulkWait is the longest a bulk tun can wait behind the pacer: a full +// bulk queue of the largest messages, paced to tunPace of the service's byte +// limit. A phone's Noise transport takes a message at most +// appwire.ReplayWindow behind the newest, so a session must send fewer lane +// 0 frames than that while one bulk message waits. +func MaxBulkWait() time.Duration { + seconds := float64(tunBulkQueue*cloudwire.MaxTunFrame) / (tunPace * cloudwire.TunBytesPerSecond) + return time.Duration(seconds * float64(time.Second)) +} + +// TunBudgetUsed is the share of the tunnel's 24-hour budget spent, from 0 +// to 1 and past it when the budget is gone. +func (l *Link) TunBudgetUsed() float64 { + l.mu.Lock() + defer l.mu.Unlock() + return float64(l.tunBudget.total(l.uptimeMs())) / tunBudgetBytes +} + // pacer keeps bulk under the service's tun limits: a message bucket and a // byte bucket at tunPace of them. Taking may leave either in debt; the wait // is how long until it is paid. diff --git a/go/internal/cloudlink/link.go b/go/internal/cloudlink/link.go index 5af9b2a05..e70689733 100644 --- a/go/internal/cloudlink/link.go +++ b/go/internal/cloudlink/link.go @@ -194,7 +194,11 @@ type Options struct { // Tunnel serves the phones' sessions. Nil takes no tun: the box does not // list tun in hello.caps, and the service sends it none. Tunnel Tunnel - Logger *slog.Logger + // Arrivals hears each finished driver poll (telemetry.Store.Arrivals). + // While a phone watches, live goes when a poll lands rather than on the + // next check. Nil leaves live on its checks alone. + Arrivals <-chan struct{} + Logger *slog.Logger // Seams for tests. Production leaves them zero. Endpoint string diff --git a/go/internal/cloudlink/link_test.go b/go/internal/cloudlink/link_test.go index 039f90860..8b4e38089 100644 --- a/go/internal/cloudlink/link_test.go +++ b/go/internal/cloudlink/link_test.go @@ -394,6 +394,8 @@ type harness struct { cancel context.CancelFunc done chan error pushed chan string + // arrivals is the driver registry's signal that a poll finished. + arrivals chan struct{} } func newHarness(t *testing.T, svc *fakeService) *harness { @@ -416,12 +418,13 @@ func newHarnessWith(t *testing.T, svc *fakeService, keys []*ecdsa.PublicKey, kv } h := &harness{svc: svc, site: &fakeSite{grid: 1200, state: protocol.SourceLive}, kv: kv, keyPath: filepath.Join(t.TempDir(), "cloud.key"), pushed: make(chan string, 8), - roster: &fakeRosterState{}, tunnel: &fakeTunnel{ch: make(chan cloudwire.Tun, 64)}} + roster: &fakeRosterState{}, tunnel: &fakeTunnel{ch: make(chan cloudwire.Tun, 64)}, + arrivals: make(chan struct{}, 1)} opts := Options{ Site: h.site, Roster: fakeRoster{state: h.roster}, KV: h.kv, KeyPath: h.keyPath, Build: "test", Started: time.Now().Add(-time.Minute), ClockSynced: func() bool { return true }, - Notify: func(kind string) { h.pushed <- kind }, - Tunnel: h.tunnel, + Notify: func(kind string) { h.pushed <- kind }, + Tunnel: h.tunnel, Arrivals: h.arrivals, Endpoint: svc.endpoint(), ServiceKeys: keys, RootCAs: svc.roots(), Tick: 20 * time.Millisecond, Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), @@ -669,6 +672,37 @@ func TestWatchRaisesTheCadence(t *testing.T) { } } +// While a phone watches, a reading goes when its poll lands rather than on +// the next look at the box, and never sooner than a period after the last. +func TestAWatchedReadingGoesAsItsPollLands(t *testing.T) { + svc := newFakeService(t) + svc.afterHello = func(conn *websocket.Conn, home string) { + svc.write(conn, nil, home, cloudwire.MsgWatch, cloudwire.Watch{Hz: 1}) + } + h := newHarness(t, svc) + h.grant(t) + svc.waitFor(cloudwire.MsgLive, 5*time.Second) + for i := range 3 { + time.Sleep(time.Duration(1100+150*i) * time.Millisecond) + h.site.setGrid(int64(2000 + i)) + start := time.Now() + h.arrivals <- struct{}{} + svc.waitFor(cloudwire.MsgLive, 2*time.Second) + if d := time.Since(start); d > 200*time.Millisecond { + t.Fatalf("a watched reading took %s after its poll", d) + } + } + + // A second poll right after: its live waits out the period. + h.site.setGrid(3000) + start := time.Now() + h.arrivals <- struct{}{} + svc.waitFor(cloudwire.MsgLive, 3*time.Second) + if d := time.Since(start); d < 900*time.Millisecond || d > 1300*time.Millisecond { + t.Fatalf("the next live went %s after the last, want one period", d) + } +} + func TestATableChangeGoesAtOnceWithANewRevision(t *testing.T) { svc := newFakeService(t) h := newHarness(t, svc) diff --git a/go/internal/cloudlink/sender.go b/go/internal/cloudlink/sender.go index d548bbc04..c8cda30b4 100644 --- a/go/internal/cloudlink/sender.go +++ b/go/internal/cloudlink/sender.go @@ -94,6 +94,9 @@ type cadence struct { // message could go. liveCheckedAt, lpCheckedAt time.Time liveWaiting, lpWaiting bool + // arrived says a driver finished a poll since live was last composed, + // while a phone watches. + arrived bool } // liveCheckEvery is how often the source table is looked at, so a table @@ -110,6 +113,8 @@ func (w *writer) run(ctx context.Context, events <-chan cloudwire.Envelope) erro // service counts that from when it received the last one. st := &cadence{full: true} var purgeSince time.Time + // slot fires when an arrival waits for the watched live's period. + var slot <-chan time.Time for { snap := l.current() @@ -139,13 +144,26 @@ func (w *writer) run(ctx context.Context, events <-chan cloudwire.Envelope) erro if err := w.tick(snap, st, now); err != nil { return err } + if wait := st.liveAt.Add(periodWatched).Sub(now); st.arrived && slot == nil && wait > 0 { + slot = time.After(wait) + } } + // Polls are heard only while a phone watches; otherwise the copy + // keeps its period and a poll needs no wakeup. + var arrivals <-chan struct{} + if time.Now().Before(st.watchUntil) { + arrivals = l.opts.Arrivals + } select { case <-ctx.Done(): return ctx.Err() case <-tick.C: case <-l.wake: + case <-arrivals: + st.arrived = true + case <-slot: + slot = nil case env := <-events: if err := w.downlink(snap, st, env); err != nil { return err @@ -251,9 +269,13 @@ func (w *writer) tick(snap grantSnapshot, st *cadence, now time.Time) error { full := st.full || now.Sub(st.lastFull) >= fullEvery hbDue := now.Sub(st.anyAt) >= h + // While a phone watches, a poll's readings go as they land, one period + // after the last live at the soonest, so the copy costs what it did. + arrival := st.arrived && now.Before(st.watchUntil) && now.Sub(st.liveAt) >= period + var ages map[string]int64 - if snap.allowed(cloudwire.MsgLive) && (full || hbDue || now.Sub(st.liveCheckedAt) >= liveCheckEvery) { - st.liveCheckedAt = now + if snap.allowed(cloudwire.MsgLive) && (full || hbDue || arrival || now.Sub(st.liveCheckedAt) >= liveCheckEvery) { + st.liveCheckedAt, st.arrived = now, false live, tableChanged := l.projectLive(snap, l.opts.Site.Live()) ages = live.Ages key := liveKey(live) diff --git a/go/internal/drivers/host.go b/go/internal/drivers/host.go index 2a01cf25f..4bb01f601 100644 --- a/go/internal/drivers/host.go +++ b/go/internal/drivers/host.go @@ -817,10 +817,12 @@ func (h *HostEnv) standingPollIntervalLocked() time.Duration { return time.Duration(h.PollIntervalMS) * time.Millisecond } -// Bounds for an interval a driver returns from driver_poll. Lane 0 frames -// leave once a second and the control loop ticks every two seconds by -// default, so a faster read reaches nothing but the device's bus; some -// catalog drivers return 200 ms or 50 ms. The ceiling is half the default +// Bounds for an interval a driver returns from driver_poll. The control +// loop ticks every two seconds by default, and a phone sees a reading once a +// second over the relay, so a read faster than a second mostly costs the +// device's bus; some catalog drivers return 200 ms or 50 ms. Through FTW +// Cloud each poll reaches the phone as it lands, up to five a second, so a +// driver that sets a faster interval is seen at that rate. The ceiling is half the default // 60 s watchdog window, so a returned wait cannot make a driver stale by // itself: myuplink returns 60 s after a good read. A driver that needs longer // sets it with host.set_poll_interval, as the vehicle drivers do. diff --git a/go/internal/drivers/registry.go b/go/internal/drivers/registry.go index 222e71ea9..f8e0d4127 100644 --- a/go/internal/drivers/registry.go +++ b/go/internal/drivers/registry.go @@ -1073,6 +1073,11 @@ func (r *Registry) runLoop(rd *runningDriver) { // re-stamps LastSuccess every tick from cached values. r.tel.RecordDriverTick(rd.cfg.Name) } + if r.tel != nil { + // The poll's readings are in the store: whoever streams + // them sends now, not on its next pulse. + r.tel.Arrived() + } // Re-arm at the interval the poll returned, or the standing one. interval = rd.env.nextPollInterval(returned) timer.Reset(interval) diff --git a/go/internal/ocpp/handlers.go b/go/internal/ocpp/handlers.go index 2873dda94..2d1fcdcd4 100644 --- a/go/internal/ocpp/handlers.go +++ b/go/internal/ocpp/handlers.go @@ -177,6 +177,9 @@ func (h *Handler) SetApprovedIDs(ids []string) { blob, _ := json.Marshal(map[string]any{"type": "ev", "w": 0.0, "forecast_power": telemetry.ForecastPowerSample{Version: 1}}) h.tel.Update(id, telemetry.DerEV, 0, nil, blob) } + if len(revoked) > 0 { + h.tel.Arrived() + } } // IsApproved reports whether a charger id is on the site allowlist and may @@ -663,6 +666,8 @@ func (h *Handler) pushReading(id string, s *chargerState) { } blob, _ := json.Marshal(data) h.tel.Update(id, telemetry.DerEV, w, nil, blob) + // A charger sends its readings unpolled; this is its poll's end. + h.tel.Arrived() } // powerSample is one Power.Active.Import after unit conversion. An empty phase diff --git a/go/internal/telemetry/arrivals.go b/go/internal/telemetry/arrivals.go new file mode 100644 index 000000000..6c5f19cbd --- /dev/null +++ b/go/internal/telemetry/arrivals.go @@ -0,0 +1,37 @@ +package telemetry + +// Arrived tells every reader of Arrivals that a driver finished a poll, so +// a reader that streams live values can send them as they land instead of +// on a clock of its own. The driver registry calls it once per poll, after +// the poll's readings are in the store: a poll's readings belong together, +// and the house load is derived from all of them at once, so a reader that +// woke on each emit could send a load that never existed. +func (s *Store) Arrived() { + s.arrivalsMu.Lock() + defer s.arrivalsMu.Unlock() + for ch := range s.arrivals { + select { + case ch <- struct{}{}: + default: + // One is waiting already; this poll merges into it. + } + } +} + +// Arrivals returns a channel that receives after each finished poll, and a +// func that stops it. Polls that finish while a signal waits merge into +// that signal, so a slow reader never holds up a driver. +func (s *Store) Arrivals() (<-chan struct{}, func()) { + ch := make(chan struct{}, 1) + s.arrivalsMu.Lock() + if s.arrivals == nil { + s.arrivals = map[chan struct{}]struct{}{} + } + s.arrivals[ch] = struct{}{} + s.arrivalsMu.Unlock() + return ch, func() { + s.arrivalsMu.Lock() + delete(s.arrivals, ch) + s.arrivalsMu.Unlock() + } +} diff --git a/go/internal/telemetry/arrivals_test.go b/go/internal/telemetry/arrivals_test.go new file mode 100644 index 000000000..4b61d9699 --- /dev/null +++ b/go/internal/telemetry/arrivals_test.go @@ -0,0 +1,36 @@ +package telemetry + +import "testing" + +// Every reader hears a finished poll; polls that finish while one waits +// merge into it; a stopped reader hears nothing more. +func TestArrivalsReachEveryReaderAndMerge(t *testing.T) { + s := NewStore() + a, stopA := s.Arrivals() + b, stopB := s.Arrivals() + defer stopB() + + s.Arrived() + s.Arrived() + for name, ch := range map[string]<-chan struct{}{"a": a, "b": b} { + select { + case <-ch: + default: + t.Fatalf("reader %s heard no arrival", name) + } + select { + case <-ch: + t.Fatalf("reader %s heard two polls as two", name) + default: + } + } + + stopA() + s.Arrived() + select { + case <-a: + t.Fatal("a stopped reader heard an arrival") + default: + } + <-b +} diff --git a/go/internal/telemetry/store.go b/go/internal/telemetry/store.go index 3e65f4eeb..427d04353 100644 --- a/go/internal/telemetry/store.go +++ b/go/internal/telemetry/store.go @@ -314,6 +314,10 @@ type Store struct { // waiting for the control loop to flush it to the TS DB. latestMu sync.RWMutex latestMetric map[string]metricSnap + + // arrivals hear of each finished poll (see Arrived). + arrivalsMu sync.Mutex + arrivals map[chan struct{}]struct{} } type metricSnap struct { diff --git a/go/protocol/appwire/transport.go b/go/protocol/appwire/transport.go index 0a297d93d..b372da087 100644 --- a/go/protocol/appwire/transport.go +++ b/go/protocol/appwire/transport.go @@ -29,9 +29,10 @@ const ( TransportOverhead = SeqBytes + TagBytes // ReplayWindow is how far out of order a frame may arrive and still be - // accepted. At 1 Hz on lane 0 that is a minute of tolerance, far beyond - // any reordering a DataChannel produces, and small enough that the window - // is one machine word. + // accepted: one machine word. Over the relay lane 0 goes once a second, + // so that is a minute. Through FTW Cloud it goes up to five times a + // second, about 13 s, while a bulk frame the link paces waits at most + // cloudlink.MaxBulkWait, about 9 s (appuplink tests the two). ReplayWindow = 64 ) diff --git a/go/protocol/envelope.go b/go/protocol/envelope.go index 08321e5a2..b77474638 100644 --- a/go/protocol/envelope.go +++ b/go/protocol/envelope.go @@ -36,10 +36,12 @@ const ( ProtoFloor = 0 ) -// Lanes. Lane 0 is one fixed bucket at a fixed cadence for the life of a -// session, because a frame size that varies with what happened in the house -// leaks the household's load pattern through otherwise perfect encryption. -// Anything whose size depends on what the box supports goes on lane 1. +// Lanes. Lane 0 is one fixed bucket for the life of a session, because a +// frame size that varies with what happened in the house leaks the +// household's load pattern through otherwise perfect encryption. Over the +// relay it also goes at a fixed cadence; through FTW Cloud each reading goes +// as it lands (docs/protocol-cloud.md, "Tunnel"). Anything whose size +// depends on what the box supports goes on lane 1. const ( LaneControl uint8 = 0 LaneBulk uint8 = 1 diff --git a/go/protocol/messages.go b/go/protocol/messages.go index 8479b8dc4..5285f8a99 100644 --- a/go/protocol/messages.go +++ b/go/protocol/messages.go @@ -313,9 +313,11 @@ type Delta struct { // Tick says nothing changed. // -// Sent anyway, on the same cadence and in the same bucket. Silence would -// itself be information: it would tell the relay operator that nothing -// happened in the house that second. +// Sent anyway, in the same bucket. Over the relay it goes on every pulse: +// silence would tell the relay operator that nothing happened in the house +// that second. Through FTW Cloud it is the beat that goes when nothing has +// moved for the subscribed period, so the phone can tell a quiet house from +// a lost session. type Tick struct { Seq uint64 `cbor:"seq"` UptimeMs int64 `cbor:"uptimeMs"`