diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index ef5c711a11b..340110808bd 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -993,14 +993,20 @@ paths: get: summary: Subscribe to a BPS cohort description: | - WebSocket. The node joins the BPS-lite (SWIP-74) cohort `{topic, FEED_TOPIC, owner}` - at `broker` as `identity` (`broker` must be a full node started with `bps-broker`) and sends every update as a binary frame - `challenge (32) | index (8, big-endian) | soc`, where `soc` is the full single-owner - chunk (`id | signature | span | payload`) signed by `owner` with - `id = keccak256(keccak256(topic | challenge) | index)`. Updates that do not verify, - or whose index is below the subscriber's cursor, are dropped. + WebSocket. The node joins the BPS cohort `{topic, FEED_TOPIC, owner}` (SWIP-74, with + the multi-publisher mode of SWIP-87) at `broker` as `identity` and sends every + verified delivery as a binary frame + `kind (1) | owner (20) | challenge (32) | index (8, big-endian) | soc`, where `soc` is + the full single-owner chunk (`id | signature | span | payload`) signed by `owner` with + `id = keccak256(sessionTopic | index)`. `kind` is 1 (DATA): an update, by the cohort + owner or by an address on its latest roster; 4 (ROSTER): the owner's publisher set, + payload a `Roster` protobuf of 20-byte addresses; or 3 (EOS): the owner ended the + channel, after which the socket closes with 4004. `sessionTopic` is + `keccak256(topic | challenge)` for DATA and `keccak256("bps-service:v1" | kind | topic | challenge)` + otherwise. Deliveries that do not verify, by an owner the roster does not admit, or + below the subscriber's cursor are dropped. The endpoint is read-only: any client frame closes the socket with 4002. - Close codes: 4002 invalid message, 4003 broker stream closed. + Close codes: 4002 invalid message, 4003 broker stream closed, 4004 end of stream. tags: - BPS parameters: @@ -1033,7 +1039,7 @@ paths: schema: type: integer required: false - description: Lowest update index to deliver. Defaults to 0. + description: Lowest index of the owner's updates to deliver. Defaults to 0. responses: "200": description: Establishes a WebSocket subscription for incoming broadcasts @@ -1048,25 +1054,42 @@ paths: "/bps/publish/{owner}/{topic}": get: - summary: Publish to a BPS cohort as its principal + summary: Publish to a BPS cohort as its owner or as a rostered seat description: | - WebSocket. The node joins the BPS-lite (SWIP-74) cohort `{topic, FEED_TOPIC, owner}` - at `broker` declaring `owner` as its address. - - 1. Server sends `{"type":"challenge","challenge":"","broker":"","dataTopic":"","authTopic":""}`. - `dataTopic` is `keccak256(topic | challenge)`, the session feed for updates; - `authTopic` is `keccak256("bps-service:v1" | 0x02 | topic | challenge)`. - 2. Client sends binary frames `kind (1) | index (8, big-endian) | soc`, where `kind` is - 1 (DATA) or 2 (AUTH) and `soc` is a single-owner chunk signed by `owner` with - `id = keccak256(sessionTopic | index)`. The first frame must be the AUTH — the empty - chunk at index 0 of `authTopic`, which authenticates the stream — and must arrive within - 30 seconds; sent again later it is a heartbeat. DATA frames are updates of - `dataTopic`. Indexes must not be reused across sessions; frames below the broker's - cursor are dropped silently. - - Close codes: 4001 invalid soc (an AUTH at another index or with a payload included), - 4002 invalid message (wrong frame type, any frame before the AUTH, unknown kind, auth - timeout), 4003 broker stream closed. + WebSocket. The node joins the BPS cohort `{topic, FEED_TOPIC, owner}` (SWIP-74, with + the multi-publisher mode of SWIP-87) at `broker` declaring `identity` — the owner by + default, or an address the owner's roster names or will name (a seat). A seat's + stream receives from the start, like a subscriber's; once a roster naming the seat + has reached it, the server sends `{"type":"granted"}` and the seat may send its + AUTH, which upgrades the stream in place. A seat that sends anything before the + grant is closed with 4002. A seat whose address a later roster drops finds the + socket closed after that roster (4003). + + 1. Server sends `{"type":"challenge","challenge":"","broker":"","dataTopic":"","authTopic":""}`, + and to the owner also `"rosterTopic"` and `"eosTopic"`. + `dataTopic` is `keccak256(topic | challenge)`, the session feed for updates; the + others are `keccak256("bps-service:v1" | kind | topic | challenge)` with kind 0x02 + (AUTH), 0x04 (ROSTER), 0x03 (EOS). + 2. Client sends binary frames `kind (1) | index (8, big-endian) | soc`, where `soc` is + a single-owner chunk with `id = keccak256(sessionTopic | index)`, signed by + `identity` for kind 1 (DATA) and 2 (AUTH) and by `owner` for kind 4 (ROSTER) and 3 + (EOS), which only the owner may send. The first frame must be the AUTH — the empty + chunk at index 0 of `authTopic`, which authenticates the stream — and must arrive + within 25 seconds — of the connection for the owner, of the grant for a seat; + sent again later it is a heartbeat. DATA frames are updates of + `dataTopic`. A ROSTER carries a `Roster` protobuf naming every address that may + publish besides the owner, full state at each index; an EOS — the empty chunk at + index 0 of `eosTopic` — ends the channel. An owner that has a roster MUST publish + it again, at its next roster index, right after its AUTH: the broker may hold + none, after an inactivity reclaim or a restart. Indexes must not be reused across + sessions; frames below the broker's cursor are dropped silently. + 3. Server sends what the other publishers publish, and the owner's rosters — to + the owner too — as binary frames in the subscribe endpoint's format. + + Close codes: 4001 invalid soc (an AUTH or EOS at another index or with a payload + included), 4002 invalid message (wrong frame type, any frame before the AUTH, unknown + kind, a ROSTER or EOS from a seat, a roster that does not decode, a frame from a seat + before its grant, auth timeout), 4003 broker stream closed, 4004 end of stream. tags: - BPS parameters: @@ -1088,6 +1111,12 @@ paths: $ref: "SwarmCommon.yaml#/components/schemas/SwarmAddress" required: true description: Overlay address of the broker node. Must not be this node. + - in: query + name: identity + schema: + $ref: "SwarmCommon.yaml#/components/schemas/EthereumAddress" + required: false + description: The address the stream publishes as. Defaults to the owner; any other may publish once the owner's roster names it. responses: "200": description: Establishes a WebSocket subscription for publishing broadcasts diff --git a/pkg/api/bps.go b/pkg/api/bps.go index b48e8ca7a7a..3d9ac59443e 100644 --- a/pkg/api/bps.go +++ b/pkg/api/bps.go @@ -28,17 +28,21 @@ const ( bpsCloseInvalidSOC = 4001 // a chunk that does not validate under the session challenge bpsCloseInvalidMessage = 4002 // frame violates the endpoint protocol bpsCloseBrokerGone = 4003 // the p2p stream to the broker ended + bpsCloseEndOfStream = 4004 // the owner ended the channel with an EOS bpsMaxCloseReason = 123 // websocket control frame payload limit minus the code bpsFrameHeader = 1 + 8 // publish frame header: kind | index (big-endian) bpsMaxSOCSize = swarm.HashSize + swarm.SocSignatureSize + swarm.SpanSize + swarm.ChunkSize bpsMaxFrameSize = bpsFrameHeader + bpsMaxSOCSize + // delivery frame header: kind | owner | challenge | index (big-endian) + bpsDeliveryHeader = 1 + common.AddressLength + bps.ChallengeSize + 8 ) -// bpsAuthTimeout bounds how long a publisher may take to send its AUTH. It is -// counted from before the join, so that it runs out before the broker's auth -// deadline does and the broker never blocklists this node for a slow dApp. -var bpsAuthTimeout = 30 * time.Second +// bpsAuthTimeout bounds how long a publisher may take to send its AUTH. For the +// owner it is counted from before the join and is shorter than the broker's +// auth deadline, so that the broker never blocklists this node for a slow +// dApp; for a seat it is counted from the grant. +var bpsAuthTimeout = 25 * time.Second var errBPSBrokerGone = errors.New("broker stream closed") @@ -50,12 +54,41 @@ type bpsRequest struct { cursor uint64 } +// bpsGrantedMessage tells a seat's dApp that a roster naming its identity has +// reached the stream: it may now send its AUTH. +type bpsGrantedMessage struct { + Type string `json:"type"` +} + type bpsChallengeMessage struct { - Type string `json:"type"` - Challenge string `json:"challenge"` - Broker string `json:"broker"` - DataTopic string `json:"dataTopic"` - AuthTopic string `json:"authTopic"` + Type string `json:"type"` + Challenge string `json:"challenge"` + Broker string `json:"broker"` + DataTopic string `json:"dataTopic"` + AuthTopic string `json:"authTopic"` + RosterTopic string `json:"rosterTopic,omitempty"` // the owner's only + EOSTopic string `json:"eosTopic,omitempty"` // the owner's only +} + +// bpsDeliveryFrame encodes a delivery for the dApp: kind | owner | challenge | +// index | soc — what it needs to re-verify the chunk, whoever published it. +func bpsDeliveryFrame(msg bps.Message) []byte { + data := make([]byte, 0, bpsDeliveryHeader+len(msg.SOC)) + data = append(data, byte(msg.Kind)) + data = append(data, msg.Owner...) + data = append(data, msg.Challenge...) + data = binary.BigEndian.AppendUint64(data, msg.Index) + return append(data, msg.SOC...) +} + +// bpsSessionEnded closes the socket for a session whose stream ended: with +// 4004 after an EOS, 4003 otherwise. +func (s *Service) bpsSessionEnded(conn *websocket.Conn, sess bps.Session) { + if err := sess.Err(); errors.Is(err, bps.ErrEndOfStream) { + s.bpsClose(conn, bpsCloseEndOfStream, "end of stream") + return + } + s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err())) } func (s *Service) bpsSubscribeWsHandler(w http.ResponseWriter, r *http.Request) { @@ -88,6 +121,10 @@ func (s *Service) bpsPublishWsHandler(w http.ResponseWriter, r *http.Request) { if !ok { return } + // the owner publishes by default; a seat names its own address + if len(req.identity) == 0 { + req.identity = req.owner.Bytes() + } conn, ok := s.bpsUpgrade(w, r, logger) if !ok { return @@ -244,7 +281,7 @@ func (s *Service) bpsSubscribeWs(conn *websocket.Conn, req bpsRequest, logger lo Broker: req.broker, Spec: bps.CohortSpec{Topic: req.topic, Principal: req.owner.Bytes()}, Identity: req.identity, - Cursor: req.cursor, + Position: bps.Position{Cursors: map[string]uint64{string(req.owner.Bytes()): req.cursor}}, }) if err != nil { logger.Debug("join failed", "broker", req.broker, "error", err) @@ -259,15 +296,10 @@ func (s *Service) bpsSubscribeWs(conn *websocket.Conn, req bpsRequest, logger lo select { case msg, ok := <-sess.Messages(): if !ok { - s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err())) + s.bpsSessionEnded(conn, sess) return } - // challenge | index | soc: what the client needs to re-verify the chunk - data := make([]byte, 0, bps.ChallengeSize+8+len(msg.SOC)) - data = append(data, msg.Challenge...) - data = binary.BigEndian.AppendUint64(data, msg.Index) - data = append(data, msg.SOC...) - if err := s.bpsWrite(conn, websocket.BinaryMessage, data); err != nil { + if err := s.bpsWrite(conn, websocket.BinaryMessage, bpsDeliveryFrame(msg)); err != nil { logger.Debug("write broadcast failed", "error", err) return } @@ -275,7 +307,7 @@ func (s *Service) bpsSubscribeWs(conn *websocket.Conn, req bpsRequest, logger lo s.bpsClose(conn, bpsCloseInvalidMessage, "subscribe endpoint is read-only") return case <-sess.Done(): - s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err())) + s.bpsSessionEnded(conn, sess) return case <-gone: return @@ -304,14 +336,29 @@ func bpsParseFrame(f bpsFrame, req bpsRequest, challenge []byte, authenticated b if !authenticated && kind != bps.KindAuth { return 0, 0, nil, bpsCloseInvalidMessage, "expected auth" } - if kind != bps.KindData && kind != bps.KindAuth { + owner := req.identity + switch kind { + case bps.KindData, bps.KindAuth: + case bps.KindRoster, bps.KindEOS: + // the service feeds are the owner's alone + if !bytes.Equal(req.identity, req.owner.Bytes()) { + return 0, 0, nil, bpsCloseInvalidMessage, "roster and eos are the owner's" + } + owner = req.owner.Bytes() + default: return 0, 0, nil, bpsCloseInvalidMessage, "unknown kind" } index := binary.BigEndian.Uint64(f.data[1:bpsFrameHeader]) chunk := f.data[bpsFrameHeader:] - if err := bps.Verify(kind, challenge, index, chunk, req.topic, req.owner.Bytes()); err != nil { + if err := bps.Verify(kind, challenge, index, chunk, req.topic, owner); err != nil { return 0, 0, nil, bpsCloseInvalidSOC, "invalid soc" } + // the broker drops a roster that does not decode without a word + if kind == bps.KindRoster { + if _, err := bps.ParseRoster(chunk); err != nil { + return 0, 0, nil, bpsCloseInvalidMessage, "invalid roster" + } + } return kind, index, chunk, 0, "" } @@ -338,11 +385,15 @@ func (s *Service) bpsPublishWs(conn *websocket.Conn, req bpsRequest, logger log. authTimer := time.NewTimer(bpsAuthTimeout) defer authTimer.Stop() - // the publisher declares the principal as its identity and authenticates with its AUTH + // the publisher declares its identity — the owner's, or a seat the owner's + // roster names — and authenticates with its AUTH: the owner at once, a + // seat once a roster naming it has reached the stream, which upgrades it + // in place + owner := bytes.Equal(req.identity, req.owner.Bytes()) sess, err := s.bps.Join(ctx, bps.JoinRequest{ Broker: req.broker, Spec: bps.CohortSpec{Topic: req.topic, Principal: req.owner.Bytes()}, - Identity: req.owner.Bytes(), + Identity: req.identity, }) if err != nil { logger.Debug("join failed", "broker", req.broker, "error", err) @@ -351,6 +402,14 @@ func (s *Service) bpsPublishWs(conn *websocket.Conn, req bpsRequest, logger log. } defer sess.Close() + // a seat listens until it is granted; its auth deadline starts then + granted := owner + var grantedC <-chan struct{} + if !owner { + authTimer.Stop() + grantedC = sess.Granted() + } + challenge := sess.Challenge() if len(challenge) != bps.ChallengeSize { logger.Debug("invalid challenge length", "broker", req.broker, "length", len(challenge)) @@ -367,13 +426,27 @@ func (s *Service) bpsPublishWs(conn *websocket.Conn, req bpsRequest, logger log. s.bpsClose(conn, bpsCloseBrokerGone, "session topic") return } - if err := s.bpsWriteJSON(conn, bpsChallengeMessage{ + msg := bpsChallengeMessage{ Type: "challenge", Challenge: hex.EncodeToString(challenge), Broker: req.broker.String(), DataTopic: hex.EncodeToString(dataTopic), AuthTopic: hex.EncodeToString(authTopic), - }); err != nil { + } + if owner { + rosterTopic, err := bps.SessionTopic(bps.KindRoster, req.topic, challenge) + if err != nil { + s.bpsClose(conn, bpsCloseBrokerGone, "session topic") + return + } + eosTopic, err := bps.SessionTopic(bps.KindEOS, req.topic, challenge) + if err != nil { + s.bpsClose(conn, bpsCloseBrokerGone, "session topic") + return + } + msg.RosterTopic, msg.EOSTopic = hex.EncodeToString(rosterTopic), hex.EncodeToString(eosTopic) + } + if err := s.bpsWriteJSON(conn, msg); err != nil { logger.Debug("write challenge failed", "error", err) return } @@ -383,7 +456,19 @@ func (s *Service) bpsPublishWs(conn *websocket.Conn, req bpsRequest, logger log. for { select { + case <-grantedC: + grantedC, granted = nil, true + authTimer.Reset(bpsAuthTimeout) + if err := s.bpsWriteJSON(conn, bpsGrantedMessage{Type: "granted"}); err != nil { + logger.Debug("write granted failed", "error", err) + return + } case f := <-frames: + // a seat sends nothing before a roster names it + if !granted { + s.bpsClose(conn, bpsCloseInvalidMessage, "not granted") + return + } kind, index, chunk, code, reason := bpsParseFrame(f, req, challenge, authenticated) if code != 0 { s.bpsClose(conn, code, reason) @@ -397,13 +482,22 @@ func (s *Service) bpsPublishWs(conn *websocket.Conn, req bpsRequest, logger log. authenticated = true authTimer.Stop() } + case msg, ok := <-sess.Messages(): + if !ok { + s.bpsSessionEnded(conn, sess) + return + } + if err := s.bpsWrite(conn, websocket.BinaryMessage, bpsDeliveryFrame(msg)); err != nil { + logger.Debug("write broadcast failed", "error", err) + return + } case <-authTimer.C: if !authenticated { s.bpsClose(conn, bpsCloseInvalidMessage, "auth timeout") return } case <-sess.Done(): - s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err())) + s.bpsSessionEnded(conn, sess) return case <-gone: return diff --git a/pkg/api/bps_test.go b/pkg/api/bps_test.go index b124cb1d366..71e612a3d43 100644 --- a/pkg/api/bps_test.go +++ b/pkg/api/bps_test.go @@ -56,6 +56,7 @@ type published struct { type fakeSession struct { challenge []byte msgs chan bps.Message + granted chan struct{} done chan struct{} published chan published closed chan struct{} @@ -73,6 +74,7 @@ func newFakeSession() *fakeSession { return &fakeSession{ challenge: challenge, msgs: make(chan bps.Message), + granted: closedChan(), done: make(chan struct{}), published: make(chan published, 8), closed: make(chan struct{}), @@ -81,9 +83,16 @@ func newFakeSession() *fakeSession { func (f *fakeSession) Challenge() []byte { return f.challenge } func (f *fakeSession) Messages() <-chan bps.Message { return f.msgs } -func (f *fakeSession) Cursor() uint64 { return 0 } -func (f *fakeSession) Done() <-chan struct{} { return f.done } -func (f *fakeSession) Err() error { return f.err } +func (f *fakeSession) Position() bps.Position { return bps.Position{} } +func (f *fakeSession) Granted() <-chan struct{} { return f.granted } + +func closedChan() chan struct{} { + c := make(chan struct{}) + close(c) + return c +} +func (f *fakeSession) Done() <-chan struct{} { return f.done } +func (f *fakeSession) Err() error { return f.err } func (f *fakeSession) Publish(_ context.Context, kind bps.Kind, index uint64, b []byte) error { if f.publishErr != nil { @@ -157,9 +166,22 @@ func (f *bpsFixture) path(endpoint string) string { return fmt.Sprintf("/bps/%s/%s/%s", endpoint, hex.EncodeToString(f.owner), hex.EncodeToString(f.topic)) } +// dial opens the endpoint: subscribe as the fixture's identity, publish as +// the owner. func (f *bpsFixture) dial(t *testing.T, endpoint string) *websocket.Conn { t.Helper() - q := "broker=" + f.broker.String() + "&identity=" + hex.EncodeToString(f.identity) + if endpoint == "subscribe" { + return f.dialAs(t, endpoint, f.identity) + } + return f.dialAs(t, endpoint, nil) +} + +func (f *bpsFixture) dialAs(t *testing.T, endpoint string, identity []byte) *websocket.Conn { + t.Helper() + q := "broker=" + f.broker.String() + if identity != nil { + q += "&identity=" + hex.EncodeToString(identity) + } u := url.URL{Scheme: "ws", Host: f.listener, Path: f.path(endpoint), RawQuery: q} conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil) if err != nil { @@ -272,12 +294,14 @@ func TestBPSSubscribe(t *testing.T) { f := newBPSFixture(t) conn := f.dial(t, "subscribe") chunk := f.signedSOC(t, f.signer, bps.KindData, f.sess.challenge, 7, []byte("hello cohort")) - f.sess.msgs <- bps.Message{SOC: chunk, Challenge: f.sess.challenge, Index: 7} + f.sess.msgs <- bps.Message{Kind: bps.KindData, Owner: f.owner, SOC: chunk, Challenge: f.sess.challenge, Index: 7} typ, got, err := conn.ReadMessage() if err != nil { t.Fatal(err) } - want := append(append([]byte{}, f.sess.challenge...), 0, 0, 0, 0, 0, 0, 0, 7) + want := append([]byte{byte(bps.KindData)}, f.owner...) + want = append(want, f.sess.challenge...) + want = append(want, 0, 0, 0, 0, 0, 0, 0, 7) want = append(want, chunk...) if typ != websocket.BinaryMessage || !bytes.Equal(got, want) { t.Fatalf("got type %d %x, want binary %x", typ, got, want) @@ -617,3 +641,124 @@ func TestBPSPublishInvalidChallengeLength(t *testing.T) { } waitSessionClosed(t, f) } + +func TestBPSSubscribeEndOfStream(t *testing.T) { + t.Parallel() + f := newBPSFixture(t) + conn := f.dial(t, "subscribe") + f.sess.end(bps.ErrEndOfStream) + expectClose(t, conn, api.BPSCloseEndOfStream) +} + +func TestBPSPublishForwardsDeliveries(t *testing.T) { + t.Parallel() + f := newBPSFixture(t) + conn := f.dial(t, "publish") + _ = readChallenge(t, f, conn) + seat := make([]byte, 20) + copy(seat, "bps-test-seat") + f.sess.msgs <- bps.Message{Kind: bps.KindData, Owner: seat, SOC: []byte("soc"), Challenge: f.sess.challenge, Index: 1} + typ, got, err := conn.ReadMessage() + if err != nil { + t.Fatal(err) + } + if typ != websocket.BinaryMessage || got[0] != byte(bps.KindData) || !bytes.Equal(got[1:21], seat) { + t.Fatalf("got type %d %x", typ, got) + } +} + +func TestBPSPublishRoster(t *testing.T) { + t.Parallel() + f := newBPSFixture(t) + conn := f.dial(t, "publish") + challenge := readChallenge(t, f, conn) + authenticate(t, f, conn, challenge) + + payload, err := bps.RosterPayload([][]byte{f.identity}) + if err != nil { + t.Fatal(err) + } + ros := f.signedSOC(t, f.signer, bps.KindRoster, challenge, 0, payload) + if err := conn.WriteMessage(websocket.BinaryMessage, frame(bps.KindRoster, 0, ros)); err != nil { + t.Fatal(err) + } + expectPublished(t, f, bps.KindRoster, 0, ros) + + bad := f.signedSOC(t, f.signer, bps.KindRoster, challenge, 1, []byte{0x0a, 0x01, 0x01}) + if err := conn.WriteMessage(websocket.BinaryMessage, frame(bps.KindRoster, 1, bad)); err != nil { + t.Fatal(err) + } + expectClose(t, conn, api.BPSCloseInvalidMessage) +} + +func TestBPSPublishSeat(t *testing.T) { + t.Parallel() + f := newBPSFixture(t) + f.sess.granted = make(chan struct{}) + key, _ := crypto.GenerateSecp256k1Key() + seatSigner := crypto.NewDefaultSigner(key) + seatAddr, _ := seatSigner.EthereumAddress() + seat := seatAddr.Bytes() + + conn := f.dialAs(t, "publish", seat) + var m struct { + Type string `json:"type"` + Challenge string `json:"challenge"` + RosterTopic string `json:"rosterTopic"` + } + if err := conn.ReadJSON(&m); err != nil { + t.Fatal(err) + } + if m.Type != "challenge" || m.RosterTopic != "" { + t.Fatalf("unexpected challenge message %+v", m) + } + f.bps.mu.Lock() + req := f.bps.req + f.bps.mu.Unlock() + if !bytes.Equal(req.Identity, seat) { + t.Fatalf("unexpected join request %+v", req) + } + + // granted once a roster naming the seat has reached the stream + close(f.sess.granted) + if err := conn.ReadJSON(&m); err != nil || m.Type != "granted" { + t.Fatalf("want granted, got %+v %v", m, err) + } + challenge := f.sess.challenge + auth := f.signedSOC(t, seatSigner, bps.KindAuth, challenge, 0, nil) + if err := conn.WriteMessage(websocket.BinaryMessage, frame(bps.KindAuth, 0, auth)); err != nil { + t.Fatal(err) + } + expectPublished(t, f, bps.KindAuth, 0, auth) + data := f.signedSOC(t, seatSigner, bps.KindData, challenge, 0, []byte("from the seat")) + if err := conn.WriteMessage(websocket.BinaryMessage, frame(bps.KindData, 0, data)); err != nil { + t.Fatal(err) + } + expectPublished(t, f, bps.KindData, 0, data) + + // the roster is the owner's alone + payload, _ := bps.RosterPayload(nil) + ros := f.signedSOC(t, seatSigner, bps.KindRoster, challenge, 0, payload) + if err := conn.WriteMessage(websocket.BinaryMessage, frame(bps.KindRoster, 0, ros)); err != nil { + t.Fatal(err) + } + expectClose(t, conn, api.BPSCloseInvalidMessage) +} + +func TestBPSPublishSeatNotGranted(t *testing.T) { + t.Parallel() + f := newBPSFixture(t) + f.sess.granted = make(chan struct{}) + conn := f.dialAs(t, "publish", f.identity) + _ = readChallenge(t, f, conn) + // a seat listens until it is granted, and may send nothing before + f.sess.msgs <- bps.Message{Kind: bps.KindData, Owner: f.owner, SOC: []byte("soc"), Challenge: f.sess.challenge} + if typ, _, err := conn.ReadMessage(); err != nil || typ != websocket.BinaryMessage { + t.Fatalf("want a delivery, got %d %v", typ, err) + } + auth := f.signedSOC(t, f.signer, bps.KindAuth, f.sess.challenge, 0, nil) + if err := conn.WriteMessage(websocket.BinaryMessage, frame(bps.KindAuth, 0, auth)); err != nil { + t.Fatal(err) + } + expectClose(t, conn, api.BPSCloseInvalidMessage) +} diff --git a/pkg/api/export_test.go b/pkg/api/export_test.go index d4b8cf6db1f..2bd8d6b04f1 100644 --- a/pkg/api/export_test.go +++ b/pkg/api/export_test.go @@ -163,6 +163,7 @@ const ( BPSCloseInvalidSOC = bpsCloseInvalidSOC BPSCloseInvalidMessage = bpsCloseInvalidMessage BPSCloseBrokerGone = bpsCloseBrokerGone + BPSCloseEndOfStream = bpsCloseEndOfStream ) // SetBPSAuthTimeout overrides the auth timeout and returns a restore func. diff --git a/pkg/bps/bps.go b/pkg/bps/bps.go index a9823643cc7..ec396be5e0c 100644 --- a/pkg/bps/bps.go +++ b/pkg/bps/bps.go @@ -2,16 +2,19 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -// Package bps implements BPS-lite (SWIP-74): a single-publisher live stream -// over a feed, through one broker, one hop. +// Package bps implements BPS-lite (SWIP-74) with the explicit multi-publisher +// mode of SWIP-87: a live stream over feeds, through one broker, one hop. // // A cohort is identified by its spec {topic, FEED_TOPIC, principal}. Every // peer joins with the spec and its own identity and is answered with a random -// per-stream challenge. A stream declaring the principal is pending until its -// AUTH — the empty chunk at index 0 of the session's AUTH feed — authenticates -// it; a publisher stream then sends ordinary single-owner chunks that are updates -// of the session feed on keccak256(topic | challenge). The broker keeps one cursor -// per cohort and delivers every accepted update to every subscriber stream. +// per-stream challenge. A stream declaring the principal — the admin — or an +// address on the admin's latest roster is pending until its AUTH — the empty +// chunk at index 0 of the session's AUTH feed — authenticates it; a publisher +// stream then sends ordinary single-owner chunks that are updates of its session +// feed on keccak256(topic | challenge), and the admin's publisher stream also its +// service messages: ROSTER, the publisher set, and EOS, the end of the channel. +// The broker keeps a cursor per publisher and per service kind, and delivers +// every accepted frame to every receiving stream not bound to its author. package bps import ( @@ -21,7 +24,6 @@ import ( "errors" "fmt" "sync" - "sync/atomic" "time" "github.com/ethersphere/bee/v2/pkg/bps/pb" @@ -53,6 +55,10 @@ var ( errFull = errors.New("bps: capacity bound") errNotPublisher = errors.New("bps: session is not a publisher") errInvalidDelivery = errors.New("bps: invalid delivery") + errEndOfStream = errors.New("bps: end of stream") + // ErrEndOfStream is a session's Err after the admin's EOS ended the channel. + ErrEndOfStream = errEndOfStream + errChannelEnded = errors.New("bps: channel ended") ) // Options are the broker's resource bounds. Zero values take the defaults. @@ -61,23 +67,27 @@ type Options struct { MaxBrokerCohorts int // live cohorts per broker MaxCohortsPerConnection int // cohorts per peer connection MaxPeerStreamsPerCohort int // streams per peer connection per cohort - QueueCap int // outbound frames per subscriber stream + MaxPublisherStreams int // publisher streams per address per cohort + QueueCap int // outbound frames per receiving stream // InactiveTimeout time.Duration // reclaims a cohort with no accepted frame AuthWaitTimeout time.Duration // disconnects a pending stream that has not authenticated + FlushTimeout time.Duration // bounds writing the queue of a stream being closed // ProtoBreachBlocklistDuration time.Duration // blocklist duration for a protocol violation InvalidAuthBlocklistDuration time.Duration // blocklist duration for an auth timeout } -// DefaultOptions are the SWIP-74 recommended bounds. +// DefaultOptions are the SWIP-74 and SWIP-87 recommended bounds. var DefaultOptions = Options{ MaxSubsPerCohort: 1024, MaxBrokerCohorts: 512, MaxCohortsPerConnection: 16, MaxPeerStreamsPerCohort: 2, + MaxPublisherStreams: 2, InactiveTimeout: 10 * time.Minute, AuthWaitTimeout: 30 * time.Second, + FlushTimeout: 5 * time.Second, QueueCap: 64, ProtoBreachBlocklistDuration: 10 * time.Minute, InvalidAuthBlocklistDuration: time.Minute, @@ -97,12 +107,18 @@ func (o Options) withDefaults() Options { if o.MaxPeerStreamsPerCohort > 0 { d.MaxPeerStreamsPerCohort = o.MaxPeerStreamsPerCohort } + if o.MaxPublisherStreams > 0 { + d.MaxPublisherStreams = o.MaxPublisherStreams + } if o.InactiveTimeout > 0 { d.InactiveTimeout = o.InactiveTimeout } if o.AuthWaitTimeout > 0 { d.AuthWaitTimeout = o.AuthWaitTimeout } + if o.FlushTimeout > 0 { + d.FlushTimeout = o.FlushTimeout + } if o.QueueCap > 0 { d.QueueCap = o.QueueCap } @@ -180,12 +196,35 @@ type Interface interface { Join(ctx context.Context, req JoinRequest) (Session, error) } -// Message is a verified DATA delivery: the chunk and the frame fields its id -// was derived from. +// Message is a verified delivery — a DATA update of an admitted publisher, or +// one of the admin's ROSTER and EOS — with the frame fields its id was derived +// from and the owner it validated for. type Message struct { + Kind Kind + Owner []byte // the publisher; the principal for ROSTER and EOS SOC []byte Challenge []byte Index uint64 + Roster [][]byte // ROSTER only: the publisher set besides the principal +} + +// Position is a subscriber's place on a channel, kept across streams and +// sessions: a cursor per publisher, the latest roster with its cursor, and +// whether the channel has ended. +type Position struct { + Cursors map[string]uint64 // by the publisher's 20-byte address as a string: the lowest DATA index delivered next + Roster [][]byte // the latest accepted roster + RosterCursor uint64 // the lowest ROSTER index accepted next + Ended bool // an EOS was accepted: the EOS cursor is past 0 +} + +func (p Position) clone() Position { + c := Position{Cursors: make(map[string]uint64, len(p.Cursors)), RosterCursor: p.RosterCursor, Ended: p.Ended} + for k, v := range p.Cursors { + c.Cursors[k] = v + } + c.Roster = append([][]byte(nil), p.Roster...) + return c } // Session is a single joined p2p stream to a broker. @@ -193,14 +232,23 @@ type Session interface { // Challenge is the challenge the broker issued for this stream. Every // chunk published on the session must be signed under it (see ID). Challenge() []byte - // Messages yields the DATA deliveries that verify against the cohort spec - // and the session cursor. The channel is closed when the stream ends. + // Messages yields the deliveries that verify end to end against the cohort + // spec, the roster and the session's cursors: DATA from the principal or a + // rostered publisher, and the principal's ROSTER and EOS. The channel is + // closed when the stream ends; after an EOS, with Err reporting it. Messages() <-chan Message - // Cursor is the lowest DATA index the session will deliver next. - Cursor() uint64 + // Position is the session's place on the channel, to pass to a later Join. + Position() Position + // Granted is closed once the session may authenticate: at once for the + // principal, and for any other identity once a roster naming it has been + // delivered on this stream — the cue a grantee waits for before its AUTH, + // which upgrades the stream in place. A publisher whose identity a roster + // drops finds its stream closed after that roster. + Granted() <-chan struct{} // Publish writes a chunk of the given kind and index, signed under the - // session challenge, to the broker. The broker does not reply. Only a - // session whose identity is the cohort principal may publish. + // session challenge, to the broker. The broker does not reply. Only the + // principal, or an identity the latest roster names, may publish; only the + // principal may publish ROSTER and EOS. Publish(ctx context.Context, kind Kind, index uint64, soc []byte) error // Done is closed when the p2p stream ends. Done() <-chan struct{} @@ -210,11 +258,11 @@ type Session interface { Close() error } -// CohortSpec names a cohort: the feed of the principal on the topic. The -// topic binding is always FEED_TOPIC. +// CohortSpec names a cohort: the feeds on the topic of the principal and of +// whoever its roster names. The topic binding is always FEED_TOPIC. type CohortSpec struct { Topic []byte // 32 bytes: the feed topic - Principal []byte // 20 bytes: the publisher + Principal []byte // 20 bytes: the admin } func (c CohortSpec) proto() *pb.CohortSpec { @@ -226,7 +274,7 @@ type JoinRequest struct { Broker swarm.Address // the broker Spec CohortSpec // the cohort to join Identity []byte // 20 bytes: the joining stream's identity - Cursor uint64 // the lowest DATA index to deliver, the subscriber's cursor for the cohort + Position Position // the subscriber's place on the channel, from an earlier session } // Join joins a cohort at the broker. A peer whose join was refused or whose @@ -236,6 +284,10 @@ func (s *Service) Join(ctx context.Context, req JoinRequest) (Session, error) { if err := validateJoin(join); err != nil { return nil, err } + // a channel the subscriber has seen end is not rejoined + if req.Position.Ended { + return nil, errChannelEnded + } stream, err := s.streamer.NewStream(ctx, req.Broker, nil, protocolName, protocolVersion, streamName) if err != nil { return nil, fmt.Errorf("new stream: %w", err) @@ -266,11 +318,27 @@ func (s *Service) Join(ctx context.Context, req JoinRequest) (Session, error) { cancel: cancel, w: w, challenge: ack.Challenge, - publisher: bytes.Equal(req.Identity, req.Spec.Principal), + spec: req.Spec, + identity: req.Identity, + pos: req.Position.clone(), + granted: make(chan struct{}), rx: make(chan Message), done: make(chan struct{}), } - sess.cursor.Store(req.Cursor) + if bytes.Equal(req.Identity, req.Spec.Principal) { + sess.grant() + } + + // a broker that forwards what it should have refused is avoided for good: + // blocklisted for the longest period there is, which is forever + sess.forged = func(err error) { + if s.blocklister == nil { + return + } + if berr := s.blocklister.Blocklist(req.Broker, 0, err.Error()); berr != nil { + s.logger.Debug("blocklist broker failed", "broker", req.Broker, "error", berr) + } + } go sess.readLoop(stream, r, req.Spec) @@ -282,7 +350,7 @@ func (s *session) readLoop(stream p2p.Stream, r protobuf.Reader, spec CohortSpec var readErr error defer func() { s.finish(readErr) - if readErr != nil && s.ctx.Err() == nil { + if readErr != nil && !errors.Is(readErr, errEndOfStream) && s.ctx.Err() == nil { _ = stream.Reset() } else { _ = stream.FullClose() @@ -299,44 +367,128 @@ func (s *session) readLoop(stream p2p.Stream, r protobuf.Reader, spec CohortSpec readErr = err return } - // a retransmit below the cursor is skipped. The challenge is not - // compared: it is the publisher stream's, which the subscriber never sees. - if f.Kind != pb.Kind_DATA || f.Index < s.cursor.Load() { - continue - } - // re-verify end to end: a broker that forwards a chunk not valid - // for the spec is misbehaving, so the stream is ended - if err := verify(&f, spec.Topic, spec.Principal); err != nil { + msg, ok, err := s.accept(&f, spec) + if err != nil { + // a broker that forwards a chunk not valid for the spec, or by an + // owner the roster does not admit, is misbehaving: the stream is ended readErr = fmt.Errorf("%w: %w", errInvalidDelivery, err) + s.forged(readErr) return } - s.cursor.Store(f.Index + 1) + if !ok { + continue + } select { - case s.rx <- Message{SOC: f.Soc, Challenge: f.Challenge, Index: f.Index}: + case s.rx <- msg: case <-s.ctx.Done(): return } + if msg.Kind == KindEOS { + readErr = errEndOfStream + return + } } } +// accept re-verifies a delivery end to end. It returns false for a frame the +// subscriber drops: an AUTH, a kind it does not know, one below its cursor, a +// roster that does not decode. The challenge is not compared: it is the +// publisher stream's, which the subscriber never sees. +func (s *session) accept(f *pb.Broadcast, spec CohortSpec) (Message, bool, error) { + msg := Message{SOC: f.Soc, Challenge: f.Challenge, Index: f.Index} + switch f.Kind { + case pb.Kind_DATA: + owner, err := recoverOwner(f, spec.Topic) + if err != nil { + return msg, false, err + } + s.mtx.Lock() + defer s.mtx.Unlock() + if !s.admits(owner) { + return msg, false, fmt.Errorf("owner %x not rostered", owner) + } + k := string(owner) + if f.Index < s.pos.Cursors[k] { + return msg, false, nil + } + s.pos.Cursors[k] = f.Index + 1 + msg.Kind, msg.Owner = KindData, owner + return msg, true, nil + + case pb.Kind_ROSTER: + // the owner of a service chunk is forced to the principal + if err := verify(f, spec.Topic, spec.Principal); err != nil { + return msg, false, err + } + roster, ok := rosterOf(f.Soc) + if !ok { + return msg, false, nil + } + s.mtx.Lock() + defer s.mtx.Unlock() + if f.Index >= s.pos.RosterCursor { + s.pos.RosterCursor = f.Index + 1 + s.pos.Roster = roster + msg.Kind, msg.Owner, msg.Roster = KindRoster, spec.Principal, roster + } + // the cue: a roster naming the identity, delivered on this stream, + // while the latest roster still names it — the broker holds it too + if contains(roster, s.identity) && s.admits(s.identity) { + s.grant() + } + return msg, msg.Kind == KindRoster, nil + + case pb.Kind_EOS: + if err := verify(f, spec.Topic, spec.Principal); err != nil { + return msg, false, err + } + s.mtx.Lock() + s.pos.Ended = true + s.mtx.Unlock() + msg.Kind, msg.Owner = KindEOS, spec.Principal + return msg, true, nil + } + return msg, false, nil +} + +// admits reports whether owner may publish: the principal, or an address on the +// latest roster. Must be called with s.mtx held. +func (s *session) admits(owner []byte) bool { + return bytes.Equal(owner, s.spec.Principal) || contains(s.pos.Roster, owner) +} + +func contains(set [][]byte, a []byte) bool { + for _, b := range set { + if bytes.Equal(a, b) { + return true + } + } + return false +} + // session is the concrete Session implementation. type session struct { - ctx context.Context - cancel context.CancelFunc - wmtx sync.Mutex - w protobuf.Writer - challenge []byte - publisher bool // the identity is the cohort principal - cursor atomic.Uint64 - rx chan Message - done chan struct{} - err error - once sync.Once + ctx context.Context + cancel context.CancelFunc + wmtx sync.Mutex + w protobuf.Writer + challenge []byte + spec CohortSpec + identity []byte + forged func(error) // called on a delivery the broker should have refused + mtx sync.Mutex + pos Position + granted chan struct{} + grantedOnce sync.Once + rx chan Message + done chan struct{} + err error + once sync.Once } func (s *session) Challenge() []byte { return s.challenge } func (s *session) Messages() <-chan Message { return s.rx } -func (s *session) Cursor() uint64 { return s.cursor.Load() } +func (s *session) Granted() <-chan struct{} { return s.granted } func (s *session) Done() <-chan struct{} { return s.done } func (s *session) Err() error { return s.err } func (s *session) Close() error { @@ -344,8 +496,25 @@ func (s *session) Close() error { return nil } +func (s *session) Position() Position { + s.mtx.Lock() + defer s.mtx.Unlock() + return s.pos.clone() +} + +func (s *session) grant() { s.grantedOnce.Do(func() { close(s.granted) }) } + func (s *session) Publish(ctx context.Context, kind Kind, index uint64, soc []byte) error { - if !s.publisher { + principal := bytes.Equal(s.identity, s.spec.Principal) + select { + case <-s.granted: + default: + return errNotPublisher + } + s.mtx.Lock() + seated := s.admits(s.identity) + s.mtx.Unlock() + if !seated || !principal && (kind == KindRoster || kind == KindEOS) { return errNotPublisher } select { @@ -359,6 +528,8 @@ func (s *session) Publish(ctx context.Context, kind Kind, index uint64, soc []by if err := s.w.WriteMsgWithContext(ctx, &f); err != nil { return fmt.Errorf("write broadcast: %w", err) } + // a roster or an EOS the admin publishes is delivered back to it, and it + // admits by it from where it lands in its stream return nil } @@ -415,6 +586,10 @@ func (s *Service) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream) er co, m, err := s.join(p.Address, &join, func() { cancel() _ = stream.Reset() + }, func() { + // closing, not resetting, lets the frames already written arrive + _ = stream.Close() + cancel() }) if err != nil { return s.refuse(ctx, w, stream, err) @@ -427,7 +602,8 @@ func (s *Service) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream) er return fmt.Errorf("write ack: %w", err) } - // the writer side: deliveries to a subscriber stream + // the writer side: deliveries to a receiving stream, and the roster a + // pending stream is named in writerDone := make(chan struct{}) go func() { defer close(writerDone) @@ -438,6 +614,24 @@ func (s *Service) handler(ctx context.Context, p p2p.Peer, stream p2p.Stream) er m.reset() return } + case <-m.ended: + // an EOS or a revocation: write what is queued — the EOS or the + // reduced roster last — then close; a peer that does not read + // within the flush deadline is reset + flushCtx, flushCancel := context.WithTimeout(ctx, s.opts.FlushTimeout) + defer flushCancel() + for { + select { + case f := <-m.queue: + if err := w.WriteMsgWithContext(flushCtx, f); err != nil { + m.reset() + return + } + default: + m.close() + return + } + } case <-ctx.Done(): return } @@ -485,20 +679,47 @@ func (s *Service) refuse(ctx context.Context, w protobuf.Writer, stream p2p.Stre } // handleFrame validates a Broadcast arriving at the broker per SWIP-74 and -// accepts, drops or punishes it. A returned error resets the stream. +// SWIP-87 and accepts, drops or punishes it. A returned error resets the stream. func (s *Service) handleFrame(co *cohort, m *member, f *pb.Broadcast) error { + principal := co.spec.Principal + admin := bytes.Equal(m.identity, principal) + co.mtx.Lock() - role, cursor := m.role, co.cursor + role := m.role + named, cued := co.rostered(m.identity), m.cued + var cursor uint64 + switch f.Kind { + case pb.Kind_DATA: + cursor = co.cursors[string(m.identity)] + case pb.Kind_ROSTER: + cursor = co.rosterCursor + } co.mtx.Unlock() - // 1. a publisher stream sends publications and heartbeats, a pending stream - // its AUTH; anything else is not what the stream may send - if role == roleSubscriber || role == rolePending && f.Kind != pb.Kind_AUTH { + // 0. a stream being closed — revoked, or ended by an EOS — is not read: what + // it still sends was sent before it read the roster or the EOS + if role == roleRevoked || role == roleEnded { + s.dropClosing(co, role) + return nil + } + // an AUTH from a spectator a roster had named on this stream, which a later + // roster no longer names, raced its revocation: the stream is closed, the + // AUTH dropped, nothing punished + if role == roleSpectator && f.Kind == pb.Kind_AUTH && !named && cued { + s.revokeSpectator(co, m) + return nil + } + // 1. a publisher stream sends publications and heartbeats — the admin's also + // its service messages — a pending stream its AUTH, a spectator stream its + // AUTH once a roster names it; anything else is not what the stream may send + if !mayEmit(role, admin, named, f.Kind) { s.count(&co.counters.wrongStream, "wrong_stream") return fmt.Errorf("%w: %w", errViolation, errWrongStream) } - // 2. a kind BPS-lite does not define is dropped - if f.Kind != pb.Kind_DATA && f.Kind != pb.Kind_AUTH { + // 2. a kind this SWIP does not define is dropped + switch f.Kind { + case pb.Kind_DATA, pb.Kind_AUTH, pb.Kind_EOS, pb.Kind_ROSTER: + default: s.count(&co.counters.unknownKind, "unknown_kind") return nil } @@ -507,65 +728,170 @@ func (s *Service) handleFrame(co *cohort, m *member, f *pb.Broadcast) error { s.count(&co.counters.wrongChallenge, "wrong_challenge") return nil } - // 4. a DATA frame below the cursor is a retransmit - if f.Kind == pb.Kind_DATA && f.Index < cursor { + // 4. a DATA or ROSTER frame below its feed's cursor is a retransmit + if (f.Kind == pb.Kind_DATA || f.Kind == pb.Kind_ROSTER) && f.Index < cursor { s.count(&co.counters.retransmit, "retransmit") return nil } - // 5. the chunk must be the principal's at the id derived from the frame; - // an AUTH is moreover the empty chunk at index 0 - if err := verify(f, co.spec.Topic, co.spec.Principal); err != nil { + // 5. the chunk must be its owner's at the id derived from the frame: the + // stream's identity for DATA and AUTH, the principal for the service kinds; + // an AUTH or EOS is moreover the empty chunk at index 0 + owner := m.identity + if service(f.Kind) { + owner = principal + } + if err := verify(f, co.spec.Topic, owner); err != nil { s.count(&co.counters.invalidSOC, "invalid_soc") return fmt.Errorf("%w: %w", errViolation, err) } + // 6. a ROSTER must decode; one that does not changes nothing + var roster [][]byte + if f.Kind == pb.Kind_ROSTER { + var ok bool + if roster, ok = rosterOf(f.Soc); !ok { + s.count(&co.counters.invalidRoster, "invalid_roster") + return nil + } + } - var overflow []*member + var resets, evicted, closes []*member co.mtx.Lock() - switch { - case f.Kind == pb.Kind_AUTH: - // the auth of a pending stream, a heartbeat on a publisher stream - if m.role == rolePending { - m.role = rolePublisher - m.authTimer.Stop() + // a roster revoking the identity, or an EOS, was accepted since the checks + if r := m.role; r == roleRevoked || r == roleEnded { + co.mtx.Unlock() + s.dropClosing(co, r) + return nil + } + switch f.Kind { + case pb.Kind_AUTH: + switch { + case m.role == roleSpectator && !co.rostered(m.identity): + // revoked since the checks + co.mtx.Unlock() + s.revokeSpectator(co, m) + return nil + case m.role != rolePublisher: + // the auth of a pending stream or of a spectator a roster names + resets, evicted = co.upgrade(m, s.opts.MaxPublisherStreams) } + // on a publisher stream, a heartbeat co.inactiveSince = time.Now() - case f.Index < co.cursor: - // raced with another publisher stream of the principal's - s.count(&co.counters.retransmit, "retransmit") - default: - co.cursor = f.Index + 1 + + case pb.Kind_DATA: + k := string(m.identity) + if f.Index < co.cursors[k] { + // raced with another publisher stream of the same identity + s.count(&co.counters.retransmit, "retransmit") + break + } + co.cursors[k] = f.Index + 1 co.inactiveSince = time.Now() s.metrics.Delivered.Inc() - // fan-out: the accepted frame goes, unchanged, into every subscriber - // stream's queue. The send is non-blocking, so that one slow subscriber - // cannot stall the cohort; a stream whose queue is full is collected - // here and reset below, after co.mtx is released — reset ends the - // stream, whose handler then calls leave, which takes s.mtx and co.mtx, - // so resetting under the lock would deadlock. - for sub := range co.members { - if sub.role != roleSubscriber { - continue + resets = co.deliver(f, m) + + case pb.Kind_ROSTER: + if f.Index < co.rosterCursor { + s.count(&co.counters.retransmit, "retransmit") + break + } + co.rosterCursor = f.Index + 1 + co.inactiveSince = time.Now() + resets, closes = co.applyRoster(f, roster, m) + + case pb.Kind_EOS: + // the channel is ended: every stream is told — pending ones too — and + // closed, and nothing more is read from any of them + co.ended = true + resets = co.deliver(f, m) + for o := range co.members { + if o.role == rolePending && !o.enqueue(f) { + resets = append(resets, o) } - select { - case sub.queue <- f: - default: - s.count(&co.counters.queueReset, "queue_reset") - overflow = append(overflow, sub) + } + closes = co.end() + for _, o := range closes { + if o.authTimer != nil { + o.authTimer.Stop() } + o.role = roleEnded } } co.mtx.Unlock() - // the resets of the overflowed streams, outside the lock (see the fan-out above) - for _, sub := range overflow { - sub.reset() + + // the resets of the overflowed streams, outside the lock (see deliver) + for _, r := range resets { + s.count(&co.counters.queueReset, "queue_reset") + r.reset() + } + for _, r := range evicted { + r.reset() + } + if f.Kind == pb.Kind_EOS { + s.endCohort(co, closes) + return nil + } + // the revoked streams are closed once the roster is written to them + for _, r := range closes { + r.end() } return nil } +// mayEmit reports whether a stream of the role may send a frame of the kind +// (step 1); named is whether the latest roster names the stream's identity. +// Kinds the broker does not define are left to step 2. +func mayEmit(r role, admin, named bool, kind pb.Kind) bool { + switch r { + case rolePublisher: + return !service(kind) || admin + case rolePending: + return kind == pb.Kind_AUTH + case roleSpectator: + return named && kind == pb.Kind_AUTH + } + return false +} + +// dropClosing counts a frame from a stream being closed, which is not read. +func (s *Service) dropClosing(co *cohort, r role) { + if r == roleEnded { + s.count(&co.counters.ended, "ended") + return + } + s.count(&co.counters.revoked, "revoked") +} + +// revokeSpectator closes a spectator whose AUTH raced the roster revoking it. +func (s *Service) revokeSpectator(co *cohort, m *member) { + co.mtx.Lock() + if m.role == roleSpectator { + m.role = roleRevoked + co.spectators-- + } + co.mtx.Unlock() + s.count(&co.counters.revoked, "revoked") + m.end() +} + +// endCohort forgets a cohort ended by its admin's EOS and closes its streams +// once their queues — the EOS last — are written. +func (s *Service) endCohort(co *cohort, ms []*member) { + s.mtx.Lock() + if s.cohorts[co.key] == co { + co.mtx.Lock() + s.removeCohort(co) + co.mtx.Unlock() + } + s.mtx.Unlock() + for _, m := range ms { + m.end() + } +} + // join attaches a stream to the cohort named by the spec, creating the // cohort if no live one has it. An error wrapping errFull means a capacity -// bound was hit. -func (s *Service) join(peer swarm.Address, join *pb.Join, reset func()) (*cohort, *member, error) { +// bound was hit. reset ends the stream abruptly, close after its queue. +func (s *Service) join(peer swarm.Address, join *pb.Join, reset, closeFn func()) (*cohort, *member, error) { key, err := join.Cohort.Marshal() if err != nil { return nil, nil, fmt.Errorf("marshal cohort spec: %w", err) @@ -586,12 +912,17 @@ func (s *Service) join(peer swarm.Address, join *pb.Join, reset func()) (*cohort return nil, nil, fmt.Errorf("too many streams per peer per cohort: %w", errFull) } - role := roleSubscriber - if bytes.Equal(join.Identity, join.Cohort.Principal) { - role = rolePending - } - co, ok := s.cohorts[k] + if ok { + // a cohort its admin's EOS ended is being forgotten: the Join creates + // a fresh one + co.mtx.Lock() + if co.ended { + s.removeCohort(co) + ok = false + } + co.mtx.Unlock() + } if !ok { if len(s.cohorts) >= s.opts.MaxBrokerCohorts { return nil, nil, fmt.Errorf("too many cohorts per broker: %w", errFull) @@ -600,6 +931,7 @@ func (s *Service) join(peer swarm.Address, join *pb.Join, reset func()) (*cohort key: k, spec: join.Cohort, members: make(map[*member]struct{}), + cursors: make(map[string]uint64), inactiveSince: time.Now(), } co.inactivity = time.AfterFunc(s.opts.InactiveTimeout, func() { s.reclaim(co) }) @@ -609,7 +941,18 @@ func (s *Service) join(peer swarm.Address, join *pb.Join, reset func()) (*cohort co.mtx.Lock() defer co.mtx.Unlock() - if role == roleSubscriber && co.subscribers >= s.opts.MaxSubsPerCohort { + + // the admin's identity is pending, outside the fan-out bound; any other is + // a spectator, within it — except that a rostered identity finding the set + // full is pending too: it receives the roster that names it and nothing + // else until its AUTH, so the bound is not bypassed by declaring a public + // address + role := roleSpectator + full := co.spectators >= s.opts.MaxSubsPerCohort + if bytes.Equal(join.Identity, join.Cohort.Principal) || full && co.rostered(join.Identity) { + role = rolePending + } + if role == roleSpectator && full { if len(co.members) == 0 { s.removeCohort(co) } @@ -625,15 +968,37 @@ func (s *Service) join(peer swarm.Address, join *pb.Join, reset func()) (*cohort } m := &member{ role: role, + identity: join.Identity, challenge: challenge, queue: make(chan *pb.Broadcast, s.opts.QueueCap), + ended: make(chan struct{}), } - var once sync.Once + var once, endOnce sync.Once m.reset = func() { once.Do(reset) } + m.close = func() { once.Do(closeFn) } + // a stream being closed has the flush deadline to take its queue, the write + // in progress included; then it is reset, unless it closed in time + m.end = func() { + endOnce.Do(func() { + close(m.ended) + time.AfterFunc(s.opts.FlushTimeout, m.reset) + }) + } if role == rolePending { m.authTimer = time.AfterFunc(s.opts.AuthWaitTimeout, func() { s.authTimeout(peer, co, m) }) + // a rostered identity receives the roster that names it at once: its cue + if !bytes.Equal(join.Identity, join.Cohort.Principal) { + m.queue <- co.rosterFrame + m.given = co.rosterFrame + } } else { - co.subscribers++ + co.spectators++ + // the latest roster is the first delivery on a receiving stream; one + // naming the identity is the cue for its AUTH + if co.rosterFrame != nil { + m.queue <- co.rosterFrame + m.cued = co.rostered(join.Identity) + } } co.members[m] = struct{}{} @@ -668,8 +1033,8 @@ func (s *Service) leave(peer swarm.Address, co *cohort, m *member) { if m.authTimer != nil { m.authTimer.Stop() } - if m.role == roleSubscriber { - co.subscribers-- + if m.role == roleSpectator { + co.spectators-- } if len(co.members) == 0 && s.cohorts[co.key] == co { s.removeCohort(co) @@ -712,12 +1077,18 @@ func (s *Service) reclaim(co *cohort) { func (s *Service) authTimeout(peer swarm.Address, co *cohort, m *member) { co.mtx.Lock() pending := m.role == rolePending + if pending { + // an AUTH that arrives now is not accepted + m.role = roleRevoked + } co.mtx.Unlock() if !pending { return } s.count(&co.counters.authTimeout, "auth_timeout") - if s.blocklister != nil { + // a rostered identity admitted past a full fan-out set that does not + // authenticate has cost what a FULL would have: no blocklist + if s.blocklister != nil && bytes.Equal(m.identity, co.spec.Principal) { if err := s.blocklister.Blocklist(peer, s.opts.InvalidAuthBlocklistDuration, "bps auth timeout"); err != nil { s.logger.Debug("blocklist failed", "peer_address", peer, "error", err) } @@ -743,9 +1114,11 @@ func (s *Service) Counters(spec CohortSpec) (Counters, bool) { type role int const ( - roleSubscriber role = iota + 1 // receives, never publishes - rolePending // declared the principal as identity, no AUTH yet - rolePublisher // authenticated by its AUTH: may publish, receives nothing + roleSpectator role = iota + 1 // receives; sends nothing but, once a roster names it, its AUTH + rolePending // declared the principal — or a rostered address at a full fan-out set — no AUTH yet + rolePublisher // authenticated by its AUTH: publishes, and receives what others publish + roleRevoked // a publisher stream a roster dropped, or timed out pending: being closed, not read + roleEnded // a stream of a cohort its admin's EOS ended: being closed, not read ) type cohort struct { @@ -755,16 +1128,140 @@ type cohort struct { counters counters mtx sync.Mutex - cursor uint64 // the lowest DATA index accepted next + cursors map[string]uint64 // by publisher: the lowest DATA index accepted next + roster [][]byte // the latest accepted roster, the principal excluded + rosterCursor uint64 // the lowest ROSTER index accepted next + rosterFrame *pb.Broadcast // the latest accepted ROSTER frame: the first delivery inactiveSince time.Time members map[*member]struct{} - subscribers int + spectators int // the fan-out set + authSeq uint64 // orders publisher streams by their auth + ended bool // the admin's EOS was accepted } -// shutdown returns the cohort's streams, for the caller to reset. -func (co *cohort) shutdown() []*member { - co.mtx.Lock() - defer co.mtx.Unlock() +// rostered reports whether the latest roster names the address. Must be +// called with co.mtx held. +func (co *cohort) rostered(a []byte) bool { + return contains(co.roster, a) +} + +// receives reports whether deliveries are enqueued on a stream of the role. +func (r role) receives() bool { return r == roleSpectator || r == rolePublisher } + +// deliver enqueues an accepted frame on every receiving stream — for DATA, +// every one not bound to the sender's identity — and returns the streams whose +// queue was full. A ROSTER or EOS reaches the sender too: where it sits in the +// sender's stream is where it took effect, and every receiver, the admin +// included, admits by the rosters it receives. The send is +// non-blocking, so that one slow receiver cannot stall the cohort; a stream +// whose queue is full is returned and reset by the caller after co.mtx is +// released — reset ends the stream, whose handler then calls leave, which +// takes s.mtx and co.mtx, so resetting under the lock would deadlock. Must be +// called with co.mtx held. +func (co *cohort) deliver(f *pb.Broadcast, sender *member) []*member { + var overflow []*member + for m := range co.members { + if !m.role.receives() || f.Kind == pb.Kind_DATA && bytes.Equal(m.identity, sender.identity) { + continue + } + if !m.enqueue(f) { + overflow = append(overflow, m) + } + } + return overflow +} + +// upgrade makes a pending stream, or a spectator stream a roster names, a +// publisher stream, and returns the streams to reset: one whose queue was +// full, and the oldest publisher streams of the identity beyond the bound. +// Must be called with co.mtx held. +func (co *cohort) upgrade(m *member, maxStreams int) (resets, evicted []*member) { + if m.role == rolePending { + m.authTimer.Stop() + // a stream that starts receiving gets the latest roster first, unless + // it was given it as its cue + if co.rosterFrame != nil && m.given != co.rosterFrame && !m.enqueue(co.rosterFrame) { + resets = append(resets, m) + } + } else { + // a spectator leaves the fan-out set; it has the roster already + co.spectators-- + } + m.role = rolePublisher + co.authSeq++ + m.authSeq = co.authSeq + + var same []*member + for o := range co.members { + if o.role == rolePublisher && bytes.Equal(o.identity, m.identity) { + same = append(same, o) + } + } + for len(same) > maxStreams { + oldest := 0 + for i, o := range same { + if o.authSeq < same[oldest].authSeq { + oldest = i + } + } + evicted = append(evicted, same[oldest]) + same = append(same[:oldest], same[oldest+1:]...) + } + return resets, evicted +} + +// applyRoster makes an accepted roster the cohort's: it is delivered — to its +// sender too — and every publisher stream of an identity it drops is revoked: +// written the roster, then closed, what it sends meanwhile dropped; the +// cursors of dropped identities are forgotten. A spectator it names may send +// its AUTH. It returns the streams to reset for a full queue and the revoked +// streams to close. Must be called with co.mtx held. +func (co *cohort) applyRoster(f *pb.Broadcast, roster [][]byte, sender *member) (resets, closes []*member) { + principal := co.spec.Principal + co.roster = roster + co.rosterFrame = f + resets = co.deliver(f, sender) + for m := range co.members { + if bytes.Equal(m.identity, principal) { + continue + } + switch { + case m.role == roleSpectator && co.rostered(m.identity): + // delivered above: the cue for its AUTH + m.cued = true + case m.role == rolePending && co.rostered(m.identity): + // a rostered pending stream is given the latest roster + m.given = f + if !m.enqueue(f) { + resets = append(resets, m) + } + case m.role == rolePending: + // revoked before its auth: told, then closed + m.authTimer.Stop() + m.role = roleRevoked + if !m.enqueue(f) { + resets = append(resets, m) + continue + } + closes = append(closes, m) + case m.role == rolePublisher && !co.rostered(m.identity): + // revoked: it was delivered this roster above; closed once that + // is written + m.role = roleRevoked + closes = append(closes, m) + } + } + for k := range co.cursors { + if a := []byte(k); !bytes.Equal(a, principal) && !co.rostered(a) { + delete(co.cursors, k) + } + } + return resets, closes +} + +// end returns the cohort's streams for an EOS to close. Must be called with +// co.mtx held. +func (co *cohort) end() []*member { ms := make([]*member, 0, len(co.members)) for m := range co.members { ms = append(ms, m) @@ -772,11 +1269,37 @@ func (co *cohort) shutdown() []*member { return ms } -// member is one stream attached to a cohort. role is guarded by cohort.mtx. +// shutdown returns the cohort's streams, for the caller to reset. +func (co *cohort) shutdown() []*member { + co.mtx.Lock() + defer co.mtx.Unlock() + return co.end() +} + +// member is one stream attached to a cohort. role and authSeq are guarded by +// cohort.mtx. type member struct { role role + identity []byte challenge []byte queue chan *pb.Broadcast + ended chan struct{} // closed by an EOS: write the queue, then close reset func() + close func() + end func() authTimer *time.Timer + authSeq uint64 + cued bool // a roster naming the identity was enqueued on this spectator stream + given *pb.Broadcast // the roster last enqueued on a rostered pending stream +} + +// enqueue adds a frame to the stream's outbound queue, and reports false if +// the queue is full. +func (m *member) enqueue(f *pb.Broadcast) bool { + select { + case m.queue <- f: + return true + default: + return false + } } diff --git a/pkg/bps/bps_fuzz_test.go b/pkg/bps/bps_fuzz_test.go index 6b29965ad29..87d813e75cf 100644 --- a/pkg/bps/bps_fuzz_test.go +++ b/pkg/bps/bps_fuzz_test.go @@ -5,6 +5,7 @@ package bps import ( + "bytes" "testing" "github.com/ethersphere/bee/v2/pkg/bps/pb" @@ -14,9 +15,11 @@ import ( "github.com/ethersphere/bee/v2/pkg/swarm" ) -// FuzzVerify fuzzes the peer-supplied Broadcast validation path used by BPS. -// It exercises malformed frames and valid signed chunks and asserts the verifier -// never panics on hostile input. +// FuzzVerify fuzzes the peer-supplied Broadcast validation paths used by BPS — +// verify at a known owner, and recoverOwner at a subscriber — over every kind. +// It exercises malformed frames and valid signed chunks, asserts neither +// panics on hostile input, and that a chunk valid at an owner recovers that +// owner. func FuzzVerify(f *testing.F) { priv, err := crypto.GenerateSecp256k1Key() if err != nil { @@ -62,18 +65,53 @@ func FuzzVerify(f *testing.F) { f.Fatal("auth seed not the empty auth layout") } + rosterID, err := ID(KindRoster, topic, challenge, 0) + if err != nil { + f.Fatal(err) + } + rosterPayload, err := RosterPayload([][]byte{principal.Bytes()}) + if err != nil { + f.Fatal(err) + } + rosterCAC, err := cac.New(rosterPayload) + if err != nil { + f.Fatal(err) + } + rosterChunk, err := soc.New(rosterID, rosterCAC).Sign(signer) + if err != nil { + f.Fatal(err) + } + + eosID, err := ID(KindEOS, topic, challenge, 0) + if err != nil { + f.Fatal(err) + } + eosChunk, err := soc.New(eosID, authPayload).Sign(signer) + if err != nil { + f.Fatal(err) + } + f.Add(validChunk.Data(), uint8(pb.Kind_DATA), challenge, uint64(0), topic, principal.Bytes()) + f.Add(rosterChunk.Data(), uint8(pb.Kind_ROSTER), challenge, uint64(0), topic, principal.Bytes()) + f.Add(eosChunk.Data(), uint8(pb.Kind_EOS), challenge, uint64(0), topic, principal.Bytes()) f.Add(authChunk.Data(), uint8(pb.Kind_AUTH), challenge, uint64(0), topic, principal.Bytes()) f.Add([]byte{}, uint8(pb.Kind_DATA), make([]byte, ChallengeSize), uint64(0), topic, principal.Bytes()) f.Add(make([]byte, swarm.HashSize), uint8(pb.Kind_DATA), []byte{}, uint64(0), topic[:31], principal.Bytes()[:19]) f.Fuzz(func(t *testing.T, socData []byte, kind uint8, challenge []byte, index uint64, topic []byte, principal []byte) { - if kind > uint8(pb.Kind_AUTH) { + if kind > uint8(pb.Kind_ROSTER) { kind = uint8(pb.Kind_KIND_UNSPECIFIED) } - if err := verify(&pb.Broadcast{Soc: socData, Kind: pb.Kind(kind), Challenge: challenge, Index: index}, topic, principal); err != nil { + frame := &pb.Broadcast{Soc: socData, Kind: pb.Kind(kind), Challenge: challenge, Index: index} + owner, recErr := recoverOwner(frame, topic) + _, _ = rosterOf(socData) + if err := verify(frame, topic, principal); err != nil { return } + // a chunk valid at an owner recovers that owner + if recErr != nil || !bytes.Equal(owner, principal) { + t.Fatalf("valid at %x, recovered %x (%v)", principal, owner, recErr) + } if len(topic) != swarm.HashSize || len(principal) != crypto.AddressSize || len(challenge) != ChallengeSize { return @@ -85,3 +123,45 @@ func FuzzVerify(f *testing.F) { } }) } + +// FuzzRoster fuzzes the decoding of a ROSTER chunk's payload, which a peer +// supplies: it never panics, every address it returns is 20 bytes, and what +// it returns encodes and decodes to itself. +func FuzzRoster(f *testing.F) { + off := swarm.HashSize + swarm.SocSignatureSize + swarm.SpanSize + header := make([]byte, off) + one, err := RosterPayload([][]byte{make([]byte, crypto.AddressSize)}) + if err != nil { + f.Fatal(err) + } + f.Add(append(append([]byte{}, header...), one...)) + f.Add(header) + f.Add([]byte{}) + f.Add(append(append([]byte{}, header...), 0x0a, 0x02, 0x01, 0x02)) + f.Add(append(append([]byte{}, header...), 0x0a, 0xff, 0xff, 0xff, 0xff, 0x0f)) + + f.Fuzz(func(t *testing.T, chunk []byte) { + roster, err := ParseRoster(chunk) + if err != nil { + return + } + for _, a := range roster { + if len(a) != crypto.AddressSize { + t.Fatalf("address of %d bytes", len(a)) + } + } + payload, err := RosterPayload(roster) + if err != nil { + t.Fatal(err) + } + again, err := ParseRoster(append(append([]byte{}, chunk[:off]...), payload...)) + if err != nil || len(again) != len(roster) { + t.Fatalf("round trip: %d of %d (%v)", len(again), len(roster), err) + } + for i := range roster { + if !bytes.Equal(again[i], roster[i]) { + t.Fatalf("round trip: entry %d differs", i) + } + } + }) +} diff --git a/pkg/bps/bps_test.go b/pkg/bps/bps_test.go index 0e1785e2a6c..7f91183346e 100644 --- a/pkg/bps/bps_test.go +++ b/pkg/bps/bps_test.go @@ -28,16 +28,21 @@ import ( const waitFor = 5 * time.Second type blocklister struct { - mu sync.Mutex - peers map[string]string + mu sync.Mutex + peers map[string]string + durations map[string]time.Duration } func (b *blocklister) NetworkStatus() p2p.NetworkStatus { return p2p.NetworkStatusAvailable } -func (b *blocklister) Blocklist(overlay swarm.Address, _ time.Duration, reason string) error { +func (b *blocklister) Blocklist(overlay swarm.Address, d time.Duration, reason string) error { b.mu.Lock() defer b.mu.Unlock() b.peers[overlay.ByteString()] = reason + if b.durations == nil { + b.durations = make(map[string]time.Duration) + } + b.durations[overlay.ByteString()] = d return nil } @@ -486,12 +491,21 @@ func TestInvalidDelivery(t *testing.T) { }}, } rec := streamtest.New(streamtest.WithProtocols(broker), streamtest.WithBaseAddr(swarm.RandAddress(t))) - c := bps.New(rec, nil, false, log.Noop, bps.Options{}) + bl := &blocklister{peers: make(map[string]string)} + c := bps.New(rec, bl, false, log.Noop, bps.Options{}) sub := e.join(c, swarm.RandAddress(t).Bytes()[:20]) expectDone(t, sub) if err := sub.Err(); !errors.Is(err, bps.ErrInvalidDelivery) { t.Fatalf("got %v, want %v", err, bps.ErrInvalidDelivery) } + // the broker is blocklisted for the longest period there is: forever + eventually(t, func() bool { return bl.blocklisted(e.brokerAddr) }) + bl.mu.Lock() + d := bl.durations[e.brokerAddr.ByteString()] + bl.mu.Unlock() + if d != 0 { + t.Fatalf("broker blocklisted for %v, want forever (0)", d) + } } func TestJoinRejected(t *testing.T) { diff --git a/pkg/bps/export_test.go b/pkg/bps/export_test.go index 58f68d2fa7e..148efd2f87c 100644 --- a/pkg/bps/export_test.go +++ b/pkg/bps/export_test.go @@ -15,3 +15,7 @@ var ( ErrNotPublisher = errNotPublisher ErrInvalidDelivery = errInvalidDelivery ) + +var ( + ErrChannelEnded = errChannelEnded +) diff --git a/pkg/bps/flush_test.go b/pkg/bps/flush_test.go new file mode 100644 index 00000000000..1099c8474bb --- /dev/null +++ b/pkg/bps/flush_test.go @@ -0,0 +1,419 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package bps + +import ( + "bytes" + "context" + "errors" + "io" + "sync" + "testing" + "time" + + "github.com/coreos/go-semver/semver" + "github.com/ethersphere/bee/v2/pkg/bps/pb" + "github.com/ethersphere/bee/v2/pkg/cac" + "github.com/ethersphere/bee/v2/pkg/crypto" + "github.com/ethersphere/bee/v2/pkg/log" + "github.com/ethersphere/bee/v2/pkg/p2p" + "github.com/ethersphere/bee/v2/pkg/p2p/protobuf" + "github.com/ethersphere/bee/v2/pkg/soc" + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +var errFlushStreamReset = errors.New("flush stream reset") + +// flushStream is a broker-side stream whose peer is driven by the test: the +// peer's frames are written into a pipe the broker reads, and the broker's +// writes land in a buffer — or, once stalled, block as on a transport whose +// peer has stopped reading, until the stream is reset or closed. +type flushStream struct { + r *io.PipeReader + pw *io.PipeWriter + + mu sync.Mutex + out bytes.Buffer + stall bool + stalled chan struct{} // closed when a write blocks + unblock chan struct{} // closed by Reset or Close + + reset, closed chan struct{} + stallOnce, unblockOnce sync.Once + resetOnce, closedOnce sync.Once +} + +func newFlushStream() *flushStream { + r, pw := io.Pipe() + return &flushStream{ + r: r, pw: pw, + stalled: make(chan struct{}), + unblock: make(chan struct{}), + reset: make(chan struct{}), + closed: make(chan struct{}), + } +} + +func (s *flushStream) Read(p []byte) (int, error) { return s.r.Read(p) } + +func (s *flushStream) Write(p []byte) (int, error) { + s.mu.Lock() + if !s.stall { + defer s.mu.Unlock() + return s.out.Write(p) + } + s.mu.Unlock() + s.stallOnce.Do(func() { close(s.stalled) }) + <-s.unblock + return 0, errFlushStreamReset +} + +func (s *flushStream) end() { + s.unblockOnce.Do(func() { close(s.unblock) }) + _ = s.r.Close() +} + +func (s *flushStream) Reset() error { + s.resetOnce.Do(func() { close(s.reset) }) + s.end() + return nil +} + +func (s *flushStream) Close() error { + s.closedOnce.Do(func() { close(s.closed) }) + s.end() + return nil +} + +func (s *flushStream) FullClose() error { return s.Close() } +func (s *flushStream) Headers() p2p.Headers { return nil } +func (s *flushStream) ResponseHeaders() p2p.Headers { return nil } +func (s *flushStream) Version() (*semver.Version, error) { return nil, nil } + +// stallWrites makes every later broker write block. +func (s *flushStream) stallWrites() { + s.mu.Lock() + s.stall = true + s.mu.Unlock() +} + +// frames returns what the broker has written so far, after the Ack. +func (s *flushStream) frames(t *testing.T) []*pb.Broadcast { + t.Helper() + s.mu.Lock() + r := protobuf.NewReader(bytes.NewReader(append([]byte{}, s.out.Bytes()...))) + s.mu.Unlock() + var ack pb.Ack + if err := r.ReadMsg(&ack); err != nil { + t.Fatal(err) + } + var fs []*pb.Broadcast + for { + var f pb.Broadcast + if err := r.ReadMsg(&f); err != nil { + return fs + } + fs = append(fs, &f) + } +} + +// flushEnv runs the broker's handler over flushStreams. +type flushEnv struct { + t *testing.T + svc *Service + admin crypto.Signer + spec *pb.CohortSpec +} + +func newFlushEnv(t *testing.T, o Options) *flushEnv { + t.Helper() + svc := New(nil, nil, true, log.Noop, o) + t.Cleanup(func() { _ = svc.Close() }) + key, err := crypto.GenerateSecp256k1Key() + if err != nil { + t.Fatal(err) + } + signer := crypto.NewDefaultSigner(key) + principal, err := signer.EthereumAddress() + if err != nil { + t.Fatal(err) + } + topic := make([]byte, swarm.HashSize) + copy(topic, "bps-flush-topic") + return &flushEnv{t: t, svc: svc, admin: signer, + spec: &pb.CohortSpec{Topic: topic, Binding: pb.TopicBinding_FEED_TOPIC, Principal: principal.Bytes()}} +} + +func newSigner(t *testing.T) (crypto.Signer, []byte) { + t.Helper() + key, err := crypto.GenerateSecp256k1Key() + if err != nil { + t.Fatal(err) + } + signer := crypto.NewDefaultSigner(key) + addr, err := signer.EthereumAddress() + if err != nil { + t.Fatal(err) + } + return signer, addr.Bytes() +} + +// peer is the test's end of one stream. +type peer struct { + e *flushEnv + st *flushStream + w protobuf.Writer + signer crypto.Signer + challenge []byte + done chan struct{} // the handler returned +} + +// join attaches a stream declaring identity and returns once the Ack is written. +func (e *flushEnv) join(signer crypto.Signer, identity []byte) *peer { + e.t.Helper() + st := newFlushStream() + p := &peer{e: e, st: st, w: protobuf.NewWriter(st.pw), signer: signer, done: make(chan struct{})} + go func() { + defer close(p.done) + _ = e.svc.handler(context.Background(), p2p.Peer{Address: swarm.RandAddress(e.t)}, st) + }() + go func() { _ = p.w.WriteMsg(&pb.Join{Cohort: e.spec, Identity: identity}) }() + deadline := time.Now().Add(waitFlush) + for { + st.mu.Lock() + var ack pb.Ack + err := protobuf.NewReader(bytes.NewReader(st.out.Bytes())).ReadMsg(&ack) + st.mu.Unlock() + if err == nil { + if ack.Status != pb.Status_OK { + e.t.Fatalf("join: %s", ack.Status) + } + p.challenge = ack.Challenge + return p + } + if time.Now().After(deadline) { + e.t.Fatal("no ack") + } + time.Sleep(5 * time.Millisecond) + } +} + +// send signs a chunk of kind at index under the stream's challenge and writes it. +func (p *peer) send(kind pb.Kind, index uint64, payload []byte) error { + chunkID, err := id(kind, p.e.spec.Topic, p.challenge, index) + if err != nil { + return err + } + ch, err := cac.New(payload) + if err != nil { + return err + } + c, err := soc.New(chunkID, ch).Sign(p.signer) + if err != nil { + return err + } + return p.w.WriteMsg(&pb.Broadcast{Soc: c.Data(), Kind: kind, Challenge: p.challenge, Index: index}) +} + +func (p *peer) mustSend(kind pb.Kind, index uint64, payload []byte) { + p.e.t.Helper() + if err := p.send(kind, index, payload); err != nil { + p.e.t.Fatal(err) + } +} + +func (e *flushEnv) roster(admin *peer, index uint64, addrs ...[]byte) { + e.t.Helper() + payload, err := RosterPayload(addrs) + if err != nil { + e.t.Fatal(err) + } + admin.mustSend(pb.Kind_ROSTER, index, payload) +} + +func (e *flushEnv) counters() Counters { + e.t.Helper() + key, err := e.spec.Marshal() + if err != nil { + e.t.Fatal(err) + } + e.svc.mtx.Lock() + co := e.svc.cohorts[string(key)] + e.svc.mtx.Unlock() + if co == nil { + e.t.Fatal("no live cohort") + } + return co.counters.snapshot() +} + +const waitFlush = 5 * time.Second + +func waitClosed(t *testing.T, c <-chan struct{}, what string) { + t.Helper() + select { + case <-c: + case <-time.After(waitFlush): + t.Fatalf("%s: timed out", what) + } +} + +func notClosed(t *testing.T, c <-chan struct{}, what string) { + t.Helper() + select { + case <-c: + t.Fatalf("%s: unexpected", what) + default: + } +} + +func eventuallyTrue(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(waitFlush) + for !cond() { + if time.Now().After(deadline) { + t.Fatal("condition not met") + } + time.Sleep(5 * time.Millisecond) + } +} + +// TestFlushDeadline covers the flush deadline of a stream being closed — by +// an EOS or by a revocation — on its write side, where the peer has stopped +// reading and the broker's write blocks, and on its read side, where the peer +// goes on sending or never closes. +func TestFlushDeadline(t *testing.T) { + t.Parallel() + const flush = 300 * time.Millisecond + + t.Run("write: an EOS resets a stream whose peer stopped reading, at the deadline", func(t *testing.T) { + t.Parallel() + e := newFlushEnv(t, Options{FlushTimeout: flush}) + admin := e.join(e.admin, e.spec.Principal) + admin.mustSend(pb.Kind_AUTH, 0, nil) + _, other := newSigner(t) + sub := e.join(nil, other) + sub.st.stallWrites() + + // the update blocks the subscriber's writer mid-write, outside any flush + admin.mustSend(pb.Kind_DATA, 0, []byte("stuck")) + waitClosed(t, sub.st.stalled, "the write did not block") + + start := time.Now() + admin.mustSend(pb.Kind_EOS, 0, nil) + waitClosed(t, sub.st.reset, "the stalled stream was not reset") + if elapsed := time.Since(start); elapsed < flush*3/4 { + t.Fatalf("reset after %v, before the flush deadline %v", elapsed, flush) + } + waitClosed(t, sub.done, "the stalled stream's handler did not return") + notClosed(t, sub.st.closed, "the stalled stream closed") + + // the sender, which reads, takes its EOS and is closed, not reset + waitClosed(t, admin.st.closed, "the admin's stream was not closed") + notClosed(t, admin.st.reset, "the admin's stream reset") + waitClosed(t, admin.done, "the admin's handler did not return") + fs := admin.st.frames(t) + if len(fs) == 0 || fs[len(fs)-1].Kind != pb.Kind_EOS { + t.Fatalf("the admin's stream did not end on its EOS: %v", fs) + } + }) + + t.Run("write: a revocation resets a publisher stream whose peer stopped reading, at the deadline", func(t *testing.T) { + t.Parallel() + e := newFlushEnv(t, Options{FlushTimeout: flush}) + admin := e.join(e.admin, e.spec.Principal) + admin.mustSend(pb.Kind_AUTH, 0, nil) + seatSigner, seatAddr := newSigner(t) + e.roster(admin, 0, seatAddr) + seat := e.join(seatSigner, seatAddr) + eventuallyTrue(t, func() bool { return len(seat.st.frames(t)) > 0 }) // the cue + seat.mustSend(pb.Kind_AUTH, 0, nil) + eventuallyTrue(t, func() bool { + fs := admin.st.frames(t) + return len(fs) > 0 && fs[len(fs)-1].Kind == pb.Kind_ROSTER + }) + seat.st.stallWrites() + + admin.mustSend(pb.Kind_DATA, 0, []byte("stuck")) + waitClosed(t, seat.st.stalled, "the write did not block") + + start := time.Now() + e.roster(admin, 1) + waitClosed(t, seat.st.reset, "the stalled revoked stream was not reset") + if elapsed := time.Since(start); elapsed < flush*3/4 { + t.Fatalf("reset after %v, before the flush deadline %v", elapsed, flush) + } + waitClosed(t, seat.done, "the revoked stream's handler did not return") + notClosed(t, admin.st.reset, "the admin's stream reset") + }) + + t.Run("read: a revoked peer that keeps sending is closed, not reset, its frames dropped", func(t *testing.T) { + t.Parallel() + e := newFlushEnv(t, Options{FlushTimeout: 10 * flush}) + admin := e.join(e.admin, e.spec.Principal) + admin.mustSend(pb.Kind_AUTH, 0, nil) + seatSigner, seatAddr := newSigner(t) + e.roster(admin, 0, seatAddr) + seat := e.join(seatSigner, seatAddr) + eventuallyTrue(t, func() bool { return len(seat.st.frames(t)) > 0 }) + seat.mustSend(pb.Kind_AUTH, 0, nil) + eventuallyTrue(t, func() bool { + fs := admin.st.frames(t) + return len(fs) > 0 && fs[len(fs)-1].Kind == pb.Kind_ROSTER + }) + + // the seat floods its stream with updates until it is closed + flooding := make(chan struct{}) + go func() { + defer close(flooding) + for i := uint64(0); ; i++ { + if seat.send(pb.Kind_DATA, i, []byte("flood")) != nil { + return + } + } + }() + eventuallyTrue(t, func() bool { return e.counters().Retransmit == 0 && len(admin.st.frames(t)) > 2 }) + + start := time.Now() + e.roster(admin, 1) + waitClosed(t, seat.st.closed, "the revoked stream was not closed") + if elapsed := time.Since(start); elapsed >= 10*flush { + t.Fatalf("closed after %v: the flood held it to the deadline", elapsed) + } + notClosed(t, seat.st.reset, "the revoked stream reset") + waitClosed(t, seat.done, "the revoked stream's handler did not return") + waitClosed(t, flooding, "the flood did not stop") + + c := e.counters() + if c.WrongStream != 0 || c.InvalidSOC != 0 { + t.Fatalf("a revoked peer's frames punished: %+v", c) + } + fs := seat.st.frames(t) + if last := fs[len(fs)-1]; last.Kind != pb.Kind_ROSTER || last.Index != 1 { + t.Fatalf("the revoked stream did not end on the reduced roster: %s %d", last.Kind, last.Index) + } + }) + + t.Run("read: a silent peer that never closes its side does not hold an ended stream", func(t *testing.T) { + t.Parallel() + e := newFlushEnv(t, Options{FlushTimeout: 10 * flush}) + admin := e.join(e.admin, e.spec.Principal) + admin.mustSend(pb.Kind_AUTH, 0, nil) + _, other := newSigner(t) + sub := e.join(nil, other) + + start := time.Now() + admin.mustSend(pb.Kind_EOS, 0, nil) + waitClosed(t, sub.st.closed, "the ended stream was not closed") + waitClosed(t, sub.done, "the ended stream's handler did not return") + if elapsed := time.Since(start); elapsed >= 10*flush { + t.Fatalf("closed after %v: the silent peer held it to the deadline", elapsed) + } + notClosed(t, sub.st.reset, "the ended stream reset") + fs := sub.st.frames(t) + if len(fs) != 1 || fs[0].Kind != pb.Kind_EOS { + t.Fatalf("the ended stream got %v, want the EOS alone", fs) + } + }) +} diff --git a/pkg/bps/id.go b/pkg/bps/id.go index 54801b202b7..619b93903e3 100644 --- a/pkg/bps/id.go +++ b/pkg/bps/id.go @@ -24,9 +24,13 @@ const ChallengeSize = 32 var servicePrefix = []byte("bps-service:v1") // errInvalidSOC is returned for a chunk that does not have the id derived -// from its frame, or does not validate as the principal's single-owner chunk. +// from its frame, or does not validate as its owner's single-owner chunk. var errInvalidSOC = errors.New("bps: invalid soc") +// errInvalidRoster is returned for a ROSTER chunk whose payload does not +// decode as a Roster of 20-byte addresses. +var errInvalidRoster = errors.New("bps: invalid roster") + // Kind is what a chunk on the wire is. type Kind int @@ -34,8 +38,14 @@ const ( KindUnspecified Kind = iota KindData // an update of the stream's session feed // KindAuth is the empty chunk at index 0 of the session's AUTH feed: it authenticates - // a pending stream, is a heartbeat on a publisher stream, and is never delivered. + // a pending stream, or upgrades in place a spectator stream on which a roster naming + // its identity was delivered; it is a heartbeat on a publisher stream, and is never + // delivered. KindAuth + // KindEOS is the empty chunk at index 0 of the admin's EOS feed: the channel is ended. + KindEOS + // KindRoster is the admin's publisher set as of its index; its payload is a Roster. + KindRoster ) func (k Kind) proto() pb.Kind { @@ -44,6 +54,10 @@ func (k Kind) proto() pb.Kind { return pb.Kind_DATA case KindAuth: return pb.Kind_AUTH + case KindEOS: + return pb.Kind_EOS + case KindRoster: + return pb.Kind_ROSTER default: return pb.Kind_KIND_UNSPECIFIED } @@ -79,44 +93,114 @@ func id(kind pb.Kind, topic, challenge []byte, index uint64) ([]byte, error) { } // Verify checks that the frame fields kind, challenge and index give the id of -// the chunk, and that the chunk validates as the principal's single-owner chunk -// at keccak256(id | principal). An AUTH chunk must moreover be at index 0 and empty. -func Verify(kind Kind, challenge []byte, index uint64, chunk, topic, principal []byte) error { - return verify(&pb.Broadcast{Soc: chunk, Kind: kind.proto(), Challenge: challenge, Index: index}, topic, principal) +// the chunk, and that the chunk validates as owner's single-owner chunk at +// keccak256(id | owner). An AUTH or EOS chunk must moreover be at index 0 and empty. +func Verify(kind Kind, challenge []byte, index uint64, chunk, topic, owner []byte) error { + return verify(&pb.Broadcast{Soc: chunk, Kind: kind.proto(), Challenge: challenge, Index: index}, topic, owner) +} + +// service reports whether the kind is one of the admin's service kinds, whose +// owner is forced to the principal. +func service(kind pb.Kind) bool { + return kind == pb.Kind_EOS || kind == pb.Kind_ROSTER +} + +// empty reports whether the kind is an empty chunk at index 0 of its feed. +func empty(kind pb.Kind) bool { + return kind == pb.Kind_AUTH || kind == pb.Kind_EOS } -func verify(f *pb.Broadcast, topic, principal []byte) error { +// checkFrame runs the checks that need no owner: the challenge length, the +// index, and the id the frame fields derive. +func checkFrame(f *pb.Broadcast, topic []byte) ([]byte, error) { if len(f.Challenge) != ChallengeSize { - return fmt.Errorf("challenge length %d: %w", len(f.Challenge), errInvalidSOC) + return nil, fmt.Errorf("challenge length %d: %w", len(f.Challenge), errInvalidSOC) } // the last index is one no feed has: the cursor, set to index+1, never wraps if f.Index == math.MaxUint64 { - return fmt.Errorf("index %d: %w", f.Index, errInvalidSOC) + return nil, fmt.Errorf("index %d: %w", f.Index, errInvalidSOC) } - // a session has one AUTH chunk, at index 0 of its AUTH feed - if f.Kind == pb.Kind_AUTH && f.Index != 0 { - return fmt.Errorf("auth chunk at index %d: %w", f.Index, errInvalidSOC) + // a session has one AUTH chunk, and a channel one EOS, at index 0 of its feed + if empty(f.Kind) && f.Index != 0 { + return nil, fmt.Errorf("%s chunk at index %d: %w", f.Kind, f.Index, errInvalidSOC) } want, err := id(f.Kind, topic, f.Challenge, f.Index) if err != nil { - return err + return nil, err } if len(f.Soc) < swarm.HashSize || !bytes.Equal(f.Soc[:swarm.HashSize], want) { - return fmt.Errorf("id not derived from frame: %w", errInvalidSOC) + return nil, fmt.Errorf("id not derived from frame: %w", errInvalidSOC) } - addr, err := soc.CreateAddress(want, principal) + if empty(f.Kind) { + // span 0, no payload + if len(f.Soc) != swarm.HashSize+swarm.SocSignatureSize+swarm.SpanSize || + binary.LittleEndian.Uint64(f.Soc[swarm.HashSize+swarm.SocSignatureSize:]) != 0 { + return nil, fmt.Errorf("%s chunk not empty: %w", f.Kind, errInvalidSOC) + } + } + return want, nil +} + +func verify(f *pb.Broadcast, topic, owner []byte) error { + want, err := checkFrame(f, topic) + if err != nil { + return err + } + addr, err := soc.CreateAddress(want, owner) if err != nil { return err } if !soc.Valid(swarm.NewChunk(addr, f.Soc)) { - return fmt.Errorf("not the principal's soc: %w", errInvalidSOC) + return fmt.Errorf("not the owner's soc: %w", errInvalidSOC) } - if f.Kind == pb.Kind_AUTH { - // span 0, no payload - if len(f.Soc) != swarm.HashSize+swarm.SocSignatureSize+swarm.SpanSize || - binary.LittleEndian.Uint64(f.Soc[swarm.HashSize+swarm.SocSignatureSize:]) != 0 { - return fmt.Errorf("auth chunk not empty: %w", errInvalidSOC) + return nil +} + +// recoverOwner checks the frame as verify does and returns the owner the +// chunk's signature recovers: the chunk is then valid at keccak256(id | owner). +func recoverOwner(f *pb.Broadcast, topic []byte) ([]byte, error) { + if _, err := checkFrame(f, topic); err != nil { + return nil, err + } + s, err := soc.FromChunk(swarm.NewChunk(swarm.ZeroAddress, f.Soc)) + if err != nil { + return nil, fmt.Errorf("%w: %w", errInvalidSOC, err) + } + return s.OwnerAddress(), nil +} + +// decodeRoster returns the publisher set a validated ROSTER chunk carries. +func decodeRoster(chunk []byte) ([][]byte, error) { + off := swarm.HashSize + swarm.SocSignatureSize + swarm.SpanSize + if len(chunk) < off { + return nil, errInvalidRoster + } + var r pb.Roster + if err := r.Unmarshal(chunk[off:]); err != nil { + return nil, fmt.Errorf("%w: %w", errInvalidRoster, err) + } + for _, a := range r.Publishers { + if len(a) != crypto.AddressSize { + return nil, fmt.Errorf("address length %d: %w", len(a), errInvalidRoster) } } - return nil + return r.Publishers, nil +} + +// rosterOf returns the publisher set of a validated ROSTER chunk, and false if +// its payload does not decode: the admin's mistake, which is dropped, not an error. +func rosterOf(chunk []byte) ([][]byte, bool) { + roster, err := decodeRoster(chunk) + return roster, err == nil +} + +// ParseRoster returns the publisher set a ROSTER chunk's payload names, or an +// error if it does not decode as a Roster of 20-byte addresses. +func ParseRoster(chunk []byte) ([][]byte, error) { + return decodeRoster(chunk) +} + +// RosterPayload returns the payload of a ROSTER chunk naming the publishers. +func RosterPayload(publishers [][]byte) ([]byte, error) { + return (&pb.Roster{Publishers: publishers}).Marshal() } diff --git a/pkg/bps/metrics.go b/pkg/bps/metrics.go index 0473653e6dd..2e3fafcf9e8 100644 --- a/pkg/bps/metrics.go +++ b/pkg/bps/metrics.go @@ -14,13 +14,16 @@ import ( // Counters are the per-cohort counters of the silent outcomes SWIP-74 makes a // broker expose. type Counters struct { - WrongStream uint64 // a frame from a subscriber stream - UnknownKind uint64 // a frame of a kind BPS-lite does not define + WrongStream uint64 // a frame its stream may not send + UnknownKind uint64 // a frame of a kind the broker does not define WrongChallenge uint64 // a frame whose challenge is not its stream's InvalidSOC uint64 // a chunk that does not validate at the derived id AuthTimeout uint64 // a pending stream disconnected at the auth timeout Retransmit uint64 // a DATA frame below the cursor - QueueReset uint64 // a subscriber stream reset for a full queue + QueueReset uint64 // a receiving stream reset for a full queue + InvalidRoster uint64 // a valid ROSTER chunk whose payload does not decode + Revoked uint64 // a frame from a revoked stream, dropped unread + Ended uint64 // a frame from a stream of a cohort an EOS ended, dropped unread } type counters struct { @@ -31,6 +34,9 @@ type counters struct { authTimeout atomic.Uint64 retransmit atomic.Uint64 queueReset atomic.Uint64 + invalidRoster atomic.Uint64 + revoked atomic.Uint64 + ended atomic.Uint64 } func (c *counters) snapshot() Counters { @@ -42,6 +48,9 @@ func (c *counters) snapshot() Counters { AuthTimeout: c.authTimeout.Load(), Retransmit: c.retransmit.Load(), QueueReset: c.queueReset.Load(), + InvalidRoster: c.invalidRoster.Load(), + Revoked: c.revoked.Load(), + Ended: c.ended.Load(), } } @@ -65,7 +74,7 @@ func newMetrics() metrics { Namespace: m.Namespace, Subsystem: subsystem, Name: "accepted_updates", - Help: "DATA updates accepted by the broker.", + Help: "DATA updates accepted by the broker, from every publisher.", }), Cohorts: prometheus.NewGauge(prometheus.GaugeOpts{ Namespace: m.Namespace, diff --git a/pkg/bps/multipublisher_test.go b/pkg/bps/multipublisher_test.go new file mode 100644 index 00000000000..bfe053d9922 --- /dev/null +++ b/pkg/bps/multipublisher_test.go @@ -0,0 +1,538 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package bps_test + +import ( + "bytes" + "context" + "errors" + "testing" + "time" + + "github.com/ethersphere/bee/v2/pkg/bps" + "github.com/ethersphere/bee/v2/pkg/bps/pb" + "github.com/ethersphere/bee/v2/pkg/crypto" + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +// seat is a key the admin may roster. +type seat struct { + signer crypto.Signer + addr []byte +} + +func (e *env) newSeat() seat { + e.t.Helper() + key, err := crypto.GenerateSecp256k1Key() + if err != nil { + e.t.Fatal(err) + } + signer := crypto.NewDefaultSigner(key) + addr, err := signer.EthereumAddress() + if err != nil { + e.t.Fatal(err) + } + return seat{signer: signer, addr: addr.Bytes()} +} + +// joinAs joins as the seat's address from a fresh overlay. +func (e *env) joinAs(st seat) (bps.Session, swarm.Address) { + e.t.Helper() + overlay := swarm.RandAddress(e.t) + c, _ := e.client(overlay) + return e.join(c, st.addr), overlay +} + +// grant publishes a roster and waits until the broker has accepted it. +func (e *env) grant(admin bps.Session, index uint64, seats ...seat) { + e.t.Helper() + probe := e.subscribe() + e.roster(admin, index, seats...) + for { + m := <-probe.Messages() + if m.Kind == bps.KindRoster && m.Index == index { + return + } + } +} + +// publishAs signs a chunk with the seat's key under the session's challenge +// and publishes it. +func (e *env) publishAs(st seat, s bps.Session, kind bps.Kind, index uint64, payload []byte) []byte { + e.t.Helper() + c := e.chunk(st.signer, kind, s.Challenge(), index, payload) + if err := s.Publish(context.Background(), kind, index, c); err != nil { + e.t.Fatal(err) + } + return c +} + +// roster publishes the admin's roster at index on the admin's publisher stream. +func (e *env) roster(admin bps.Session, index uint64, seats ...seat) []byte { + e.t.Helper() + addrs := make([][]byte, 0, len(seats)) + for _, st := range seats { + addrs = append(addrs, st.addr) + } + payload, err := bps.RosterPayload(addrs) + if err != nil { + e.t.Fatal(err) + } + return e.publish(admin, bps.KindRoster, index, payload) +} + +func waitGranted(t *testing.T, s bps.Session) { + t.Helper() + select { + case <-s.Granted(): + case <-time.After(waitFor): + t.Fatal("not granted") + } +} + +func expectKind(t *testing.T, s bps.Session, kind bps.Kind, owner []byte, index uint64) bps.Message { + t.Helper() + select { + case m, ok := <-s.Messages(): + if !ok { + t.Fatalf("stream ended: %v", s.Err()) + } + if m.Kind != kind || !bytes.Equal(m.Owner, owner) || m.Index != index { + t.Fatalf("got kind %d owner %x index %d, want kind %d owner %x index %d", m.Kind, m.Owner, m.Index, kind, owner, index) + } + return m + case <-time.After(waitFor): + t.Fatalf("timed out waiting for kind %d index %d", kind, index) + } + return bps.Message{} +} + +// drain discards a session's deliveries, so that its stream keeps reading. +func drain(s bps.Session) { + go func() { + for range s.Messages() { + } + }() +} + +func TestGrant(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + sub := e.subscribe() + admin, _ := e.publisher() + + e.roster(admin, 0, st) + m := expectKind(t, sub, bps.KindRoster, e.principal, 0) + if len(m.Roster) != 1 || !bytes.Equal(m.Roster[0], st.addr) { + t.Fatalf("roster %x", m.Roster) + } + // the roster is delivered back to the admin: it admits by it from there + expectKind(t, admin, bps.KindRoster, e.principal, 0) + + // the seat joins as a spectator; the latest roster, naming it, is its cue + seatSess, _ := e.joinAs(st) + expectKind(t, seatSess, bps.KindRoster, e.principal, 0) + waitGranted(t, seatSess) + e.publishAs(st, seatSess, bps.KindAuth, 0, nil) + c0 := e.publishAs(st, seatSess, bps.KindData, 0, []byte("from the seat")) + + got := expectKind(t, sub, bps.KindData, st.addr, 0) + if !bytes.Equal(got.SOC, c0) { + t.Fatal("wrong chunk") + } + // the admin's publisher stream receives the seat's update; the seat not its own + expectKind(t, admin, bps.KindData, st.addr, 0) + a0 := e.publish(admin, bps.KindData, 0, []byte("from the admin")) + got = expectKind(t, seatSess, bps.KindData, e.principal, 0) + if !bytes.Equal(got.SOC, a0) { + t.Fatal("wrong chunk") + } + expectKind(t, sub, bps.KindData, e.principal, 0) + expectNoMessage(t, admin) +} + +// TestSeatBeforeRoster is the startup order the review found: the seat joins +// before the broker holds any roster, and is upgraded in place by its AUTH +// once the first roster names it. +func TestSeatBeforeRoster(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + sub := e.subscribe() + + seatSess, overlay := e.joinAs(st) + admin, _ := e.publisher() + drain(admin) + e.roster(admin, 0, st) + expectKind(t, seatSess, bps.KindRoster, e.principal, 0) + waitGranted(t, seatSess) + e.publishAs(st, seatSess, bps.KindAuth, 0, nil) + c := e.publishAs(st, seatSess, bps.KindData, 5, []byte("upgraded in place")) + expectKind(t, sub, bps.KindRoster, e.principal, 0) + got := expectKind(t, sub, bps.KindData, st.addr, 5) + if !bytes.Equal(got.SOC, c) { + t.Fatal("wrong chunk") + } + if e.bl.blocklisted(overlay) || e.counters().WrongStream != 0 { + t.Fatal("an upgrade in place is not a violation") + } +} + +// TestRosteredListens: a rostered address that never authenticates is a +// spectator like any other, and is not disconnected at the auth deadline. +func TestRosteredListens(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{AuthWaitTimeout: 100 * time.Millisecond}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + e.grant(admin, 0, st) + + listener, overlay := e.joinAs(st) + expectKind(t, listener, bps.KindRoster, e.principal, 0) + time.Sleep(300 * time.Millisecond) + e.publish(admin, bps.KindData, 0, []byte("heard")) + expectKind(t, listener, bps.KindData, e.principal, 0) + if e.bl.blocklisted(overlay) || e.counters().AuthTimeout != 0 { + t.Fatal("a listening rostered address is not timed out") + } +} + +func TestUnrosteredAuth(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + _, _ = e.publisher() + overlay := swarm.RandAddress(t) + w, _, ack := e.rawFrom(overlay, &pb.Join{Cohort: e.spec(), Identity: st.addr}) + if ack.Status != pb.Status_OK { + t.Fatalf("status %s", ack.Status) + } + auth := e.chunk(st.signer, bps.KindAuth, ack.Challenge, 0, nil) + if err := w.WriteMsg(&pb.Broadcast{Kind: pb.Kind_AUTH, Challenge: ack.Challenge, Soc: auth}); err != nil { + t.Fatal(err) + } + eventually(t, func() bool { return e.bl.blocklisted(overlay) && e.counters().WrongStream == 1 }) +} + +// TestRevokedSpectatorAuth: a spectator sends its AUTH after the roster naming +// it, but a roster revoking it reaches the broker first: the stream is closed, +// nothing is punished. +func TestRevokedSpectatorAuth(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + e.grant(admin, 0, st) + + overlay := swarm.RandAddress(t) + w, r, ack := e.rawFrom(overlay, &pb.Join{Cohort: e.spec(), Identity: st.addr}) + var cue pb.Broadcast + if err := r.ReadMsg(&cue); err != nil || cue.Kind != pb.Kind_ROSTER { + t.Fatalf("want the roster as the cue, got %s %v", cue.Kind, err) + } + e.grant(admin, 1) + auth := e.chunk(st.signer, bps.KindAuth, ack.Challenge, 0, nil) + _ = w.WriteMsg(&pb.Broadcast{Kind: pb.Kind_AUTH, Challenge: ack.Challenge, Soc: auth}) + eventually(t, func() bool { return e.counters().Revoked == 1 }) + if e.bl.blocklisted(overlay) || e.counters().WrongStream != 0 { + t.Fatal("an AUTH racing its revocation is not punished") + } +} + +func TestRevocation(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + sub := e.subscribe() + e.roster(admin, 0, st) + expectKind(t, sub, bps.KindRoster, e.principal, 0) + + // the seat authenticates on a raw stream, to send after its revocation + overlay := swarm.RandAddress(t) + w, r, ack := e.rawFrom(overlay, &pb.Join{Cohort: e.spec(), Identity: st.addr}) + if ack.Status != pb.Status_OK { + t.Fatalf("status %s", ack.Status) + } + var cue pb.Broadcast + if err := r.ReadMsg(&cue); err != nil || cue.Kind != pb.Kind_ROSTER { + t.Fatalf("want the roster as the cue, got %s %v", cue.Kind, err) + } + send := func(kind bps.Kind, index uint64, payload []byte) { + t.Helper() + c := e.chunk(st.signer, kind, ack.Challenge, index, payload) + f := &pb.Broadcast{Kind: pb.Kind(kind), Challenge: ack.Challenge, Index: index, Soc: c} + _ = w.WriteMsg(f) + } + send(bps.KindAuth, 0, nil) + send(bps.KindData, 0, []byte("before")) + expectKind(t, sub, bps.KindData, st.addr, 0) + + // the reduced roster revokes the seat: the stream is written that roster + // and closed; what it sends meanwhile is dropped, unpunished + e.grant(admin, 1) + send(bps.KindData, 1, []byte("in flight")) + m := expectKind(t, sub, bps.KindRoster, e.principal, 1) + if len(m.Roster) != 0 { + t.Fatalf("roster %x", m.Roster) + } + var next pb.Broadcast + if err := r.ReadMsg(&next); err != nil || next.Kind != pb.Kind_ROSTER || next.Index != 1 { + t.Fatalf("want the reduced roster, got %s %d %v", next.Kind, next.Index, err) + } + if err := r.ReadMsg(&next); err == nil { + t.Fatalf("want the stream closed, got %s", next.Kind) + } + expectNoMessage(t, sub) + if e.bl.blocklisted(overlay) || e.counters().WrongStream != 0 { + t.Fatal("a revoked peer is not punished") + } + if pos := sub.Position(); len(pos.Roster) != 0 || pos.RosterCursor != 2 { + t.Fatalf("position %+v", pos) + } + + // rejoining, the seat is a spectator that may not authenticate + spectator, _ := e.joinAs(st) + expectKind(t, spectator, bps.KindRoster, e.principal, 1) + select { + case <-spectator.Granted(): + t.Fatal("a revoked identity is not granted") + default: + } +} + +// TestRevokeInFlight: the admin revokes a seat whose update the broker accepted +// just before the roster; the admin's own session, admitting by the echoed +// roster, takes the update and is not reset. +func TestRevokeInFlight(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + e.roster(admin, 0, st) + expectKind(t, admin, bps.KindRoster, e.principal, 0) + seatSess, _ := e.joinAs(st) + drain(seatSess) + waitGranted(t, seatSess) + e.publishAs(st, seatSess, bps.KindAuth, 0, nil) + e.publishAs(st, seatSess, bps.KindData, 0, []byte("last")) + time.Sleep(50 * time.Millisecond) // the broker accepts the update first + e.roster(admin, 1) + expectKind(t, admin, bps.KindData, st.addr, 0) + expectKind(t, admin, bps.KindRoster, e.principal, 1) + select { + case <-admin.Done(): + t.Fatalf("the admin's stream ended: %v", admin.Err()) + default: + } +} + +func TestRevokedSession(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + e.grant(admin, 0, st) + seatSess, _ := e.joinAs(st) + expectKind(t, seatSess, bps.KindRoster, e.principal, 0) + waitGranted(t, seatSess) + e.publishAs(st, seatSess, bps.KindAuth, 0, nil) + + e.roster(admin, 1) + expectKind(t, seatSess, bps.KindRoster, e.principal, 1) + expectDone(t, seatSess) + if err := seatSess.Publish(context.Background(), bps.KindData, 0, nil); !errors.Is(err, bps.ErrNotPublisher) { + t.Fatalf("got %v, want %v", err, bps.ErrNotPublisher) + } +} + +func TestCursorsPerPublisher(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + sub := e.subscribe() + e.roster(admin, 0, st) + expectKind(t, sub, bps.KindRoster, e.principal, 0) + seatSess, _ := e.joinAs(st) + drain(seatSess) + waitGranted(t, seatSess) + e.publishAs(st, seatSess, bps.KindAuth, 0, nil) + + // each publisher has its own cursor + e.publish(admin, bps.KindData, 3, []byte("a3")) + expectKind(t, sub, bps.KindData, e.principal, 3) + e.publishAs(st, seatSess, bps.KindData, 0, []byte("s0")) + expectKind(t, sub, bps.KindData, st.addr, 0) + // below its own cursor: a retransmit + e.publishAs(st, seatSess, bps.KindData, 0, []byte("s0")) + eventually(t, func() bool { return e.counters().Retransmit == 1 }) + expectNoMessage(t, sub) + + pos := sub.Position() + if pos.Cursors[string(e.principal)] != 4 || pos.Cursors[string(st.addr)] != 1 { + t.Fatalf("cursors %v", pos.Cursors) + } +} + +func TestRosterFeed(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + sub := e.subscribe() + + e.roster(admin, 0, st) + expectKind(t, sub, bps.KindRoster, e.principal, 0) + // a roster below the roster cursor is a retransmit + e.roster(admin, 0, st) + eventually(t, func() bool { return e.counters().Retransmit == 1 }) + + // a valid chunk whose payload is not a roster changes nothing + e.publish(admin, bps.KindRoster, 1, []byte{0x0a, 0x02, 0x01, 0x02}) + eventually(t, func() bool { return e.counters().InvalidRoster == 1 }) + e.roster(admin, 1) + expectKind(t, sub, bps.KindRoster, e.principal, 1) + + // a late joiner receives the latest roster first + late := e.subscribe() + m := expectKind(t, late, bps.KindRoster, e.principal, 1) + if len(m.Roster) != 0 { + t.Fatalf("roster %x", m.Roster) + } +} + +func TestServiceKindsAdminOnly(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + e.grant(admin, 0, st) + + overlay := swarm.RandAddress(t) + w, r, ack := e.rawFrom(overlay, &pb.Join{Cohort: e.spec(), Identity: st.addr}) + var cue pb.Broadcast + if err := r.ReadMsg(&cue); err != nil { + t.Fatal(err) + } + auth := e.chunk(st.signer, bps.KindAuth, ack.Challenge, 0, nil) + if err := w.WriteMsg(&pb.Broadcast{Kind: pb.Kind_AUTH, Challenge: ack.Challenge, Soc: auth}); err != nil { + t.Fatal(err) + } + payload, _ := bps.RosterPayload([][]byte{st.addr}) + ros := e.chunk(st.signer, bps.KindRoster, ack.Challenge, 1, payload) + if err := w.WriteMsg(&pb.Broadcast{Kind: pb.Kind_ROSTER, Challenge: ack.Challenge, Index: 1, Soc: ros}); err != nil { + t.Fatal(err) + } + eventually(t, func() bool { return e.bl.blocklisted(overlay) && e.counters().WrongStream == 1 }) +} + +func TestEndOfStream(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{}) + admin, _ := e.publisher() + sub := e.subscribe() + second, _ := e.pending() + e.publish(admin, bps.KindData, 0, []byte("last words")) + expectKind(t, sub, bps.KindData, e.principal, 0) + + e.publish(admin, bps.KindEOS, 0, nil) + // every stream is told — the sender and a pending one included — and closed + for _, s := range []bps.Session{sub, admin, second} { + expectKind(t, s, bps.KindEOS, e.principal, 0) + expectDone(t, s) + if !errors.Is(s.Err(), bps.ErrEndOfStream) { + t.Fatalf("got %v, want %v", s.Err(), bps.ErrEndOfStream) + } + } + eventually(t, func() bool { + _, ok := e.broker.Counters(bps.CohortSpec{Topic: e.topic, Principal: e.principal}) + return !ok + }) + + // a channel the subscriber has seen end is not rejoined + pos := sub.Position() + if !pos.Ended { + t.Fatal("not ended") + } + c, _ := e.client(swarm.RandAddress(t)) + addr := make([]byte, 20) + _, err := c.Join(context.Background(), bps.JoinRequest{Broker: e.brokerAddr, Spec: bps.CohortSpec{Topic: e.topic, Principal: e.principal}, Identity: addr, Position: pos}) + if !errors.Is(err, bps.ErrChannelEnded) { + t.Fatalf("got %v, want %v", err, bps.ErrChannelEnded) + } + // a Join that does not know creates a fresh cohort + fresh := e.subscribe() + expectNoMessage(t, fresh) + if _, ok := e.broker.Counters(bps.CohortSpec{Topic: e.topic, Principal: e.principal}); !ok { + t.Fatal("no fresh cohort") + } +} + +func TestPublisherStreamsBound(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{MaxPublisherStreams: 2}) + st := e.newSeat() + admin, _ := e.publisher() + drain(admin) + e.grant(admin, 0, st) + + sess := make([]bps.Session, 0, 3) + for range 3 { + s, _ := e.joinAs(st) + drain(s) + waitGranted(t, s) + e.publishAs(st, s, bps.KindAuth, 0, nil) + sess = append(sess, s) + time.Sleep(50 * time.Millisecond) + } + // the third auth resets the oldest + expectDone(t, sess[0]) + select { + case <-sess[1].Done(): + t.Fatal("second stream reset") + case <-sess[2].Done(): + t.Fatal("third stream reset") + default: + } +} + +// TestRosteredPastFullSet: a rostered identity that finds the fan-out set full +// is pending — the roster, nothing else — and authenticates as any grantee; +// one that does not is reset at the auth deadline, not blocklisted. +func TestRosteredPastFullSet(t *testing.T) { + t.Parallel() + e := newEnv(t, bps.Options{MaxSubsPerCohort: 1, AuthWaitTimeout: 300 * time.Millisecond}) + st, idle := e.newSeat(), e.newSeat() + admin, _ := e.publisher() + drain(admin) + probe := e.subscribe() // fills the fan-out set + e.roster(admin, 0, st, idle) + expectKind(t, probe, bps.KindRoster, e.principal, 0) + + seatSess, _ := e.joinAs(st) + expectKind(t, seatSess, bps.KindRoster, e.principal, 0) + waitGranted(t, seatSess) + e.publishAs(st, seatSess, bps.KindAuth, 0, nil) + e.publishAs(st, seatSess, bps.KindData, 0, []byte("past the bound")) + expectKind(t, probe, bps.KindData, st.addr, 0) + + listener, overlay := e.joinAs(idle) + expectKind(t, listener, bps.KindRoster, e.principal, 0) + e.publish(admin, bps.KindData, 0, []byte("not for a pending stream")) + expectKind(t, probe, bps.KindData, e.principal, 0) + expectDone(t, listener) + if e.bl.blocklisted(overlay) || e.counters().AuthTimeout != 1 { + t.Fatal("a rostered pending stream times out without a blocklist") + } +} diff --git a/pkg/bps/pb/bps.pb.go b/pkg/bps/pb/bps.pb.go index 9ba6b51cdc0..a40c3e39789 100644 --- a/pkg/bps/pb/bps.pb.go +++ b/pkg/bps/pb/bps.pb.go @@ -86,18 +86,26 @@ const ( Kind_KIND_UNSPECIFIED Kind = 0 Kind_DATA Kind = 1 Kind_AUTH Kind = 2 + // a pending stream, or a spectator stream a delivered roster names; + // a heartbeat on a publisher stream; never delivered + Kind_EOS Kind = 3 + Kind_ROSTER Kind = 4 ) var Kind_name = map[int32]string{ 0: "KIND_UNSPECIFIED", 1: "DATA", 2: "AUTH", + 3: "EOS", + 4: "ROSTER", } var Kind_value = map[string]int32{ "KIND_UNSPECIFIED": 0, "DATA": 1, "AUTH": 2, + "EOS": 3, + "ROSTER": 4, } func (x Kind) String() string { @@ -275,6 +283,51 @@ func (m *Ack) GetChallenge() []byte { return nil } +// The payload of a ROSTER chunk. +type Roster struct { + Publishers [][]byte `protobuf:"bytes,1,rep,name=publishers,proto3" json:"publishers,omitempty"` +} + +func (m *Roster) Reset() { *m = Roster{} } +func (m *Roster) String() string { return proto.CompactTextString(m) } +func (*Roster) ProtoMessage() {} +func (*Roster) Descriptor() ([]byte, []int) { + return fileDescriptor_786299925f8760d5, []int{3} +} +func (m *Roster) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Roster) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Roster.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalToSizedBuffer(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (m *Roster) XXX_Merge(src proto.Message) { + xxx_messageInfo_Roster.Merge(m, src) +} +func (m *Roster) XXX_Size() int { + return m.Size() +} +func (m *Roster) XXX_DiscardUnknown() { + xxx_messageInfo_Roster.DiscardUnknown(m) +} + +var xxx_messageInfo_Roster proto.InternalMessageInfo + +func (m *Roster) GetPublishers() [][]byte { + if m != nil { + return m.Publishers + } + return nil +} + // Every frame after the handshake. soc is the index-th update of the session feed: // a single-owner chunk whose id is derived from kind, challenge and index (see bps.ID). type Broadcast struct { @@ -288,7 +341,7 @@ func (m *Broadcast) Reset() { *m = Broadcast{} } func (m *Broadcast) String() string { return proto.CompactTextString(m) } func (*Broadcast) ProtoMessage() {} func (*Broadcast) Descriptor() ([]byte, []int) { - return fileDescriptor_786299925f8760d5, []int{3} + return fileDescriptor_786299925f8760d5, []int{4} } func (m *Broadcast) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -352,40 +405,43 @@ func init() { proto.RegisterType((*CohortSpec)(nil), "bps.CohortSpec") proto.RegisterType((*Join)(nil), "bps.Join") proto.RegisterType((*Ack)(nil), "bps.Ack") + proto.RegisterType((*Roster)(nil), "bps.Roster") proto.RegisterType((*Broadcast)(nil), "bps.Broadcast") } func init() { proto.RegisterFile("bps.proto", fileDescriptor_786299925f8760d5) } var fileDescriptor_786299925f8760d5 = []byte{ - // 424 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x5c, 0x92, 0xcd, 0x6e, 0xd3, 0x40, - 0x10, 0xc7, 0xe3, 0x0f, 0x4c, 0x32, 0x8d, 0xc2, 0x32, 0xaa, 0x50, 0x40, 0xad, 0x15, 0x85, 0x03, - 0x51, 0x90, 0x2a, 0x14, 0xce, 0x48, 0x38, 0xb1, 0x43, 0xdd, 0x54, 0x69, 0xb5, 0x76, 0x2e, 0x5c, - 0x22, 0x7f, 0xa9, 0x5d, 0x35, 0xec, 0xae, 0xe2, 0x45, 0x82, 0xb7, 0xe0, 0xb1, 0x38, 0xf6, 0xc8, - 0x11, 0x25, 0x2f, 0x82, 0xbc, 0x4e, 0x08, 0xe4, 0xb6, 0xf3, 0xff, 0xcf, 0xfc, 0xe6, 0x43, 0x0b, - 0xad, 0x54, 0x96, 0x17, 0x72, 0x2d, 0x94, 0x40, 0x2b, 0x95, 0x65, 0xff, 0x0b, 0xc0, 0x44, 0xdc, - 0x8b, 0xb5, 0x8a, 0x64, 0x91, 0xe1, 0x29, 0x3c, 0x51, 0x42, 0xb2, 0xac, 0x6b, 0xf4, 0x8c, 0x41, - 0x9b, 0xd6, 0x01, 0xbe, 0x85, 0xa7, 0x29, 0xe3, 0x39, 0xe3, 0x77, 0x5d, 0xb3, 0x67, 0x0c, 0x3a, - 0xa3, 0xe7, 0x17, 0x15, 0x25, 0xae, 0xcc, 0x71, 0x6d, 0xd0, 0x7d, 0x06, 0x9e, 0x41, 0x4b, 0xae, - 0x19, 0xcf, 0x98, 0x4c, 0x56, 0x5d, 0x4b, 0x63, 0x0e, 0x42, 0x7f, 0x06, 0xf6, 0x95, 0x60, 0x1c, - 0xdf, 0x80, 0x93, 0xe9, 0xb6, 0xba, 0xd3, 0xc9, 0xe8, 0x99, 0x26, 0x1e, 0x26, 0xa1, 0x3b, 0x1b, - 0x5f, 0x41, 0x93, 0xe5, 0x05, 0x57, 0x4c, 0x7d, 0xd7, 0xcd, 0xdb, 0xf4, 0x6f, 0xdc, 0xbf, 0x04, - 0xcb, 0xcb, 0x1e, 0xf0, 0x35, 0x38, 0xa5, 0x4a, 0xd4, 0xd7, 0x52, 0xb3, 0x3a, 0xa3, 0x13, 0xcd, - 0x8a, 0xb4, 0x44, 0x77, 0x56, 0x35, 0x56, 0x76, 0x9f, 0xac, 0x56, 0x05, 0xbf, 0x2b, 0x76, 0xa0, - 0x83, 0xd0, 0x97, 0xd0, 0x1a, 0xaf, 0x45, 0x92, 0x67, 0x49, 0xa9, 0x90, 0x80, 0x55, 0x8a, 0xfd, - 0x09, 0xaa, 0x27, 0x9e, 0x83, 0xfd, 0xc0, 0x78, 0xbe, 0xdb, 0xbe, 0xa5, 0xf9, 0x33, 0xc6, 0x73, - 0xaa, 0xe5, 0xff, 0xd9, 0xd6, 0x11, 0xbb, 0xba, 0x29, 0xe3, 0x79, 0xf1, 0xad, 0x6b, 0xf7, 0x8c, - 0x81, 0x4d, 0xeb, 0x60, 0xf8, 0x01, 0xda, 0xff, 0xde, 0x0f, 0xcf, 0xe1, 0x65, 0x7c, 0x73, 0x1b, - 0x4e, 0x96, 0xe3, 0x70, 0xee, 0x87, 0xf3, 0x4f, 0xcb, 0xc5, 0x3c, 0xba, 0x0d, 0x26, 0xe1, 0x34, - 0x0c, 0x7c, 0xd2, 0xc0, 0x0e, 0xc0, 0x34, 0x08, 0xfc, 0xa5, 0xce, 0x21, 0xf6, 0xf0, 0x23, 0x38, - 0xf5, 0x82, 0xf8, 0x02, 0x30, 0x8a, 0xbd, 0x78, 0x11, 0x1d, 0x55, 0x38, 0x60, 0xde, 0xcc, 0x88, - 0x81, 0x4d, 0xb0, 0xa7, 0x8b, 0xeb, 0x6b, 0x62, 0x62, 0x1b, 0x9a, 0x34, 0xb8, 0x0a, 0x26, 0x71, - 0xe0, 0x13, 0x6b, 0xf8, 0x0e, 0xec, 0x6a, 0x05, 0x3c, 0x05, 0x32, 0x0b, 0xe7, 0xfe, 0x51, 0x75, - 0x13, 0x6c, 0xdf, 0x8b, 0xbd, 0xba, 0xde, 0x5b, 0xc4, 0x97, 0xc4, 0x1c, 0x9f, 0xfd, 0xdc, 0xb8, - 0xc6, 0xe3, 0xc6, 0x35, 0x7e, 0x6f, 0x5c, 0xe3, 0xc7, 0xd6, 0x6d, 0x3c, 0x6e, 0xdd, 0xc6, 0xaf, - 0xad, 0xdb, 0xf8, 0x6c, 0xca, 0x34, 0x75, 0xf4, 0xa7, 0x7a, 0xff, 0x27, 0x00, 0x00, 0xff, 0xff, - 0x30, 0x35, 0xa6, 0xe5, 0x61, 0x02, 0x00, 0x00, + // 463 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x5c, 0x92, 0xcf, 0x6e, 0xd3, 0x40, + 0x10, 0xc6, 0xe3, 0xd8, 0xb8, 0xc9, 0xd4, 0x0a, 0xcb, 0xa8, 0x42, 0x01, 0xb5, 0x56, 0x64, 0x0e, + 0x58, 0x41, 0xea, 0x21, 0x9c, 0x91, 0x70, 0xe2, 0x0d, 0x75, 0x53, 0x25, 0xd5, 0xda, 0xb9, 0x70, + 0x89, 0xfc, 0x4f, 0xcd, 0xaa, 0xc1, 0x5e, 0xd9, 0xae, 0x04, 0x6f, 0xc1, 0x63, 0x71, 0xec, 0x91, + 0x23, 0x4a, 0x5e, 0x04, 0x79, 0x9d, 0x92, 0x92, 0x9b, 0xe7, 0x9b, 0x99, 0xdf, 0x7c, 0xfe, 0xb4, + 0xd0, 0x8d, 0x44, 0x79, 0x29, 0x8a, 0xbc, 0xca, 0x51, 0x8d, 0x44, 0x69, 0x7d, 0x03, 0x98, 0xe4, + 0xeb, 0xbc, 0xa8, 0x7c, 0x91, 0xc6, 0x78, 0x06, 0x2f, 0xaa, 0x5c, 0xf0, 0xb8, 0xaf, 0x0c, 0x14, + 0xdb, 0x60, 0x4d, 0x81, 0x1f, 0xe0, 0x24, 0xe2, 0x59, 0xc2, 0xb3, 0xbb, 0x7e, 0x7b, 0xa0, 0xd8, + 0xbd, 0xd1, 0xab, 0xcb, 0x9a, 0x12, 0xd4, 0xcd, 0x71, 0xd3, 0x60, 0x4f, 0x13, 0x78, 0x0e, 0x5d, + 0x51, 0xf0, 0x2c, 0xe6, 0x22, 0xdc, 0xf4, 0x55, 0x89, 0x39, 0x08, 0xd6, 0x0c, 0xb4, 0xeb, 0x9c, + 0x67, 0xf8, 0x1e, 0xf4, 0x58, 0x9e, 0x95, 0x97, 0x4e, 0x47, 0x2f, 0x25, 0xf1, 0xe0, 0x84, 0xed, + 0xdb, 0xf8, 0x16, 0x3a, 0x3c, 0x49, 0xb3, 0x8a, 0x57, 0x3f, 0xe4, 0x71, 0x83, 0xfd, 0xab, 0xad, + 0x2b, 0x50, 0x9d, 0xf8, 0x1e, 0xdf, 0x81, 0x5e, 0x56, 0x61, 0xf5, 0x50, 0x4a, 0x56, 0x6f, 0x74, + 0x2a, 0x59, 0xbe, 0x94, 0xd8, 0xbe, 0x55, 0xdb, 0x8a, 0xd7, 0xe1, 0x66, 0x93, 0x66, 0x77, 0xe9, + 0x1e, 0x74, 0x10, 0x2c, 0x1b, 0x74, 0x96, 0x97, 0x55, 0x5a, 0xa0, 0x09, 0x20, 0x1e, 0xa2, 0x0d, + 0x2f, 0xd7, 0x69, 0x51, 0x03, 0x55, 0xdb, 0x60, 0xcf, 0x14, 0x4b, 0x40, 0x77, 0x5c, 0xe4, 0x61, + 0x12, 0x87, 0x65, 0x85, 0x04, 0xd4, 0x32, 0x7f, 0x0a, 0xab, 0xfe, 0xc4, 0x0b, 0xd0, 0xee, 0x79, + 0x96, 0xec, 0x73, 0xea, 0x4a, 0x27, 0x33, 0x9e, 0x25, 0x4c, 0xca, 0xff, 0xbb, 0x50, 0x8f, 0x5c, + 0xd4, 0xe9, 0xf3, 0x2c, 0x49, 0xbf, 0xf7, 0xb5, 0x81, 0x62, 0x6b, 0xac, 0x29, 0x86, 0x9f, 0xc0, + 0x78, 0x9e, 0x34, 0x5e, 0xc0, 0x9b, 0x60, 0x71, 0xeb, 0x4d, 0x56, 0x63, 0x6f, 0xee, 0x7a, 0xf3, + 0x2f, 0xab, 0xe5, 0xdc, 0xbf, 0xa5, 0x13, 0x6f, 0xea, 0x51, 0x97, 0xb4, 0xb0, 0x07, 0x30, 0xa5, + 0xd4, 0x5d, 0xc9, 0x19, 0xa2, 0x0d, 0x3f, 0x83, 0xde, 0x44, 0x81, 0xaf, 0x01, 0xfd, 0xc0, 0x09, + 0x96, 0xfe, 0xd1, 0x86, 0x0e, 0xed, 0xc5, 0x8c, 0x28, 0xd8, 0x01, 0x6d, 0xba, 0xbc, 0xb9, 0x21, + 0x6d, 0x34, 0xa0, 0xc3, 0xe8, 0x35, 0x9d, 0x04, 0xd4, 0x25, 0xea, 0x90, 0x82, 0x56, 0xff, 0x02, + 0x9e, 0x01, 0x99, 0x79, 0x73, 0xf7, 0x68, 0xbb, 0x03, 0x9a, 0xeb, 0x04, 0x4e, 0xb3, 0xef, 0x2c, + 0x83, 0x2b, 0xd2, 0xc6, 0x13, 0x50, 0xe9, 0xc2, 0x27, 0x2a, 0x02, 0xe8, 0x6c, 0xe1, 0x07, 0x94, + 0x11, 0x6d, 0x7c, 0xfe, 0x6b, 0x6b, 0x2a, 0x8f, 0x5b, 0x53, 0xf9, 0xb3, 0x35, 0x95, 0x9f, 0x3b, + 0xb3, 0xf5, 0xb8, 0x33, 0x5b, 0xbf, 0x77, 0x66, 0xeb, 0x6b, 0x5b, 0x44, 0x91, 0x2e, 0xdf, 0xe4, + 0xc7, 0xbf, 0x01, 0x00, 0x00, 0xff, 0xff, 0x50, 0xef, 0xea, 0xc9, 0xa0, 0x02, 0x00, 0x00, } func (m *CohortSpec) Marshal() (dAtA []byte, err error) { @@ -507,6 +563,38 @@ func (m *Ack) MarshalToSizedBuffer(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *Roster) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBuffer(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *Roster) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Roster) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.Publishers) > 0 { + for iNdEx := len(m.Publishers) - 1; iNdEx >= 0; iNdEx-- { + i -= len(m.Publishers[iNdEx]) + copy(dAtA[i:], m.Publishers[iNdEx]) + i = encodeVarintBps(dAtA, i, uint64(len(m.Publishers[iNdEx]))) + i-- + dAtA[i] = 0xa + } + } + return len(dAtA) - i, nil +} + func (m *Broadcast) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -618,6 +706,21 @@ func (m *Ack) Size() (n int) { return n } +func (m *Roster) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if len(m.Publishers) > 0 { + for _, b := range m.Publishers { + l = len(b) + n += 1 + l + sovBps(uint64(l)) + } + } + return n +} + func (m *Broadcast) Size() (n int) { if m == nil { return 0 @@ -1016,6 +1119,91 @@ func (m *Ack) Unmarshal(dAtA []byte) error { } return nil } +func (m *Roster) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowBps + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: Roster: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Roster: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Publishers", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowBps + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthBps + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthBps + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Publishers = append(m.Publishers, make([]byte, postIndex-iNdEx)) + copy(m.Publishers[len(m.Publishers)-1], dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipBps(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthBps + } + if (iNdEx + skippy) < 0 { + return ErrInvalidLengthBps + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} func (m *Broadcast) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 diff --git a/pkg/bps/pb/bps.proto b/pkg/bps/pb/bps.proto index 286b58f6834..07281645da0 100644 --- a/pkg/bps/pb/bps.proto +++ b/pkg/bps/pb/bps.proto @@ -2,8 +2,10 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -// Wire format of SWIP-74 (BPS-lite). Field and enum numbers are shared with -// SWIP-60, which extends these messages by adding fields and values only. +// Wire format of SWIP-74 (BPS-lite) with the admin's service kinds of SWIP-87 +// (explicit multi-publisher): EOS, ROSTER and the Roster payload. Field and enum +// numbers are shared with SWIP-60, which extends these messages by adding fields +// and values only. syntax = "proto3"; package bps; @@ -20,7 +22,7 @@ enum TopicBinding { message CohortSpec { bytes topic = 1; // 32 bytes TopicBinding binding = 2; // FEED_TOPIC - bytes principal = 3; // 20 bytes, required: the publisher + bytes principal = 3; // 20 bytes, required: the admin } // Peer -> broker: the first frame on a fresh stream. @@ -47,7 +49,15 @@ enum Kind { KIND_UNSPECIFIED = 0; // invalid on the wire DATA = 1; // an update of the stream's session feed AUTH = 2; // the empty chunk at index 0 of the session's AUTH feed: authenticates - // a pending stream, a heartbeat on a publisher stream, never delivered + // a pending stream, or a spectator stream a delivered roster names; + // a heartbeat on a publisher stream; never delivered + EOS = 3; // the empty chunk at index 0 of the admin's EOS feed: the channel is ended + ROSTER = 4; // from the admin: the publisher set as of this index; payload = Roster +} + +// The payload of a ROSTER chunk. +message Roster { + repeated bytes publishers = 1; // 20-byte addresses: the complete set excluding the admin } // Every frame after the handshake. soc is the index-th update of the session feed: