diff --git a/README.md b/README.md index ac6118e..dbd5acf 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ to it yet. messages signed by that box's key. - Each home's newest message of each class lives in memory, as the exact bytes the box sent. Readings are dropped an hour after the box goes quiet and are - never written to disk. + never written to disk. History tiles stay longer, also in memory only. - What the service holds follows the newest roster's grant, never `hello`: the readable copy can turn on or off while the uplink stays up. - The registry (SQLite) holds the home id, the box key, the signed roster, @@ -132,6 +132,29 @@ to it yet. The widgets then read `/v1/now`. At most 4 an hour per home, since Apple budgets widget reloads. A device holds at most 4 tokens of each kind; a new one pushes out its oldest. +- **History tiles (C6).** A box whose owner's grant has `ftw.history.read` + sends its last 48 hours of 5-minute history as `tiles`: 12-hour tiles, each + closed one the same bytes the box serves over Noise, and the tile still + filling whole or as an append when a bucket closes. The service checks + each against the protocol's three shapes and the four series, refuses an + envelope over 4 KiB, and holds it in memory as the box signed it: per home + at most five closed tiles, 16 pieces of the filling one and 64 KiB, all + homes together at most 256 MiB. A closed tile is replaced only by one the + box sent later. A tile the box did not sign closes the uplink, as any such + message does; one of no shape is dropped and counted. `hello_ok.have` + names the closed tiles held, so a box that comes back sends only what is + missing, and the service sends `resync {what: [tiles]}`, at most once per + 5 minutes, when a roster gives history back or room frees after a drop. A + phone with `ftw.history.read` asks with `tiles.get {have}`, gets the tiles + it lacks and then `tiles.end {n}`, and gets each new tile as it comes. The + snapshot, `/v1/app/once` and `/v1/now` carry none. Tiles are dated history, + so they outlive the hour after which readings go: each stays until 48 + hours after it ends, up to 60 hours of history, while the roster is in + force. With the box away a phone can therefore read them for 18 to 24 + hours. A revoke, a grant without history, a roster that runs out, a purge + and the 90-day expiry delete them. Assistants read history from the box, + never from these tiles; while the box is away, `history_query` says so and + when FTW Cloud last heard from it. - **Widget token.** A device asks for its extension token over its app session with `token.get`. A widget reads `GET /v1/now` with it: the newest `live`, `prices` and `hb` as the box signed them, and nothing else. The @@ -309,7 +332,6 @@ Not built yet: | Step | What | Where | |---|---|---| -| C6 | Tiles | `internal/server` | | C7–C8 | Statistics | implemented with collection off; review pending | ## How it holds a home diff --git a/go.mod b/go.mod index f974bc9..99ea8c5 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/fxamacker/cbor/v2 v2.9.4 github.com/gorilla/websocket v1.5.3 github.com/quic-go/quic-go v0.63.0 - github.com/srcfl/ftw/go/protocol v0.0.0-20261018181148-499d5de65104 + github.com/srcfl/ftw/go/protocol v0.0.0-20261023161258-bac597f77f0a golang.org/x/crypto v0.57.0 golang.org/x/net v0.58.0 golang.org/x/sys v0.48.0 diff --git a/go.sum b/go.sum index 528c4e3..091766b 100644 --- a/go.sum +++ b/go.sum @@ -52,8 +52,8 @@ github.com/quic-go/quic-go v0.63.0 h1:LIFGHI4PFUhhw2dDD1ARHdCff143ffMHwZtbnbuJ78 github.com/quic-go/quic-go v0.63.0/go.mod h1:RAro2j2yN9a9EiPACLHT9IB2NXCvGQmmo/alT0yYI0w= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= -github.com/srcfl/ftw/go/protocol v0.0.0-20261018181148-499d5de65104 h1:nwfkV8mMubBLB+cx7mxFYrVT6Asgl8ctXWfjXci8MZU= -github.com/srcfl/ftw/go/protocol v0.0.0-20261018181148-499d5de65104/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023161258-bac597f77f0a h1:1Bxa1t8Kum4ECjMCogXw87M4RyfMJFJT7vWzCXLyw08= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023161258-bac597f77f0a/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= diff --git a/internal/appsession/appsession.go b/internal/appsession/appsession.go index 4b94a4a..25f60db 100644 --- a/internal/appsession/appsession.go +++ b/internal/appsession/appsession.go @@ -757,6 +757,12 @@ func (c *phoneConn) apply(frame []byte) (int, error) { if c.s.Push != nil { return 0, pushed(c.s.Push.Unregister(c.home, c.dev.Device(), r)) } + case cloudwire.MsgTilesGet: + var g cloudwire.TilesGet + if err := cloudwire.Unmarshal(env.B, &g); err != nil { + return cloudwire.CloseProtocol, err + } + return 0, c.tiles(g.Have) case cloudwire.MsgTokenGet: if c.s.Tokens == nil { return 0, nil @@ -770,12 +776,33 @@ func (c *phoneConn) apply(frame []byte) (int, error) { return 0, nil } +// tiles answers tiles.get: the box's tiles envelopes the phone lacks, as +// the box signed them, then tiles.end with how many went. A device that may +// not read tiles gets tiles.end {n: 0}. +func (c *phoneConn) tiles(have map[string]string) error { + envs, err := c.dev.Tiles(have) + if err != nil { + return err + } + for _, env := range envs { + if err := c.reply(env); err != nil { + return err + } + } + return c.say(cloudwire.MsgTilesEnd, cloudwire.TilesEnd{N: len(envs)}) +} + // say hands an answer to the session's goroutine to send. func (c *phoneConn) say(t string, body any) error { env, err := cloudwire.Unsigned(t, body) if err != nil { return err } + return c.reply(env) +} + +// reply hands one envelope to the session's goroutine to send. +func (c *phoneConn) reply(env []byte) error { select { case c.replies <- env: return nil diff --git a/internal/appsession/tiles_test.go b/internal/appsession/tiles_test.go new file mode 100644 index 0000000..0cf22ad --- /dev/null +++ b/internal/appsession/tiles_test.go @@ -0,0 +1,197 @@ +package appsession + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "testing" + "time" + + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +type tileVectorFile struct { + BoxKey struct { + D string `json:"d"` + Home string `json:"home"` + } `json:"boxKey"` + Tiles []struct { + Name string `json:"name"` + TileID string `json:"tileId"` + Etag string `json:"etag"` + Envelope string `json:"envelope"` + } `json:"tiles"` + Have map[string][]string `json:"have"` + Get struct { + Body string `json:"body"` + } `json:"tilesGet"` + End struct { + Envelope string `json:"envelope"` + } `json:"tilesEnd"` +} + +// tileHarness is the harness on the vectors' box, at 17:16 on the vectors' +// day, with ftw.history.read in the grant and the vectors' tiles held. +func tileHarness(t *testing.T) (*harness, tileVectorFile) { + t.Helper() + var v tileVectorFile + readVectors(t, "cloud-tile-vectors.json", &v) + h := newHarness(t) + h.shift.Store(int64(time.Until(time.UnixMilli(1_791_806_400_000 + (5*60+16)*60_000)))) + key, err := ecdsa.ParseRawPrivateKey(elliptic.P256(), unhex(t, v.BoxKey.D)) + if err != nil { + t.Fatal(err) + } + pk, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) + if err := h.reg.Claim(v.BoxKey.Home, pk, time.Now()); err != nil { + t.Fatal(err) + } + box, err := h.store.For(storePrincipalBox(v.BoxKey.Home)) + if err != nil { + t.Fatal(err) + } + box.Connected() + h.boxKey, h.home, h.box = key, v.BoxKey.Home, box + h.setRoster(h.devices(1, 1), protocol.ScopeLiveRead, protocol.ScopeHistoryRead) + return h, v +} + +// putTile holds one of the vectors' envelopes, as the uplink does once it +// has verified and checked it. +func (h *harness) putTile(t *testing.T, raw []byte) { + t.Helper() + env, err := cloudwire.DecodeEnvelope(raw) + if err != nil { + t.Fatal(err) + } + if cloudwire.Verify(&h.boxKey.PublicKey, h.home, env) != nil { + t.Fatal("a vector does not verify") + } + var b cloudwire.Tiles + if err := cloudwire.Unmarshal(env.B, &b); err != nil || b.Check() != nil { + t.Fatalf("a vector is not a tile: %v", err) + } + if kept, _ := h.box.PutTile(raw, b); !kept { + t.Fatal("a vector tile was not held") + } +} + +// tilesAnswer reads the answer to one tiles.get: the envelopes, then +// tiles.end, which it returns. +func (p *phone) tilesAnswer(t *testing.T) ([][]byte, cloudwire.TilesEnd) { + t.Helper() + var got [][]byte + for { + env, code := p.next(2 * time.Second) + if code != 0 { + t.Fatalf("closed with %d", code) + } + e, err := cloudwire.DecodeEnvelope(env) + if err != nil { + t.Fatal(err) + } + if e.T == cloudwire.MsgTilesEnd { + var end cloudwire.TilesEnd + if len(e.Sig) != 0 || cloudwire.Unmarshal(e.B, &end) != nil { + t.Fatal("tiles.end is not the service's own") + } + return got, end + } + got = append(got, env) + } +} + +func TestTilesGetAnswersWhatThePhoneLacksAsTheBoxSignedIt(t *testing.T) { + h, v := tileHarness(t) + for _, tv := range v.Tiles { + h.putTile(t, unhex(t, tv.Envelope)) + } + p := h.mustDial(dialOpts{key: h.owner, hz: 1}) + if len(p.snap.Envelopes) != 0 { + t.Fatalf("message 2 carries %d envelopes", len(p.snap.Envelopes)) + } + + p.send(cloudwire.MsgTilesGet, cloudwire.TilesGet{}) + got, end := p.tilesAnswer(t) + if end.N != 4 || len(got) != 4 { + t.Fatalf("tiles.get {} answered %d, tiles.end says %d", len(got), end.N) + } + for i, tv := range v.Tiles { + if !bytes.Equal(got[i], unhex(t, tv.Envelope)) { + t.Fatalf("envelope %d (%s) is not the box's bytes", i, tv.Name) + } + } + + // The vectors' have holds the closed tile: the pieces come alone, as + // the vectors say. + var g cloudwire.TilesGet + if err := cloudwire.Unmarshal(unhex(t, v.Get.Body), &g); err != nil { + t.Fatal(err) + } + p.send(cloudwire.MsgTilesGet, g) + got, end = p.tilesAnswer(t) + if end.N != len(v.Have["answer"]) || len(got) != end.N { + t.Fatalf("with have, %d envelopes and n %d", len(got), end.N) + } + for i, name := range v.Have["answer"] { + if !bytes.Equal(got[i], unhex(t, v.Tiles[i+1].Envelope)) || v.Tiles[i+1].Name != name { + t.Fatalf("answer %d is not %s", i, name) + } + } +} + +func TestATileGoesToTheOwnersSessionAsItComesAndNotToAViewers(t *testing.T) { + h, v := tileHarness(t) + owner := h.mustDial(dialOpts{key: h.owner, hz: 1}) + viewer := h.mustDial(dialOpts{key: h.viewer, hz: 1}) + whole := unhex(t, v.Tiles[1].Envelope) + h.putTile(t, whole) + env, code := owner.next(2 * time.Second) + if code != 0 || !bytes.Equal(env, whole) { + t.Fatal("the owner did not get the tile as the box signed it") + } + // The viewer may not read history: tiles.get answers nothing, and the + // tile never came. + viewer.send(cloudwire.MsgTilesGet, cloudwire.TilesGet{}) + got, end := viewer.tilesAnswer(t) + if len(got) != 0 || end.N != 0 { + t.Fatalf("a viewer got %d tiles", len(got)) + } +} + +// The box went away hours ago: no readings, but the tiles of the last 48 +// hours still come, as the box dated them. +func TestTilesComeWhileTheBoxIsAway(t *testing.T) { + h, v := tileHarness(t) + h.live(true) + for _, tv := range v.Tiles { + h.putTile(t, unhex(t, tv.Envelope)) + } + h.box.Disconnected(1) + h.shift.Add(int64(3 * time.Hour)) + h.store.Sweep() + p := h.mustDial(dialOpts{key: h.owner}) + if len(p.snap.Envelopes) != 0 || p.snap.Online { + t.Fatalf("after three hours away the snapshot holds %d readings", len(p.snap.Envelopes)) + } + p.send(cloudwire.MsgTilesGet, cloudwire.TilesGet{}) + if got, end := p.tilesAnswer(t); end.N != 4 || len(got) != 4 { + t.Fatalf("away, tiles.get answered %d", len(got)) + } +} + +// tiles.end is the bytes srcfl/ftw froze. +func TestTilesEndIsTheVector(t *testing.T) { + var v tileVectorFile + readVectors(t, "cloud-tile-vectors.json", &v) + env, err := cloudwire.Unsigned(cloudwire.MsgTilesEnd, cloudwire.TilesEnd{N: 3}) + if err != nil || !bytes.Equal(env, unhex(t, v.End.Envelope)) { + t.Fatal("tiles.end differs from the vector") + } +} + +func storePrincipalBox(home string) store.Principal { + return store.Principal{Kind: store.Box, Home: home} +} diff --git a/internal/mcp/agent.go b/internal/mcp/agent.go index 43137cd..6b09f0c 100644 --- a/internal/mcp/agent.go +++ b/internal/mcp/agent.go @@ -582,6 +582,17 @@ func (c *agentConn) send(ctx context.Context, t string, id *uint32, body any) er return nil } +// away says whether a home's box is away, and when the service last heard +// from it, for an agent the home still admits. +func (p *Pool) away(sub oauth.Subject) (time.Time, bool) { + h, err := p.store.For(store.Principal{Kind: store.Agent, Home: sub.Home, DeviceKey: sub.Key}) + if err != nil { + return time.Time{}, false + } + online, seen := h.Presence() + return seen, !online +} + // take waits for the session's turn. At most maxWaiting calls wait. func (c *agentConn) take(ctx context.Context) (func(), error) { c.mu.Lock() diff --git a/internal/mcp/history_test.go b/internal/mcp/history_test.go index 9fa8a6b..73f9947 100644 --- a/internal/mcp/history_test.go +++ b/internal/mcp/history_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw/go/protocol" ) @@ -124,3 +125,43 @@ func TestStepLabel(t *testing.T) { } } } + +// A box that is away: history_query says so, and when FTW Cloud last heard +// from it. It reads nothing from FTW Cloud's copy instead. +func TestHistoryFromABoxThatIsAwaySaysWhenItWasLastSeen(t *testing.T) { + w := newWorld(t) + b := w.connectBox() + a := w.connect(b, "") + if _, isErr, text := a.call("history_query", map[string]any{"range": "1h"}); isErr { + t.Fatalf("history_query with the box up: %s", text) + } + + _ = b.conn.Close() + box, err := w.st.For(store.Principal{Kind: store.Box, Home: b.home}) + if err != nil { + t.Fatal(err) + } + waitFor(t, func() bool { online, _ := box.Presence(); return !online }) + _, seen := box.Presence() + _, isErr, text := a.call("history_query", map[string]any{"range": "1h"}) + if !isErr || !strings.Contains(text, "The box is away") || + !strings.Contains(text, seen.UTC().Format(time.RFC3339)) || !strings.Contains(text, "ago") { + t.Fatalf("history_query with the box away: %v %s", isErr, text) + } + b.mu.Lock() + defer b.mu.Unlock() + if n := strings.Count(strings.Join(b.asked, "\n"), protocol.MsgHistQuery+" "); n != 1 { + t.Fatalf("the box was asked hist.query %d times", n) + } +} + +func TestAgoWords(t *testing.T) { + for d, want := range map[time.Duration]string{ + 10 * time.Second: "under a minute", time.Minute: "1 minute", 59 * time.Minute: "59 minutes", + 3 * time.Hour: "3 hours", 47 * time.Hour: "47 hours", 72 * time.Hour: "3 days", + } { + if got := agoWords(d); got != want { + t.Errorf("%s: %q, want %q", d, got, want) + } + } +} diff --git a/internal/mcp/tools.go b/internal/mcp/tools.go index 78020c1..361ee3d 100644 --- a/internal/mcp/tools.go +++ b/internal/mcp/tools.go @@ -296,13 +296,13 @@ func readTools() []*tool { FromMs: fromMs, ToMs: toMs, MaxPoints: &ask} c, err := cc.conn(ctx) if err != nil { - return nil, err + return nil, cc.away(err) } ctx, cancel := context.WithTimeout(ctx, cc.s.pool.Call) defer cancel() chunks, end, err := c.hist(ctx, q) if err != nil { - return nil, words(err) + return nil, cc.away(words(err)) } return historyData(chunks, end, fromMs, toMs, int(points)), nil }, @@ -494,6 +494,41 @@ func tilePoints(res protocol.Resolution, fromMs, toMs int64) int64 { return min((last+span-first)/protocol.ResolutionStepMs[res], 2000) } +// away puts a box that is away in words: when FTW Cloud last heard from it, +// and that its history stays there. An agent never reads FTW Cloud's +// readable copy, so every read it makes is one the box audits and a revoke +// stops (srcfl/ftw docs/protocol-cloud.md, "Who may open an agent's +// session"), and there is nothing to give instead. err goes back unchanged +// while the box is up. +func (cc *callCtx) away(err error) error { + seen, away := cc.s.pool.away(cc.grant.Subject) + if !away || seen.IsZero() { + return err + } + return userError(fmt.Sprintf("The box is away. FTW Cloud last heard from it at %s, %s ago. Its history is read from the box only, so ask again when it is back.", + seen.UTC().Format(time.RFC3339), agoWords(cc.s.pool.now().Sub(seen)))) +} + +// agoWords is a time since in the largest whole unit: "under a minute", +// "1 minute", "3 hours", "2 days". +func agoWords(d time.Duration) string { + unit := func(n int64, one string) string { + if n == 1 { + return "1 " + one + } + return fmt.Sprintf("%d %ss", n, one) + } + switch { + case d < time.Minute: + return "under a minute" + case d < time.Hour: + return unit(int64(d/time.Minute), "minute") + case d < 48*time.Hour: + return unit(int64(d/time.Hour), "hour") + } + return unit(int64(d/(24*time.Hour)), "day") +} + // historyData unpacks hist.chunk tiles into one list of points per series // over [fromMs, toMs), at most most of them: the box's tiles reach past both // ends of the range, into the future too, so points outside it are dropped, diff --git a/internal/store/inspect_dev.go b/internal/store/inspect_dev.go index 9b34672..ef4789f 100644 --- a/internal/store/inspect_dev.go +++ b/internal/store/inspect_dev.go @@ -2,6 +2,8 @@ package store +import "strconv" + // Inspect returns every envelope a home holds. It exists only in local test // binaries built with -tags ftwclouddev, so an end-to-end run can compare // what the service holds with what the box says it sent. A release build has @@ -22,5 +24,11 @@ func (s *Store) Inspect(id string) (latest map[string][]byte, online bool, ok bo if h.full != nil { latest["live+sources"] = h.full.Envelope } + for id, c := range h.tiles.closed { + latest["tiles "+id] = c.env + } + for i, p := range h.tiles.open { + latest["tiles "+p.id+" piece "+strconv.Itoa(i)] = p.env + } return latest, h.online, true } diff --git a/internal/store/store.go b/internal/store/store.go index 87eb21a..8a31a7b 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -16,6 +16,7 @@ import ( "slices" "sort" "sync" + "sync/atomic" "time" "github.com/srcfl/ftw/go/protocol" @@ -76,6 +77,9 @@ type Store struct { // suspended reports an owner's suspension of a roster device. Nil // suspends nobody. suspended func(home, deviceID string, epoch uint64) bool + // tileBytes is what every home's tiles hold, at most tileBytesMax. + tileBytes atomic.Int64 + tileBytesMax int64 } // New starts an empty store. now is for tests; nil uses the clock. @@ -83,7 +87,7 @@ func New(now func() time.Time) *Store { if now == nil { now = time.Now } - return &Store{homes: map[string]*home{}, now: now} + return &Store{homes: map[string]*home{}, now: now, tileBytesMax: TileBytesMax} } // Homes is how many homes the store holds. @@ -118,6 +122,9 @@ type home struct { pending map[string]time.Time spent map[string]time.Time invitePending map[string]time.Time + // tiles are the last 48 hours of history, kept apart from latest: they + // outlive the hour after which readings go, and a snapshot carries none. + tiles tileSet } // Held is one envelope as received. @@ -368,6 +375,7 @@ func (x *Handle) DropCopy() { x.h.mu.Lock() defer x.h.mu.Unlock() x.h.latest, x.h.full = map[string]Held{}, nil + x.s.dropTiles(x.h) } // Put keeps env as the newest of its class. @@ -436,6 +444,9 @@ func (x *Handle) SetRoster(r *cloudwire.Roster) bool { // The copy is off or withdrawn: no reading stays in memory. x.h.latest, x.h.full = map[string]Held{}, nil } + if !x.h.holdsTilesLocked() { + x.s.dropTiles(x.h) + } x.h.setPendingLocked(r, x.h.rosterAt) x.recheckLocked() return true @@ -523,6 +534,7 @@ func (x *Handle) Delete() { x.s.mu.Unlock() x.h.mu.Lock() x.h.latest, x.h.full, x.h.roster = map[string]Held{}, nil, nil + x.s.dropTiles(x.h) for sess := range x.h.sessions { sess.endLocked(cloudwire.CloseDenied) } @@ -542,6 +554,15 @@ type Snapshot struct { Online bool } +// Presence is whether the box's uplink is up and when the service last heard +// from it: the service's own word, never a reading. Any principal the home +// admits may ask, an agent too. +func (x *Handle) Presence() (online bool, lastSeen time.Time) { + x.h.mu.Lock() + defer x.h.mu.Unlock() + return x.h.online, x.h.lastSeen +} + // Snapshot returns the device's view. func (x *Handle) Snapshot() (Snapshot, error) { if x.kind != Device { @@ -609,7 +630,8 @@ func (x *Handle) mayRead(class string) bool { // -------------------------------------------------------------------------- // Sweep drops the readings of every home whose box has been gone longer than -// ReadingsKeptFor. What the box was last seen at stays. +// ReadingsKeptFor, and every tile past its 48 hours. What the box was last +// seen at stays. func (s *Store) Sweep() { now := s.now() s.mu.Lock() @@ -623,6 +645,7 @@ func (s *Store) Sweep() { if !h.online && now.Sub(h.lastSeen) > ReadingsKeptFor { h.latest, h.full = map[string]Held{}, nil } + s.sweepTiles(h, now) // A roster that ran out closes the home's sessions: the box has not // confirmed who may read for a day. if r := h.roster; r != nil && now.After(h.rosterAt.Add(time.Duration(r.TTLS)*time.Second)) { @@ -672,6 +695,7 @@ func (s *Store) Forget(id string) { s.mu.Unlock() if h != nil { h.mu.Lock() + s.dropTiles(h) for sess := range h.sessions { sess.endLocked(cloudwire.CloseDenied) } diff --git a/internal/store/tiles.go b/internal/store/tiles.go new file mode 100644 index 0000000..57f5ceb --- /dev/null +++ b/internal/store/tiles.go @@ -0,0 +1,317 @@ +package store + +import ( + "slices" + "sort" + "time" + + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" + + "github.com/srcfl/ftw-cloud/internal/metrics" +) + +// Tiles: each home's last 48 hours of 5-minute history, as the box signed +// it, so a phone can draw history while the box cannot answer +// (docs/protocol-cloud.md, tiles). They are dated history, never a reading +// of now, so they outlive the hour after which the box's readings go; and +// like the readings they live only in memory, only while the roster's grant +// serves a copy with ftw.history.read. + +// TileBytesMax bounds what every home's tiles hold together, and +// TileBytesPerHome what one home's do. A home of five closed tiles and 16 +// pieces holds about 20 KB, so the first is some 13,000 homes and at least +// 4,000 at the second; a shard takes on no more than 4,000. Past either, new +// tiles are dropped and counted. +const ( + TileBytesMax = 256 << 20 + TileBytesPerHome = 64 << 10 +) + +var ( + tileBytesHeld = metrics.Default.Gauge("ftwcloud_tile_bytes", "Bytes of history tiles held in memory, all homes.") + dropTileOld = metrics.Drop("tile_outside_48h") + dropTileFull = metrics.Drop("tile_memory_full") + dropTileClosed = metrics.Drop("tile_piece_after_close") + dropTilePieces = metrics.Drop("tile_too_many_pieces") + dropTileHome = metrics.Drop("tile_home_full") + dropTileOlder = metrics.Drop("tile_older_than_held") +) + +// heldTile is one tiles envelope as the box sent it. +type heldTile struct { + id string + etag string + start int64 // the tile's start, wall clock + // wallMs and seq are when the box sent it, in the order it sent them: + // wall time first, since seq starts again when the box does. + wallMs int64 + seq uint64 + env []byte +} + +// before says a was sent before b. +func (a heldTile) before(b heldTile) bool { + return a.wallMs < b.wallMs || (a.wallMs == b.wallMs && a.seq < b.seq) +} + +func (t heldTile) end() int64 { + return t.start + protocol.ResolutionTileSpanMs[cloudwire.TileRes] +} + +// tileSet is one home's tiles: each closed tile by id, and the pieces of the +// trailing tile in the order they came. A whole piece replaces the pieces +// before it, so laying them in order rebuilds the tile. +type tileSet struct { + closed map[string]heldTile + open []heldTile + bytes int64 +} + +// PutTile keeps one tiles envelope the box signed, which the caller has +// verified and checked (cloudwire.Tiles.Check), and forwards it to the +// phones reading the home that may read tiles. kept is false when it was +// not kept: the grant holds no tiles, it is older than 48 hours or from the +// future, sent before what is held of its tile, a piece of a tile already +// closed, an append past the pieces a tile may have, or memory for tiles is +// full, which full says. +func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) (kept, full bool) { + if x.kind != Box { + return false, false + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + x.h.lastSeen = x.s.now() + if !x.h.holdsTilesLocked() { + return false, false + } + now := x.s.now().UnixMilli() + x.pruneTilesLocked(now) + held := heldTile{id: t.TileID, etag: t.Etag, start: t.TileStart(), seq: t.Seq, env: env} + if t.WallMs != nil { + held.wallMs = *t.WallMs + } + if held.end() <= now-cloudwire.TileKeepMs || held.start > now+protocol.ResolutionTileSpanMs[cloudwire.TileRes] { + dropTileOld.Inc() + return false, false + } + ts := &x.h.tiles + // What the tile replaces, first, so the memory check counts the net + // change: a whole piece or a closed tile often frees more than it takes. + var replaced int64 + dropOpen := false + if !t.Partial { + if old, ok := ts.closed[held.id]; ok { + if held.before(old) { + // An older version of a tile the box sent again since. + dropTileOlder.Inc() + return false, false + } + replaced += int64(len(old.env)) + } + dropOpen = len(ts.open) > 0 && ts.open[0].id == held.id + } else { + if _, ok := ts.closed[held.id]; ok { + dropTileClosed.Inc() + return false, false + } + if n := len(ts.open); n > 0 && ts.open[0].id == held.id && held.before(ts.open[n-1]) { + // A piece that came after one the box sent later. + dropTileOlder.Inc() + return false, false + } + if len(ts.open) > 0 && ts.open[0].id != held.id { + if held.start < ts.open[0].start { + // A piece of an older tile than the one filling. + dropTileClosed.Inc() + return false, false + } + dropOpen = true + } + dropOpen = dropOpen || t.Whole() + if !dropOpen && len(ts.open) >= cloudwire.TilePiecesMax { + // More appends than a box sends between whole pieces. The + // pieces held still lay back into the tile; this one waits + // for the box's next whole piece, which carries it too. + dropTilePieces.Inc() + return false, false + } + } + if dropOpen { + for _, p := range ts.open { + replaced += int64(len(p.env)) + } + } + if grow := int64(len(env)) - replaced; grow > 0 { + if ts.bytes+grow > TileBytesPerHome { + dropTileHome.Inc() + return false, true + } + if x.s.tileBytes.Load()+grow > x.s.tileBytesMax { + dropTileFull.Inc() + return false, true + } + } + if dropOpen { + x.dropOpenLocked() + } + if !t.Partial { + if old, ok := ts.closed[held.id]; ok { + x.tileBytesLocked(-int64(len(old.env))) + } + if ts.closed == nil { + ts.closed = map[string]heldTile{} + } + ts.closed[held.id] = held + x.tileBytesLocked(int64(len(env))) + // Never more than 48 hours can touch. + for len(ts.closed) > cloudwire.TilesClosedMax { + oldest := "" + for id, c := range ts.closed { + if oldest == "" || c.start < ts.closed[oldest].start { + oldest = id + } + } + x.tileBytesLocked(-int64(len(ts.closed[oldest].env))) + delete(ts.closed, oldest) + } + } else { + ts.open = append(ts.open, held) + x.tileBytesLocked(int64(len(env))) + } + at := stamp() + for sess := range x.h.sessions { + if sess.x.mayRead(cloudwire.MsgTiles) { + sess.deliver(env, at) + } + } + return true, false +} + +// TileRoom says whether a tile dropped for lack of memory would be kept now: +// the store and the home are both well under their bounds. +func (x *Handle) TileRoom() bool { + x.h.mu.Lock() + defer x.h.mu.Unlock() + return x.s.tileBytes.Load() < x.s.tileBytesMax/10*9 && x.h.tiles.bytes < TileBytesPerHome/4*3 +} + +// HoldsTiles says whether the roster in force lets the service hold tiles. +func (x *Handle) HoldsTiles() bool { + x.h.mu.Lock() + defer x.h.mu.Unlock() + return x.h.holdsTilesLocked() +} + +// TilesHave is each closed tile the home holds, by id, for hello_ok. +func (x *Handle) TilesHave() map[string]string { + have := map[string]string{} + if x.kind != Box { + return have + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + x.pruneTilesLocked(x.s.now().UnixMilli()) + for id, c := range x.h.tiles.closed { + have[id] = c.etag + } + return have +} + +// Tiles answers tiles.get: each closed tile the device lacks at its etag, +// oldest first, then every piece of the trailing tile in the order the box +// sent them, each envelope as the box signed it. A device that may not read +// tiles gets none. It answers while the box is away too: tiles are history, +// dated by the box. +func (x *Handle) Tiles(have map[string]string) ([][]byte, error) { + if x.kind != Device { + return nil, ErrDenied + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + if !x.mayRead(cloudwire.MsgTiles) || !x.h.holdsTilesLocked() { + return nil, nil + } + x.pruneTilesLocked(x.s.now().UnixMilli()) + closed := make([]heldTile, 0, len(x.h.tiles.closed)) + for _, c := range x.h.tiles.closed { + if have[c.id] != c.etag { + closed = append(closed, c) + } + } + sort.Slice(closed, func(a, b int) bool { return closed[a].start < closed[b].start }) + var out [][]byte + for _, c := range closed { + out = append(out, c.env) + } + for _, p := range x.h.tiles.open { + out = append(out, p.env) + } + return out, nil +} + +// holdsTilesLocked says whether the roster in force lets the service hold +// tiles: a copy it serves, with ftw.history.read in the grant. The caller +// holds h.mu. +func (h *home) holdsTilesLocked() bool { + r := h.roster + return r != nil && slices.Contains(r.Grant.Purposes, protocol.CloudPurposeServe) && + slices.Contains(r.Grant.Scopes, protocol.ScopeHistoryRead) +} + +// pruneTilesLocked forgets the tiles that ended 48 hours or more before now. +// The caller holds h.mu. +func (x *Handle) pruneTilesLocked(now int64) { x.s.pruneTiles(x.h, now) } + +func (s *Store) pruneTiles(h *home, now int64) { + ts := &h.tiles + for id, c := range ts.closed { + if c.end() <= now-cloudwire.TileKeepMs { + s.addTileBytes(h, -int64(len(c.env))) + delete(ts.closed, id) + } + } + if len(ts.open) > 0 && ts.open[0].end() <= now-cloudwire.TileKeepMs { + s.dropOpen(h) + } +} + +// dropTiles forgets every tile of the home. The caller holds h.mu. +func (s *Store) dropTiles(h *home) { + s.addTileBytes(h, -h.tiles.bytes) + h.tiles = tileSet{} +} + +func (x *Handle) dropOpenLocked() { x.s.dropOpen(x.h) } + +func (s *Store) dropOpen(h *home) { + for _, p := range h.tiles.open { + s.addTileBytes(h, -int64(len(p.env))) + } + h.tiles.open = nil +} + +func (x *Handle) tileBytesLocked(n int64) { x.s.addTileBytes(x.h, n) } + +// addTileBytes counts n bytes of tiles into the home and the store. The +// caller holds h.mu. +func (s *Store) addTileBytes(h *home, n int64) { + h.tiles.bytes += n + s.tileBytes.Add(n) + tileBytesHeld.Add(n) +} + +// TileBytes is how many bytes of tiles the store holds, all homes. +func (s *Store) TileBytes() int64 { return s.tileBytes.Load() } + +// sweepTiles drops a home's tiles past 48 hours, and all of them once the +// roster has run out: no phone may read them then, and a revoke made while +// the box was away cannot reach the service. Sweep calls it for every home, +// the box online or not. The caller holds h.mu. +func (s *Store) sweepTiles(h *home, now time.Time) { + if r := h.roster; r == nil || now.After(h.rosterAt.Add(time.Duration(r.TTLS)*time.Second)) { + s.dropTiles(h) + return + } + s.pruneTiles(h, now.UnixMilli()) +} diff --git a/internal/store/tiles_test.go b/internal/store/tiles_test.go new file mode 100644 index 0000000..fad77e7 --- /dev/null +++ b/internal/store/tiles_test.go @@ -0,0 +1,456 @@ +package store + +import ( + "bytes" + "errors" + "testing" + "time" + + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// The tiles of these tests: the one that starts 2026-10-12 12:00 UTC fills, +// and the clock reads 17:16. +const ( + tStart = int64(1_791_806_400_000) + tSpan = int64(43_200_000) +) + +var tileNow = time.UnixMilli(tStart + (5*60+16)*60_000) + +// windows has a value for every bucket up to the clock. +func windows() map[string][]protocol.HistorySample { + out := map[string][]protocol.HistorySample{} + for at := tStart - 5*tSpan; at < tileNow.UnixMilli(); at += 300_000 { + for _, name := range cloudwire.TileSeries { + out[name] = append(out[name], protocol.HistorySample{StartMs: at, W: float64(at / 300_000 % 977)}) + } + } + return out +} + +func closedTile(start int64) cloudwire.Tiles { + c := protocol.PackTile(cloudwire.TileRes, 1, start, cloudwire.TileSeries, windows(), start+tSpan) + return cloudwire.Tiles{TileID: c.TileID, Etag: c.Etag, Res: c.Res, StartMs: c.StartMs, StepMs: c.StepMs, + Series: c.Series, Data: c.Data} +} + +// piece is points [from, to) of the tile at start, as the box cuts them. +func piece(start int64, from, to int) cloudwire.Tiles { + c := protocol.PackTile(cloudwire.TileRes, 1, start, cloudwire.TileSeries, windows(), tileNow.UnixMilli()) + return cloudwire.Piece(c, from, to) +} + +// env stands for the box's signed envelope; the store holds bytes and reads +// what the uplink checked. +func env(b cloudwire.Tiles, seq uint64) []byte { + return []byte(b.TileID + "|" + b.Etag + "|" + time.UnixMilli(b.StartMs).UTC().Format("02T15:04") + "|" + string(rune('a'+seq%26))) +} + +func tileHome(t *testing.T, grant ...string) (*Store, *Handle, *clock) { + t.Helper() + c := &clock{t: tileNow} + s := New(c.now) + box, err := s.For(Principal{Kind: Box, Home: "h1"}) + if err != nil { + t.Fatal(err) + } + box.Connected() + if len(grant) == 0 { + grant = []string{protocol.ScopeLiveRead, protocol.ScopeHistoryRead} + } + box.SetRoster(roster(1, grant...)) + return s, box, c +} + +func put(t *testing.T, box *Handle, b cloudwire.Tiles, seq uint64) []byte { + t.Helper() + if err := b.Check(); err != nil { + t.Fatal(err) + } + e := env(b, seq) + if !kept(box, e, b, seq) { + t.Fatalf("%s at %d not held", b.TileID, b.StartMs) + } + return e +} + +// kept puts a tile sent as seq, and says whether the store kept it. +func kept(box *Handle, e []byte, b cloudwire.Tiles, seq uint64) bool { + b.Seq = seq + k, _ := box.PutTile(e, b) + return k +} + +func owner(t *testing.T, s *Store) *Handle { + t.Helper() + x, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + if err != nil { + t.Fatal(err) + } + return x +} + +func TestTheStoreAnswersWhatThePhoneLacksClosedFirstThenPiecesInOrder(t *testing.T) { + s, box, _ := tileHome(t) + older := put(t, box, closedTile(tStart-2*tSpan), 1) + newer := put(t, box, closedTile(tStart-tSpan), 2) + whole := put(t, box, piece(tStart, 0, 144), 3) + app1 := put(t, box, piece(tStart, 62, 63), 4) + app2 := put(t, box, piece(tStart, 63, 64), 5) + + got, err := owner(t, s).Tiles(nil) + if err != nil { + t.Fatal(err) + } + if want := [][]byte{older, newer, whole, app1, app2}; !equalEnvs(got, want) { + t.Fatalf("tiles.get {} answered %q", got) + } + // What the phone holds at the same etag stays home; a stale etag does + // not. + have := box.TilesHave() + if len(have) != 2 || have[closedTile(tStart-tSpan).TileID] != closedTile(tStart-tSpan).Etag { + t.Fatalf("have %v", have) + } + stale := map[string]string{closedTile(tStart - tSpan).TileID: closedTile(tStart - tSpan).Etag, + closedTile(tStart - 2*tSpan).TileID: "00000000"} + got, _ = owner(t, s).Tiles(stale) + if want := [][]byte{older, whole, app1, app2}; !equalEnvs(got, want) { + t.Fatalf("tiles.get with have answered %q", got) + } +} + +func TestAWholePieceStartsTheTileOverAndTheClosedTileEndsIt(t *testing.T) { + s, box, _ := tileHome(t) + put(t, box, piece(tStart, 0, 144), 1) + put(t, box, piece(tStart, 62, 63), 2) + whole := put(t, box, piece(tStart, 0, 144), 3) + got, _ := owner(t, s).Tiles(nil) + if !equalEnvs(got, [][]byte{whole}) { + t.Fatalf("after a second whole piece: %q", got) + } + // The tile closes: its pieces go, and a late piece of it is refused. + closed := closedTile(tStart) + ce := env(closed, 4) + if !kept(box, ce, closed, 4) { + t.Fatal("closed tile not held") + } + if late := piece(tStart, 63, 64); kept(box, env(late, 5), late, 5) { + t.Fatal("a piece of a closed tile was held") + } + got, _ = owner(t, s).Tiles(nil) + if !equalEnvs(got, [][]byte{ce}) { + t.Fatalf("after the tile closed: %q", got) + } + // The next tile's first piece starts a new trailing tile. + next := closedTile(tStart + tSpan) + next.Partial = true + if !kept(box, env(next, 6), next, 6) { + t.Fatal("the next tile's piece was refused") + } + if got, _ = owner(t, s).Tiles(nil); len(got) != 2 { + t.Fatalf("%d tiles", len(got)) + } +} + +func TestTilesAreBoundedPerHomeAndInAll(t *testing.T) { + s, box, _ := tileHome(t) + // The tile that ended 12:00 two days before ended more than 48 hours ago. + if old := closedTile(tStart - 5*tSpan); kept(box, env(old, 1), old, 1) { + t.Fatal("a tile that ended more than 48 hours ago was held") + } + for i := int64(4); i >= 1; i-- { + put(t, box, closedTile(tStart-i*tSpan), uint64(10-i)) + } + // A whole piece and 15 appends; past them appends are refused, and what + // is held still lays back into the tile, its first bucket included. + put(t, box, piece(tStart, 0, 144), 20) + for i := 1; i < cloudwire.TilePiecesMax; i++ { + put(t, box, piece(tStart, 50+i, 51+i), uint64(20+i)) + } + for i := 0; i < 4; i++ { + if late := piece(tStart, 70+i, 71+i); kept(box, env(late, uint64(40+i)), late, uint64(40+i)) { + t.Fatal("an append past the pieces a tile may have was held") + } + } + got, _ := owner(t, s).Tiles(nil) + if len(box.TilesHave()) != 4 || len(got)-4 != cloudwire.TilePiecesMax || !bytes.Equal(got[4], env(piece(tStart, 0, 144), 20)) { + t.Fatalf("held %d closed and %d pieces", len(box.TilesHave()), len(got)-len(box.TilesHave())) + } + // The next whole piece starts the tile over. + put(t, box, piece(tStart, 0, 144), 50) + if got, _ = owner(t, s).Tiles(nil); len(got) != 5 { + t.Fatalf("after a new whole piece, %d tiles", len(got)) + } + + // All homes together: past the cap a tile that adds bytes is dropped, + // and one that frees more than it takes is held. + for i := 1; i <= 3; i++ { + put(t, box, piece(tStart, 50+i, 51+i), uint64(50+i)) + } + s.tileBytesMax = s.TileBytes() + 10 + extra := closedTile(tStart - tSpan) + extra.Data[0] ^= 1 + extra.Etag = protocol.EtagOf(extra.Data) + if kept(box, bytes.Repeat([]byte{1}, 64), extra, 60) { + t.Fatal("a tile past the store's cap was held") + } + before := s.TileBytes() + whole := piece(tStart, 0, 144) + if !kept(box, bytes.Repeat([]byte{2}, 40), whole, 61) || s.TileBytes() >= before { + t.Fatal("a whole piece that frees bytes was refused at the cap") + } +} + +func TestTilesOutliveTheHourOfSilenceAndGoAfter48Hours(t *testing.T) { + s, box, c := tileHome(t) + _ = box.Put(cloudwire.MsgLive, []byte("live"), 1, true) + put(t, box, closedTile(tStart-tSpan), 2) + put(t, box, piece(tStart, 0, 144), 3) + box.Disconnected(1) + + c.t = tileNow.Add(2 * time.Hour) + s.Sweep() + snap, _ := owner(t, s).Snapshot() + got, _ := owner(t, s).Tiles(nil) + if len(snap.Envelopes) != 0 || len(got) != 2 { + t.Fatalf("after two silent hours: %d readings, %d tiles", len(snap.Envelopes), len(got)) + } + // The closed tile ended at 12:00; two days later it is gone, and the + // trailing one, which ends at midnight, stays until then. + c.t = time.UnixMilli(tStart + cloudwire.TileKeepMs + time.Minute.Milliseconds()) + // The box sent a roster since, or no phone would be let in at all. + box.SetRoster(roster(2, protocol.ScopeLiveRead, protocol.ScopeHistoryRead)) + s.Sweep() + got, _ = owner(t, s).Tiles(nil) + if len(got) != 1 || len(box.TilesHave()) != 0 { + t.Fatalf("48 hours after the closed tile ended: %d tiles", len(got)) + } + // A tile from the future is refused. + future := piece(tStart, 0, 1) + c.t = time.UnixMilli(tStart - 2*tSpan) + if kept(box, env(future, 9), future, 9) { + t.Fatal("a tile more than a tile ahead of the clock was held") + } +} + +func TestRevokePurgeOrAGrantWithoutHistoryDropsTheTiles(t *testing.T) { + cases := map[string]func(*Store, *Handle){ + "copy withdrawn": func(_ *Store, box *Handle) { + r := roster(2) + r.Grant.Purposes = nil + box.SetRoster(r) + }, + "history taken out": func(_ *Store, box *Handle) { box.SetRoster(roster(2, protocol.ScopeLiveRead)) }, + "copy dropped": func(_ *Store, box *Handle) { box.DropCopy() }, + "home purged": func(_ *Store, box *Handle) { box.Delete() }, + "home forgotten": func(s *Store, _ *Handle) { s.Forget("h1") }, + } + for name, end := range cases { + t.Run(name, func(t *testing.T) { + s, box, _ := tileHome(t) + put(t, box, closedTile(tStart-tSpan), 1) + put(t, box, piece(tStart, 0, 144), 2) + if s.TileBytes() == 0 { + t.Fatal("nothing counted") + } + end(s, box) + if s.TileBytes() != 0 || len(box.TilesHave()) != 0 { + t.Fatalf("%d bytes of tiles left", s.TileBytes()) + } + if x, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}); err == nil { + if got, _ := x.Tiles(nil); len(got) != 0 { + t.Fatalf("%d tiles still served", len(got)) + } + } + // And none is taken while it lasts. + if b := closedTile(tStart - tSpan); name == "history taken out" && kept(box, env(b, 3), b, 3) { + t.Fatal("a tile was held under a grant without ftw.history.read") + } + }) + } +} + +func TestOnlyADeviceWithTheHistoryScopeReadsTiles(t *testing.T) { + s, box, _ := tileHome(t) + put(t, box, closedTile(tStart-tSpan), 1) + viewer, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: viewerKey}) + if err != nil { + t.Fatal(err) + } + if got, err := viewer.Tiles(nil); err != nil || len(got) != 0 { + t.Fatalf("a viewer without ftw.history.read read %d tiles", len(got)) + } + if _, err := box.Tiles(nil); !errors.Is(err, ErrDenied) { + t.Fatal("the box's handle reads tiles") + } + // The snapshot carries none: a phone asks. + snap, _ := owner(t, s).Snapshot() + if len(snap.Envelopes) != 0 { + t.Fatalf("the snapshot carries %d envelopes", len(snap.Envelopes)) + } + // Nor does a widget's token. + _ = box.Put(cloudwire.MsgLive, []byte("live"), 2, true) + if !kept(box, env(piece(tStart, 0, 144), 3), piece(tStart, 0, 144), 3) { + t.Fatal("piece not held") + } + tok, err := s.For(Principal{Kind: Token, Home: "h1", DeviceID: "owner"}) + if err != nil { + t.Fatal(err) + } + now, _ := tok.Now() + if !equalEnvs(now.Envelopes, [][]byte{[]byte("live")}) { + t.Fatalf("a token read %q", now.Envelopes) + } +} + +// A phone reading the home gets each new tile as it comes, if it may read +// tiles. +func TestANewTileGoesToThePhonesThatMayReadIt(t *testing.T) { + s, box, _ := tileHome(t) + o, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + if err != nil { + t.Fatal(err) + } + _, os, err := o.Open() + if err != nil { + t.Fatal(err) + } + defer os.Close() + v, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: viewerKey}) + if err != nil { + t.Fatal(err) + } + _, vs, err := v.Open() + if err != nil { + t.Fatal(err) + } + defer vs.Close() + e := put(t, box, piece(tStart, 0, 144), 1) + select { + case out := <-os.C(): + if !bytes.Equal(out.Env, e) { + t.Fatal("the owner got other bytes") + } + case <-time.After(time.Second): + t.Fatal("the owner got no tile") + } + select { + case out := <-vs.C(): + t.Fatalf("the viewer got %q", out.Env) + case <-time.After(50 * time.Millisecond): + } +} + +func equalEnvs(a, b [][]byte) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if !bytes.Equal(a[i], b[i]) { + return false + } + } + return true +} + +// An agent reads history from the box, never from these tiles: its handle +// is refused, and its session gets none as they come. +func TestAnAgentGetsNoTiles(t *testing.T) { + s, box, c := tileHome(t) + r := rosterWithAgent(2, c.t.Add(time.Hour)) + r.Grant.Scopes = []string{protocol.ScopeLiveRead, protocol.ScopeHistoryRead} + r.Devices[2].Agent.Scopes = []string{protocol.ScopeHistoryRead} + box.SetRoster(r) + put(t, box, closedTile(tStart-tSpan), 1) + agent, err := s.For(Principal{Kind: Agent, Home: "h1", DeviceKey: agentKey}) + if err != nil { + t.Fatal(err) + } + if _, err := agent.Tiles(nil); !errors.Is(err, ErrDenied) { + t.Fatalf("an agent read tiles: %v", err) + } + _, sess, err := agent.Open() + if err != nil { + t.Fatal(err) + } + defer sess.Close() + put(t, box, piece(tStart, 0, 144), 2) + select { + case out := <-sess.C(): + t.Fatalf("an agent session got %q", out.Env) + case <-time.After(50 * time.Millisecond): + } +} + +// The box sends a closed tile again when its ledger changed: the later one +// replaces the first, and an older one that comes late replaces nothing. +// A piece that comes after one the box sent later is dropped too. +func TestAClosedTileIsReplacedOnlyByOneSentLater(t *testing.T) { + s, box, _ := tileHome(t) + first := closedTile(tStart - tSpan) + fixed := first + fixed.Data = bytes.Clone(first.Data) + fixed.Data[0] ^= 1 + fixed.Etag = protocol.EtagOf(fixed.Data) + put(t, box, first, 10) + put(t, box, fixed, 20) + if box.TilesHave()[first.TileID] != fixed.Etag { + t.Fatal("a later closed tile did not replace the first") + } + if kept(box, env(first, 15), first, 15) || box.TilesHave()[first.TileID] != fixed.Etag { + t.Fatal("an older closed tile replaced a later one") + } + put(t, box, piece(tStart, 0, 144), 30) + put(t, box, piece(tStart, 62, 63), 32) + if late := piece(tStart, 61, 62); kept(box, env(late, 31), late, 31) { + t.Fatal("a piece sent before the one held was laid after it") + } + if got, _ := owner(t, s).Tiles(map[string]string{first.TileID: fixed.Etag}); len(got) != 2 { + t.Fatalf("%d pieces", len(got)) + } +} + +// One home holds at most TileBytesPerHome, whatever room the store has. +func TestAHomeHoldsAtMostItsShareOfTileBytes(t *testing.T) { + _, box, _ := tileHome(t) + big := func(n int) []byte { return bytes.Repeat([]byte{byte(n)}, 4000) } + for i := int64(1); i <= 4; i++ { + if !kept(box, big(int(i)), closedTile(tStart-i*tSpan), uint64(i)) { + t.Fatalf("closed tile %d not held", i) + } + } + held := 0 + for i := 0; i < cloudwire.TilePiecesMax; i++ { + p := piece(tStart, 0, 144) + if i > 0 { + p = piece(tStart, 50+i, 51+i) + } + p.Seq = uint64(10 + i) + k, full := box.PutTile(big(10+i), p) + if k { + held++ + continue + } + if !full { + t.Fatal("a tile past the home's share was refused for another reason") + } + break + } + if total := 4*4000 + held*4000; total > TileBytesPerHome || held == cloudwire.TilePiecesMax { + t.Fatalf("held %d pieces, %d bytes", held, total) + } +} + +// Once the roster runs out no phone may read the tiles, so they go. +func TestTilesGoWhenTheRosterRunsOut(t *testing.T) { + s, box, c := tileHome(t) + put(t, box, closedTile(tStart-tSpan), 1) + box.Disconnected(1) + c.t = c.t.Add(25 * time.Hour) + s.Sweep() + if s.TileBytes() != 0 || len(box.TilesHave()) != 0 { + t.Fatalf("%d bytes of tiles after the roster ran out", s.TileBytes()) + } +} diff --git a/internal/uplink/tiles_test.go b/internal/uplink/tiles_test.go new file mode 100644 index 0000000..51eafc7 --- /dev/null +++ b/internal/uplink/tiles_test.go @@ -0,0 +1,315 @@ +package uplink + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "encoding/hex" + "encoding/json" + "io" + "log/slog" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// tileVectors is srcfl/ftw's cloud-tile-vectors.json, from the go/protocol +// this build pins. +type tileVectors struct { + BoxKey struct { + D string `json:"d"` + PK string `json:"pk"` + Home string `json:"home"` + } `json:"boxKey"` + Tiles []struct { + Name string `json:"name"` + TileID string `json:"tileId"` + Etag string `json:"etag"` + Partial bool `json:"partial"` + Envelope string `json:"envelope"` + Frame string `json:"frame"` + } `json:"tiles"` + Forged struct { + Frame string `json:"frame"` + } `json:"forged"` +} + +func readTileVectors(t *testing.T) tileVectors { + t.Helper() + out, err := exec.Command("go", "list", "-m", "-f", "{{.Dir}}", "github.com/srcfl/ftw/go/protocol").Output() + if err != nil { + t.Fatalf("go list: %v", err) + } + raw, err := os.ReadFile(filepath.Join(strings.TrimSpace(string(out)), "cloudwire", "testdata", "cloud-tile-vectors.json")) + if err != nil { + t.Fatal(err) + } + var v tileVectors + if err := json.Unmarshal(raw, &v); err != nil { + t.Fatal(err) + } + return v +} + +func unhex(t *testing.T, s string) []byte { + t.Helper() + b, err := hex.DecodeString(s) + if err != nil { + t.Fatal(err) + } + return b +} + +// vectorClock is 2026-10-12 17:16 UTC, when the vectors' last append had +// gone. +var vectorClock = time.UnixMilli(1_791_806_400_000 + (5*60+16)*60_000) + +// newServiceAt is newService with the store on a clock of its own. +func newServiceAt(t *testing.T, now func() time.Time) *service { + t.Helper() + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + st := store.New(now) + up := New(st, reg, key, slog.New(slog.NewTextHandler(io.Discard, nil))) + srv := httptest.NewUnstartedServer(up) + srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} + srv.StartTLS() + t.Cleanup(srv.Close) + return &service{srv: srv, up: up, store: st, reg: reg, key: key} +} + +// vectorBox is the box that signed the vectors. +func vectorBox(t *testing.T, v tileVectors) *box { + t.Helper() + key, err := ecdsa.ParseRawPrivateKey(elliptic.P256(), unhex(t, v.BoxKey.D)) + if err != nil { + t.Fatal(err) + } + pk, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) + if cloudwire.HomeID(pk) != v.BoxKey.Home { + t.Fatal("the vectors' home is not their key's") + } + return &box{t: t, key: key, pk: pk, home: v.BoxKey.Home, boot: bytes.Repeat([]byte{0xb0}, 16), seq: 100} +} + +func (b *box) frame(raw []byte) { + b.t.Helper() + if err := b.conn.WriteMessage(websocket.BinaryMessage, raw); err != nil { + b.t.Fatal(err) + } +} + +func ownerTiles(t *testing.T, s *service, home string) [][]byte { + t.Helper() + dev, err := s.store.For(store.Principal{Kind: store.Device, Home: home, DeviceKey: ownerKey}) + if err != nil { + return nil + } + got, _ := dev.Tiles(nil) + return got +} + +// The service holds the tiles the box signed, exactly, names the closed one +// in hello_ok when the box comes back, and serves them after the box left. +func TestTheServiceHoldsTheBoxsTilesAndNamesTheClosedOnesAtReconnect(t *testing.T) { + v := readTileVectors(t) + s := newServiceAt(t, func() time.Time { return vectorClock }) + b := vectorBox(t, v) + ok, err := b.connect(s, nil) + if err != nil { + t.Fatal(err) + } + if len(ok.Have) != 0 { + t.Fatalf("a new home's have is %v", ok.Have) + } + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead, protocol.ScopeHistoryRead)) + var want [][]byte + for _, tv := range v.Tiles { + b.frame(unhex(t, tv.Frame)) + want = append(want, unhex(t, tv.Envelope)) + } + wait(t, "four tiles", func() bool { return len(ownerTiles(t, s, b.home)) == 4 }) + for i, got := range ownerTiles(t, s, b.home) { + if !bytes.Equal(got, want[i]) { + t.Fatalf("tile %d is not the box's bytes", i) + } + } + + _ = b.conn.Close() + wait(t, "the box gone", func() bool { return s.up.Boxes() == 0 }) + ok, err = b.connect(s, nil) + if err != nil { + t.Fatal(err) + } + if len(ok.Have) != 1 || ok.Have[v.Tiles[0].TileID] != v.Tiles[0].Etag { + t.Fatalf("hello_ok.have %v, want the closed tile alone", ok.Have) + } +} + +// A tile the box did not sign closes the uplink, as any such message does, +// and is never held. +func TestAForgedTileClosesTheUplinkAndIsNeverHeld(t *testing.T) { + v := readTileVectors(t) + s := newServiceAt(t, func() time.Time { return vectorClock }) + b := vectorBox(t, v) + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead, protocol.ScopeHistoryRead)) + b.frame(unhex(t, v.Forged.Frame)) + if !b.closed() { + t.Fatal("the service kept the uplink after a forged tile") + } + if got := ownerTiles(t, s, b.home); len(got) != 0 { + t.Fatalf("%d tiles held", len(got)) + } + // Nor one signed by another key. + b2 := vectorBox(t, v) + if _, err := b2.connect(s, nil); err != nil { + t.Fatal(err) + } + other, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + var body cloudwire.Tiles + env, _ := cloudwire.DecodeEnvelope(unhex(t, v.Tiles[1].Envelope)) + _ = cloudwire.Unmarshal(env.B, &body) + b2.sendAs(other, cloudwire.MsgTiles, body) + if !b2.closed() || len(ownerTiles(t, s, b2.home)) != 0 { + t.Fatal("a tile under another key was held") + } +} + +// A tile the box signed that is not one of the three shapes is dropped and +// counted; the uplink and the rest of the copy go on. +func TestATileOfNoShapeIsDroppedAndTheUplinkStays(t *testing.T) { + v := readTileVectors(t) + s := newServiceAt(t, func() time.Time { return vectorClock }) + b := vectorBox(t, v) + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead, protocol.ScopeHistoryRead)) + env, _ := cloudwire.DecodeEnvelope(unhex(t, v.Tiles[2].Envelope)) + var body cloudwire.Tiles + _ = cloudwire.Unmarshal(env.B, &body) + body.Header = b.header() + body.Etag = "00000000" + b.send(cloudwire.MsgTiles, body) + live := b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 1200}, + Sources: map[string]cloudwire.Source{"x": {Kind: "driver", State: "live", StaleAfterMs: 10_000}}, SrcRev: 1}) + wait(t, "the live after it", func() bool { + snap := snapshot(t, s, b.home) + return len(snap) == 1 && bytes.Equal(snap[0], live) + }) + if got := ownerTiles(t, s, b.home); len(got) != 0 { + t.Fatalf("a tile of no shape was held") + } +} + +// Without ftw.history.read in the grant the service holds no tiles. +func TestTilesNeedTheHistoryScopeInTheGrant(t *testing.T) { + v := readTileVectors(t) + s := newServiceAt(t, func() time.Time { return vectorClock }) + b := vectorBox(t, v) + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead)) + for _, tv := range v.Tiles { + b.frame(unhex(t, tv.Frame)) + } + live := b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 1}, SrcRev: 1}) + wait(t, "the live after the tiles", func() bool { + snap := snapshot(t, s, b.home) + return len(snap) == 1 && bytes.Equal(snap[0], live) + }) + dev, err := s.store.For(store.Principal{Kind: store.Box, Home: b.home}) + if err != nil { + t.Fatal(err) + } + if have := dev.TilesHave(); len(have) != 0 || s.store.TileBytes() != 0 { + t.Fatalf("held %v under a grant without history", have) + } +} + +// A tiles envelope over cloudwire.TileEnvelopeMax is dropped, signed or not, +// and the uplink goes on. +func TestAnOversizedTileIsDroppedAndTheUplinkStays(t *testing.T) { + v := readTileVectors(t) + s := newServiceAt(t, func() time.Time { return vectorClock }) + b := vectorBox(t, v) + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead, protocol.ScopeHistoryRead)) + env, _ := cloudwire.DecodeEnvelope(unhex(t, v.Tiles[0].Envelope)) + var body cloudwire.Tiles + _ = cloudwire.Unmarshal(env.B, &body) + // The closed tile with a key the service ignores, to pad it. + padded := struct { + cloudwire.Tiles + Pad []byte `cbor:"pad"` + }{body, bytes.Repeat([]byte{1}, cloudwire.TileEnvelopeMax)} + padded.Header = b.header() + b.send(cloudwire.MsgTiles, padded) + live := b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 1}, SrcRev: 1}) + wait(t, "the live after it", func() bool { + snap := snapshot(t, s, b.home) + return len(snap) == 1 && bytes.Equal(snap[0], live) + }) + if got := ownerTiles(t, s, b.home); len(got) != 0 { + t.Fatal("an oversized tile was held") + } +} + +// History taken out of the grant and put back while the box stays +// connected: the service dropped the tiles, so it asks the box for them +// again, once. +func TestHistoryBackInTheGrantAsksTheBoxForItsTiles(t *testing.T) { + v := readTileVectors(t) + s := newServiceAt(t, func() time.Time { return vectorClock }) + b := vectorBox(t, v) + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + roster := func(seq uint64, scopes ...string) { + r := b.roster(scopes...) + r.RosterSeq = seq + b.send(cloudwire.MsgRoster, r) + } + roster(1, protocol.ScopeLiveRead, protocol.ScopeHistoryRead) + b.frame(unhex(t, v.Tiles[0].Frame)) + wait(t, "the tile", func() bool { return len(ownerTiles(t, s, b.home)) == 1 }) + + roster(2, protocol.ScopeLiveRead) + roster(3, protocol.ScopeLiveRead, protocol.ScopeHistoryRead) + b.send(cloudwire.MsgHB, cloudwire.HB{Header: b.header(), Clock: cloudwire.ClockNTP}) + env := b.read() + var r cloudwire.Resync + if env.T != cloudwire.MsgResync || cloudwire.Unmarshal(env.B, &r) != nil || len(r.What) != 1 || r.What[0] != cloudwire.ResyncTiles { + t.Fatalf("the service sent %q, want resync for tiles", env.T) + } + // Not again within five minutes. + roster(4, protocol.ScopeLiveRead) + roster(5, protocol.ScopeLiveRead, protocol.ScopeHistoryRead) + b.send(cloudwire.MsgHB, cloudwire.HB{Header: b.header(), Clock: cloudwire.ClockNTP}) + _ = b.conn.SetReadDeadline(time.Now().Add(300 * time.Millisecond)) + if _, _, err := b.conn.ReadMessage(); err == nil { + t.Fatal("a second resync within five minutes") + } +} diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index 6e63fc1..2e78b88 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -91,6 +91,8 @@ var ( dropTunUp = metrics.Drop("tun_rate_up") dropTunDown = metrics.Drop("tun_rate_down") dropUnknown = metrics.Drop("box_unknown_type") + dropBadTile = metrics.Drop("box_tile_of_no_shape") + dropBigTile = metrics.Drop("box_tile_too_large") dropConnect = metrics.Drop("box_connect_rate") boxesFollow = metrics.Default.Gauge("ftwcloud_boxes_following_shards", "Connected boxes that list shard in hello.caps.") ) @@ -458,8 +460,20 @@ type boxConn struct { newest map[string]uint64 tableRev uint64 statsAllowed bool // current connection accepted the full statistics roster + + // rosters counts the rosters taken on this connection. tilesLost says + // tiles were dropped for lack of memory, or history came back to the + // grant, so the box is asked for its tiles again; tilesAskedAt is when + // it last was. + rosters int + tilesLost bool + tilesAskedAt time.Time } +// tilesResyncEvery is how often the service asks a box for its tiles again, +// as often as the box honours a resync. +const tilesResyncEvery = 5 * time.Minute + // bucket is a token bucket. type bucket struct { tokens, burst, rate float64 @@ -575,7 +589,11 @@ func (c *boxConn) run() error { if c.u.Statistics { statsPolicy = cloudwire.StatsPolicy } - if err := c.send(helloOK{StatsPolicy: statsPolicy, BoxNonce: hello.BoxNonce, EKM: c.ekm, Have: map[string]string{}, RosterSeq: c.rosterSeq()}); err != nil { + have := map[string]string{} + if c.handle != nil { + have = c.handle.TilesHave() + } + if err := c.send(helloOK{StatsPolicy: statsPolicy, BoxNonce: hello.BoxNonce, EKM: c.ekm, Have: have, RosterSeq: c.rosterSeq()}); err != nil { return err } boxHandshake.Since(c.started) @@ -653,6 +671,9 @@ func (c *boxConn) run() error { if err != nil || stop { return err } + if err := c.askTiles(); err != nil { + return err + } } } @@ -814,6 +835,9 @@ func (c *boxConn) apply(env cloudwire.Envelope) (stop bool, err error) { // The readable copy is off: the service holds no readings. return false, nil } + if env.T == cloudwire.MsgTiles { + return false, c.tile(env) + } rev, table := uint64(0), false if env.T == cloudwire.MsgLive { rev, table = tableOf(env.B) @@ -849,6 +873,42 @@ func (c *boxConn) apply(env cloudwire.Envelope) (stop bool, err error) { return false, nil } +// tile holds one tile the box signed, if it has one of the three shapes +// (cloudwire.Tiles.Check) and fits cloudwire.TileEnvelopeMax; anything else +// is dropped and counted, and the connection goes on. A tile dropped for +// lack of memory is asked for again once there is room. +func (c *boxConn) tile(env cloudwire.Envelope) error { + raw, err := env.Encode() + if err != nil { + return err + } + if len(raw) > cloudwire.TileEnvelopeMax { + dropBigTile.Inc() + return nil + } + var t cloudwire.Tiles + if cloudwire.Unmarshal(env.B, &t) != nil || t.Check() != nil { + dropBadTile.Inc() + return nil + } + if _, full := c.handle.PutTile(raw, t); full { + c.tilesLost = true + } + return nil +} + +// askTiles sends the box resync {what: [tiles]} when tiles are owed and +// there is room for them, at most once per tilesResyncEvery. The box sends +// what the service lacks. +func (c *boxConn) askTiles() error { + if !c.tilesLost || c.purged || c.handle == nil || !c.handle.HoldsTiles() || !c.handle.TileRoom() || + c.u.now().Sub(c.tilesAskedAt) < tilesResyncEvery { + return nil + } + c.tilesLost, c.tilesAskedAt = false, c.u.now() + return c.send(resync{What: []string{cloudwire.ResyncTiles}}) +} + // newer reports whether a message of a newest-wins class is newer than any // of its class on this connection, and remembers it. Over a WebSocket every // one is; over QUIC a datagram can arrive after a later message. @@ -959,9 +1019,17 @@ func (c *boxConn) roster(env cloudwire.Envelope) error { c.u.store.Forget(c.home) return fmt.Errorf("%w: updated Cloud terms including statistics are required", errProtocol) } + hadTiles := c.handle.HoldsTiles() if !c.handle.SetRoster(&r) { return nil } + // History back in the grant while the box stays connected: the service + // dropped its tiles when it went, and the box sends them again only when + // asked. The first roster of a connection needs no asking. + c.rosters++ + if c.rosters > 1 && !hadTiles && c.handle.HoldsTiles() { + c.tilesLost = true + } if c.u.Statistics || c.u.StatisticsStore { grant := g if !allowed {