From c54bd11a7e6c095caf4e9392a10c92581e002f1f Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 16:10:59 +0200 Subject: [PATCH 1/6] feat: hold 48 hours of history tiles and answer tiles.get (C6) A box with ftw.history.read in its grant sends tiles. The service checks each against the protocol's three shapes and keeps it in memory as the box signed it: five closed tiles and 16 pieces of the filling tile per home, 256 MiB in all, each until 48 hours after it ends, box online or not. hello_ok.have names the closed tiles held. A phone that may read history asks with tiles.get {have}, gets what it lacks and tiles.end, and gets each new tile as it comes. A revoke, a grant without history, a purge or the 90-day expiry deletes them. Pins go/protocol at srcfl/ftw 7dedcbf4 (feat/cloud-c6-tiles). Co-Authored-By: Claude Opus 5.5 --- README.md | 20 +- go.mod | 2 +- go.sum | 3 + internal/appsession/appsession.go | 27 +++ internal/appsession/tiles_test.go | 197 ++++++++++++++++++ internal/store/store.go | 19 +- internal/store/tiles.go | 241 ++++++++++++++++++++++ internal/store/tiles_test.go | 328 ++++++++++++++++++++++++++++++ internal/uplink/tiles_test.go | 248 ++++++++++++++++++++++ internal/uplink/uplink.go | 27 ++- 10 files changed, 1106 insertions(+), 6 deletions(-) create mode 100644 internal/appsession/tiles_test.go create mode 100644 internal/store/tiles.go create mode 100644 internal/store/tiles_test.go create mode 100644 internal/uplink/tiles_test.go diff --git a/README.md b/README.md index 102eea0..8516d5e 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 48 hours, 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,23 @@ 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 holds it in memory as the box + signed it: per home at most five closed tiles and 16 pieces of the filling + one, all homes together at most 256 MiB, each until 48 hours after it + ends. 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. 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, but a phone still needs + a roster in force to get in. A revoke, a grant without history, a purge + and the 90-day expiry delete them. Assistants read history from the box, + never from these tiles. - **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 +326,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..7b8705a 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-20261023135616-7dedcbf40d88 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..2ffa902 100644 --- a/go.sum +++ b/go.sum @@ -52,8 +52,11 @@ 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 v3.8.0-beta.1.0.20261023135616-7dedcbf40d88+incompatible h1:E7jKXntZtOPrn24ZbObaIiFPFgVPnmyUvxSZuigGTjQ= 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-20261023135616-7dedcbf40d88 h1:90QNPvhsMGZJcK9F3HxMT7+rwQei1K8MqUm4bJFhXWI= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023135616-7dedcbf40d88/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..7392424 --- /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 !h.box.PutTile(raw, b) { + 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/store/store.go b/internal/store/store.go index 87eb21a..7aff974 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) } @@ -609,7 +621,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 +636,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 +686,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..b71df38 --- /dev/null +++ b/internal/store/tiles.go @@ -0,0 +1,241 @@ +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. A home holds at +// most five closed tiles and 16 pieces, about 20 KB, so this is some 13,000 +// homes; a shard takes on no more than 4,000. Past it, new tiles are dropped +// and counted. +const TileBytesMax = 256 << 20 + +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") +) + +// 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 + seq uint64 + env []byte +} + +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. False says it was not held: +// the grant holds no tiles, it is older than 48 hours or from the future, a +// piece of a tile already closed, or memory for tiles is full. +func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { + if x.kind != Box { + return false + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + x.h.lastSeen = x.s.now() + if !x.h.holdsTilesLocked() { + return 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 held.end() <= now-cloudwire.TileKeepMs || held.start > now+protocol.ResolutionTileSpanMs[cloudwire.TileRes] { + dropTileOld.Inc() + return false + } + ts := &x.h.tiles + if x.s.tileBytes.Load()+int64(len(env)) > x.s.tileBytesMax { + dropTileFull.Inc() + return false + } + if !t.Partial { + if old, ok := ts.closed[held.id]; ok { + x.tileBytesLocked(-int64(len(old.env))) + } + if len(ts.open) > 0 && ts.open[0].id == held.id { + x.dropOpenLocked() + } + 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 { + if _, ok := ts.closed[held.id]; ok { + dropTileClosed.Inc() + return 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 + } + x.dropOpenLocked() + } + if t.Whole() { + x.dropOpenLocked() + } + ts.open = append(ts.open, held) + x.tileBytesLocked(int64(len(env))) + for len(ts.open) > cloudwire.TilePiecesMax { + x.tileBytesLocked(-int64(len(ts.open[0].env))) + ts.open = ts.open[1:] + } + } + at := stamp() + for sess := range x.h.sessions { + if sess.x.mayRead(cloudwire.MsgTiles) { + sess.deliver(env, at) + } + } + return true +} + +// 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. Sweep calls it for every +// home, the box online or not. +func (s *Store) sweepTiles(h *home, now time.Time) { 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..6e247dd --- /dev/null +++ b/internal/store/tiles_test.go @@ -0,0 +1,328 @@ +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) + } + b.Seq = seq + e := env(b, seq) + if !box.PutTile(e, b) { + t.Fatalf("%s at %d not held", b.TileID, b.StartMs) + } + return e +} + +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 !box.PutTile(ce, closed) { + t.Fatal("closed tile not held") + } + if late := piece(tStart, 63, 64); box.PutTile(env(late, 5), late) { + 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 !box.PutTile(env(next, 6), next) { + 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); box.PutTile(env(old, 1), old) { + 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)) + } + for i := 0; i < cloudwire.TilePiecesMax+4; i++ { + put(t, box, piece(tStart, i, i+1), uint64(20+i)) + } + got, _ := owner(t, s).Tiles(nil) + if len(box.TilesHave()) != 4 || len(got)-4 != cloudwire.TilePiecesMax { + t.Fatalf("held %d closed and %d pieces", len(box.TilesHave()), len(got)-len(box.TilesHave())) + } + + // All homes together: past the cap a new tile is dropped. + s.tileBytesMax = s.TileBytes() + 10 + extra := closedTile(tStart - tSpan) + extra.Data[0] ^= 1 + extra.Etag = protocol.EtagOf(extra.Data) + if box.PutTile(bytes.Repeat([]byte{1}, 64), extra) { + t.Fatal("a tile past the store's cap was held") + } +} + +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 box.PutTile(env(future, 9), future) { + 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" && box.PutTile(env(b, 3), b) { + 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 !box.PutTile(env(piece(tStart, 0, 144), 3), piece(tStart, 0, 144)) { + 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 +} diff --git a/internal/uplink/tiles_test.go b/internal/uplink/tiles_test.go new file mode 100644 index 0000000..82c6a70 --- /dev/null +++ b/internal/uplink/tiles_test.go @@ -0,0 +1,248 @@ +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) + } +} diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index 6e63fc1..7205ae6 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -91,6 +91,7 @@ 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") dropConnect = metrics.Drop("box_connect_rate") boxesFollow = metrics.Default.Gauge("ftwcloud_boxes_following_shards", "Connected boxes that list shard in hello.caps.") ) @@ -575,7 +576,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) @@ -814,6 +819,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 +857,23 @@ 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); anything else is dropped and counted, and the +// connection goes on. +func (c *boxConn) tile(env cloudwire.Envelope) error { + var t cloudwire.Tiles + if cloudwire.Unmarshal(env.B, &t) != nil || t.Check() != nil { + dropBadTile.Inc() + return nil + } + raw, err := env.Encode() + if err != nil { + return err + } + c.handle.PutTile(raw, t) + return nil +} + // 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. From 59c94106777920e677edc75417c6c7257cbcf596 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 16:22:19 +0200 Subject: [PATCH 2/6] test(dev): /dev/held lists the tiles a home holds So an end-to-end run can check what a real box sent. Co-Authored-By: Claude Opus 5.5 --- internal/store/inspect_dev.go | 8 ++++++++ 1 file changed, 8 insertions(+) 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 } From ae4746a5a46d7aa70b74bf4156f521a52bb6896d Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 16:32:11 +0200 Subject: [PATCH 3/6] feat(mcp): history_query says when an away box was last heard from An assistant reads history from the box only, never from FTW Cloud's copy, so every read stays one the box audits and a revoke stops. While the box is away, history_query now says so, with when FTW Cloud last heard from it, instead of only that it is offline. Co-Authored-By: Claude Opus 5.5 --- README.md | 3 ++- internal/mcp/agent.go | 11 ++++++++++ internal/mcp/history_test.go | 41 ++++++++++++++++++++++++++++++++++++ internal/mcp/tools.go | 39 ++++++++++++++++++++++++++++++++-- internal/store/store.go | 9 ++++++++ 5 files changed, 100 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 8516d5e..5d8a05c 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,8 @@ to it yet. so they outlive the hour after which readings go, but a phone still needs a roster in force to get in. A revoke, a grant without history, a purge and the 90-day expiry delete them. Assistants read history from the box, - never from these tiles. + 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 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/store.go b/internal/store/store.go index 7aff974..8a31a7b 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -554,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 { From 882a4b52d559000470b1a9237c341207410bd681 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 16:37:33 +0200 Subject: [PATCH 4/6] fix(store): keep a tile's whole piece at the piece bound; cap memory by net change At the bound of 16 pieces the store dropped the oldest, which is the whole piece the appends lay on, so a phone could no longer rebuild the tile. It now refuses the extra append and keeps what lays back together; the box's next whole piece carries it. The memory cap counted a tile's full size before what it replaces, so near the cap it refused a whole piece or a closed tile that would have freed bytes. It counts the net change now. Co-Authored-By: Claude Opus 5.5 --- internal/store/tiles.go | 67 +++++++++++++++++++++++------------- internal/store/tiles_test.go | 30 +++++++++++++--- 2 files changed, 69 insertions(+), 28 deletions(-) diff --git a/internal/store/tiles.go b/internal/store/tiles.go index b71df38..b56d79e 100644 --- a/internal/store/tiles.go +++ b/internal/store/tiles.go @@ -29,6 +29,7 @@ var ( 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") ) // heldTile is one tiles envelope as the box sent it. @@ -57,7 +58,8 @@ type tileSet struct { // verified and checked (cloudwire.Tiles.Check), and forwards it to the // phones reading the home that may read tiles. False says it was not held: // the grant holds no tiles, it is older than 48 hours or from the future, a -// piece of a tile already closed, or memory for tiles is full. +// piece of a tile already closed, an append past the pieces a tile may +// have, or memory for tiles is full. func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { if x.kind != Box { return false @@ -76,17 +78,53 @@ func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { return false } ts := &x.h.tiles - if x.s.tileBytes.Load()+int64(len(env)) > x.s.tileBytesMax { + // 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 { + 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 + } + 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 + } + 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 + } + } + if dropOpen { + for _, p := range ts.open { + replaced += int64(len(p.env)) + } + } + if grow := int64(len(env)) - replaced; grow > 0 && x.s.tileBytes.Load()+grow > x.s.tileBytesMax { dropTileFull.Inc() return false } + if dropOpen { + x.dropOpenLocked() + } if !t.Partial { if old, ok := ts.closed[held.id]; ok { x.tileBytesLocked(-int64(len(old.env))) } - if len(ts.open) > 0 && ts.open[0].id == held.id { - x.dropOpenLocked() - } if ts.closed == nil { ts.closed = map[string]heldTile{} } @@ -104,27 +142,8 @@ func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { delete(ts.closed, oldest) } } else { - if _, ok := ts.closed[held.id]; ok { - dropTileClosed.Inc() - return 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 - } - x.dropOpenLocked() - } - if t.Whole() { - x.dropOpenLocked() - } ts.open = append(ts.open, held) x.tileBytesLocked(int64(len(env))) - for len(ts.open) > cloudwire.TilePiecesMax { - x.tileBytesLocked(-int64(len(ts.open[0].env))) - ts.open = ts.open[1:] - } } at := stamp() for sess := range x.h.sessions { diff --git a/internal/store/tiles_test.go b/internal/store/tiles_test.go index 6e247dd..2b708c1 100644 --- a/internal/store/tiles_test.go +++ b/internal/store/tiles_test.go @@ -157,15 +157,32 @@ func TestTilesAreBoundedPerHomeAndInAll(t *testing.T) { for i := int64(4); i >= 1; i-- { put(t, box, closedTile(tStart-i*tSpan), uint64(10-i)) } - for i := 0; i < cloudwire.TilePiecesMax+4; i++ { - put(t, box, piece(tStart, i, i+1), uint64(20+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); box.PutTile(env(late, 40), late) { + 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 { + 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 new tile is dropped. + // 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 @@ -173,6 +190,11 @@ func TestTilesAreBoundedPerHomeAndInAll(t *testing.T) { if box.PutTile(bytes.Repeat([]byte{1}, 64), extra) { t.Fatal("a tile past the store's cap was held") } + before := s.TileBytes() + whole := piece(tStart, 0, 144) + if !box.PutTile(bytes.Repeat([]byte{2}, 40), whole) || s.TileBytes() >= before { + t.Fatal("a whole piece that frees bytes was refused at the cap") + } } func TestTilesOutliveTheHourOfSilenceAndGoAfter48Hours(t *testing.T) { From 724d9b0f0822007e45784eede85538ea48f8d9f7 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 17:44:06 +0200 Subject: [PATCH 5/6] fix: tiles after review: bounded per home, ordered, gone with the roster - An envelope over 4 KiB is dropped, Check takes only the four series (go/protocol pinned at srcfl/ftw b10a6939), and a home holds at most 64 KiB of tiles, so padded tiles from a few boxes cannot fill memory for every home. - A closed tile is replaced only by one the box sent later, by wallMs then seq; a piece sent before the newest held piece is dropped. - Tiles go when the roster runs out: no phone can read them then, and a revoke made while the box was away cannot arrive. - The service sends resync {what: [tiles]}, at most once per 5 minutes, when a roster gives history back or room frees after a drop. - Tests: an agent gets no tiles, the ordering, the home's share, roster expiry, an oversized envelope, and the resync. Co-Authored-By: Claude Opus 5.5 --- README.md | 23 +++--- go.mod | 2 +- go.sum | 3 + internal/appsession/tiles_test.go | 2 +- internal/store/tiles.go | 107 ++++++++++++++++++------ internal/store/tiles_test.go | 130 +++++++++++++++++++++++++++--- internal/uplink/tiles_test.go | 67 +++++++++++++++ internal/uplink/uplink.go | 55 +++++++++++-- 8 files changed, 335 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index 5d8a05c..ef71c4f 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. History tiles stay 48 hours, also in memory only. + 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, @@ -136,17 +136,22 @@ to it yet. 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 holds it in memory as the box - signed it: per home at most five closed tiles and 16 pieces of the filling - one, all homes together at most 256 MiB, each until 48 hours after it - ends. 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. A + 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, but a phone still needs - a roster in force to get in. A revoke, a grant without history, a purge + 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. diff --git a/go.mod b/go.mod index 7b8705a..da70fbc 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-20261023135616-7dedcbf40d88 + github.com/srcfl/ftw/go/protocol v0.0.0-20261023153609-b10a69394f99 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 2ffa902..3a2faec 100644 --- a/go.sum +++ b/go.sum @@ -53,10 +53,13 @@ github.com/quic-go/quic-go v0.63.0/go.mod h1:RAro2j2yN9a9EiPACLHT9IB2NXCvGQmmo/a 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 v3.8.0-beta.1.0.20261023135616-7dedcbf40d88+incompatible h1:E7jKXntZtOPrn24ZbObaIiFPFgVPnmyUvxSZuigGTjQ= +github.com/srcfl/ftw v3.8.0-beta.1.0.20261023153609-b10a69394f99+incompatible h1:40VkJSduSbT7tdENLvaU3u4ntHltQJhXFVhxF69g3Vo= 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-20261023135616-7dedcbf40d88 h1:90QNPvhsMGZJcK9F3HxMT7+rwQei1K8MqUm4bJFhXWI= github.com/srcfl/ftw/go/protocol v0.0.0-20261023135616-7dedcbf40d88/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023153609-b10a69394f99 h1:GXgOwvGKt9lIby/V02LJGdWz1P4LRxC/sDEa0LimtlI= +github.com/srcfl/ftw/go/protocol v0.0.0-20261023153609-b10a69394f99/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/tiles_test.go b/internal/appsession/tiles_test.go index 7392424..0cf22ad 100644 --- a/internal/appsession/tiles_test.go +++ b/internal/appsession/tiles_test.go @@ -73,7 +73,7 @@ func (h *harness) putTile(t *testing.T, raw []byte) { if err := cloudwire.Unmarshal(env.B, &b); err != nil || b.Check() != nil { t.Fatalf("a vector is not a tile: %v", err) } - if !h.box.PutTile(raw, b) { + if kept, _ := h.box.PutTile(raw, b); !kept { t.Fatal("a vector tile was not held") } } diff --git a/internal/store/tiles.go b/internal/store/tiles.go index b56d79e..57f5ceb 100644 --- a/internal/store/tiles.go +++ b/internal/store/tiles.go @@ -18,11 +18,15 @@ import ( // 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. A home holds at -// most five closed tiles and 16 pieces, about 20 KB, so this is some 13,000 -// homes; a shard takes on no more than 4,000. Past it, new tiles are dropped -// and counted. -const TileBytesMax = 256 << 20 +// 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.") @@ -30,6 +34,8 @@ var ( 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. @@ -37,8 +43,16 @@ type heldTile struct { id string etag string start int64 // the tile's start, wall clock - seq uint64 - env []byte + // 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 { @@ -56,26 +70,30 @@ type tileSet struct { // 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. False says it was not held: -// the grant holds no tiles, it is older than 48 hours or from the future, a -// piece of a tile already closed, an append past the pieces a tile may -// have, or memory for tiles is full. -func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { +// 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 + return false, false } x.h.mu.Lock() defer x.h.mu.Unlock() x.h.lastSeen = x.s.now() if !x.h.holdsTilesLocked() { - return false + 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 + return false, false } ts := &x.h.tiles // What the tile replaces, first, so the memory check counts the net @@ -84,19 +102,29 @@ func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { 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 + 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 + return false, false } dropOpen = true } @@ -106,7 +134,7 @@ func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { // 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 + return false, false } } if dropOpen { @@ -114,9 +142,15 @@ func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { replaced += int64(len(p.env)) } } - if grow := int64(len(env)) - replaced; grow > 0 && x.s.tileBytes.Load()+grow > x.s.tileBytesMax { - dropTileFull.Inc() - return false + 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() @@ -151,7 +185,22 @@ func (x *Handle) PutTile(env []byte, t cloudwire.Tiles) bool { sess.deliver(env, at) } } - return true + 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. @@ -255,6 +304,14 @@ func (s *Store) addTileBytes(h *home, n int64) { // 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. Sweep calls it for every -// home, the box online or not. -func (s *Store) sweepTiles(h *home, now time.Time) { s.pruneTiles(h, now.UnixMilli()) } +// 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 index 2b708c1..fad77e7 100644 --- a/internal/store/tiles_test.go +++ b/internal/store/tiles_test.go @@ -69,14 +69,20 @@ func put(t *testing.T, box *Handle, b cloudwire.Tiles, seq uint64) []byte { if err := b.Check(); err != nil { t.Fatal(err) } - b.Seq = seq e := env(b, seq) - if !box.PutTile(e, b) { + 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}) @@ -127,10 +133,10 @@ func TestAWholePieceStartsTheTileOverAndTheClosedTileEndsIt(t *testing.T) { // The tile closes: its pieces go, and a late piece of it is refused. closed := closedTile(tStart) ce := env(closed, 4) - if !box.PutTile(ce, closed) { + if !kept(box, ce, closed, 4) { t.Fatal("closed tile not held") } - if late := piece(tStart, 63, 64); box.PutTile(env(late, 5), late) { + 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) @@ -140,7 +146,7 @@ func TestAWholePieceStartsTheTileOverAndTheClosedTileEndsIt(t *testing.T) { // The next tile's first piece starts a new trailing tile. next := closedTile(tStart + tSpan) next.Partial = true - if !box.PutTile(env(next, 6), next) { + 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 { @@ -151,7 +157,7 @@ func TestAWholePieceStartsTheTileOverAndTheClosedTileEndsIt(t *testing.T) { 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); box.PutTile(env(old, 1), old) { + 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-- { @@ -164,7 +170,7 @@ func TestTilesAreBoundedPerHomeAndInAll(t *testing.T) { 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); box.PutTile(env(late, 40), late) { + 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") } } @@ -187,12 +193,12 @@ func TestTilesAreBoundedPerHomeAndInAll(t *testing.T) { extra := closedTile(tStart - tSpan) extra.Data[0] ^= 1 extra.Etag = protocol.EtagOf(extra.Data) - if box.PutTile(bytes.Repeat([]byte{1}, 64), extra) { + 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 !box.PutTile(bytes.Repeat([]byte{2}, 40), whole) || s.TileBytes() >= before { + 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") } } @@ -224,7 +230,7 @@ func TestTilesOutliveTheHourOfSilenceAndGoAfter48Hours(t *testing.T) { // A tile from the future is refused. future := piece(tStart, 0, 1) c.t = time.UnixMilli(tStart - 2*tSpan) - if box.PutTile(env(future, 9), future) { + if kept(box, env(future, 9), future, 9) { t.Fatal("a tile more than a tile ahead of the clock was held") } } @@ -259,7 +265,7 @@ func TestRevokePurgeOrAGrantWithoutHistoryDropsTheTiles(t *testing.T) { } } // And none is taken while it lasts. - if b := closedTile(tStart - tSpan); name == "history taken out" && box.PutTile(env(b, 3), b) { + 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") } }) @@ -286,7 +292,7 @@ func TestOnlyADeviceWithTheHistoryScopeReadsTiles(t *testing.T) { } // Nor does a widget's token. _ = box.Put(cloudwire.MsgLive, []byte("live"), 2, true) - if !box.PutTile(env(piece(tStart, 0, 144), 3), piece(tStart, 0, 144)) { + 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"}) @@ -348,3 +354,103 @@ func equalEnvs(a, b [][]byte) bool { } 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 index 82c6a70..51eafc7 100644 --- a/internal/uplink/tiles_test.go +++ b/internal/uplink/tiles_test.go @@ -246,3 +246,70 @@ func TestTilesNeedTheHistoryScopeInTheGrant(t *testing.T) { 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 7205ae6..2e78b88 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -92,6 +92,7 @@ var ( 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.") ) @@ -459,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 @@ -658,6 +671,9 @@ func (c *boxConn) run() error { if err != nil || stop { return err } + if err := c.askTiles(); err != nil { + return err + } } } @@ -858,22 +874,41 @@ func (c *boxConn) apply(env cloudwire.Envelope) (stop bool, err error) { } // tile holds one tile the box signed, if it has one of the three shapes -// (cloudwire.Tiles.Check); anything else is dropped and counted, and the -// connection goes on. +// (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 } - raw, err := env.Encode() - if err != nil { - return err + if _, full := c.handle.PutTile(raw, t); full { + c.tilesLost = true } - c.handle.PutTile(raw, t) 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. @@ -984,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 { From c66bc511494eebf01bd530a3c452a562ed433377 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Fri, 23 Oct 2026 18:13:29 +0200 Subject: [PATCH 6/6] chore: pin the protocol to Core master bac597f7 (#1817 merged) Co-Authored-By: Claude Opus 5.5 --- go.mod | 2 +- go.sum | 10 ++-------- 2 files changed, 3 insertions(+), 9 deletions(-) diff --git a/go.mod b/go.mod index da70fbc..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-20261023153609-b10a69394f99 + 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 3a2faec..091766b 100644 --- a/go.sum +++ b/go.sum @@ -52,14 +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 v3.8.0-beta.1.0.20261023135616-7dedcbf40d88+incompatible h1:E7jKXntZtOPrn24ZbObaIiFPFgVPnmyUvxSZuigGTjQ= -github.com/srcfl/ftw v3.8.0-beta.1.0.20261023153609-b10a69394f99+incompatible h1:40VkJSduSbT7tdENLvaU3u4ntHltQJhXFVhxF69g3Vo= -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-20261023135616-7dedcbf40d88 h1:90QNPvhsMGZJcK9F3HxMT7+rwQei1K8MqUm4bJFhXWI= -github.com/srcfl/ftw/go/protocol v0.0.0-20261023135616-7dedcbf40d88/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= -github.com/srcfl/ftw/go/protocol v0.0.0-20261023153609-b10a69394f99 h1:GXgOwvGKt9lIby/V02LJGdWz1P4LRxC/sDEa0LimtlI= -github.com/srcfl/ftw/go/protocol v0.0.0-20261023153609-b10a69394f99/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=