feat(pubsub): stateless space-scoped pub/sub over any-sync

Makefile

@@ -53,6 +53,7 @@ proto:
 
 	$(call generate_drpc,,net/secureservice/handshake/handshakeproto/protos)
 	$(call generate_drpc,,net/rpc/limiter/limiterproto/protos)
+	$(call generate_drpc,,commonspace/pubsub/pubsubproto/protos)
 	$(call generate_drpc,$(PKGMAP),coordinator/coordinatorproto/protos)
 	$(call generate_drpc,,consensus/consensusproto/protos)
 	$(call generate_drpc,,identityrepo/identityrepoproto/protos)

commonspace/pubsub/dedup.go

@@ -0,0 +1,48 @@
+package pubsub
+
+import "sync"
+
+const msgIdLen = 16
+
+// msgIdDedup is a fixed-size duplicate-suppression cache keyed by message id.
+// It is a FIFO ring: adding a new id evicts the oldest one, keeping memory flat.
+type msgIdDedup struct {
+	mu   sync.Mutex
+	set  map[[msgIdLen]byte]struct{}
+	ring [][msgIdLen]byte
+	pos  int
+	full bool
+}
+
+func newMsgIdDedup(size int) *msgIdDedup {
+	return &msgIdDedup{
+		set:  make(map[[msgIdLen]byte]struct{}, size),
+		ring: make([][msgIdLen]byte, size),
+	}
+}
+
+// seen reports whether id was already recorded; if not, it records it,
+// evicting the oldest entry when the cache is full.
+func (d *msgIdDedup) seen(id []byte) bool {
+	if len(id) != msgIdLen {
+		return false
+	}
+	var key [msgIdLen]byte
+	copy(key[:], id)
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if _, ok := d.set[key]; ok {
+		return true
+	}
+	if d.full {
+		delete(d.set, d.ring[d.pos])
+	}
+	d.ring[d.pos] = key
+	d.set[key] = struct{}{}
+	d.pos++
+	if d.pos == len(d.ring) {
+		d.pos = 0
+		d.full = true
+	}
+	return false
+}

commonspace/pubsub/dedup_test.go

@@ -0,0 +1,41 @@
+package pubsub
+
+import (
+	"encoding/binary"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func testMsgId(n int) []byte {
+	id := make([]byte, msgIdLen)
+	binary.LittleEndian.PutUint64(id, uint64(n))
+	return id
+}
+
+func TestDedupSeen(t *testing.T) {
+	d := newMsgIdDedup(4)
+	require.False(t, d.seen(testMsgId(1)))
+	require.True(t, d.seen(testMsgId(1)))
+	require.False(t, d.seen(testMsgId(2)))
+	require.True(t, d.seen(testMsgId(2)))
+}
+
+func TestDedupEviction(t *testing.T) {
+	d := newMsgIdDedup(4)
+	for i := 1; i <= 4; i++ {
+		require.False(t, d.seen(testMsgId(i)))
+	}
+	// adding a fifth evicts the oldest (1)
+	require.False(t, d.seen(testMsgId(5)))
+	require.False(t, d.seen(testMsgId(1)), "oldest id should have been evicted")
+	// 1 was re-recorded above, evicting 2
+	require.True(t, d.seen(testMsgId(5)))
+	require.False(t, d.seen(testMsgId(2)))
+}
+
+func TestDedupInvalidLength(t *testing.T) {
+	d := newMsgIdDedup(4)
+	require.False(t, d.seen([]byte("short")))
+	require.False(t, d.seen([]byte("short")), "invalid ids are never recorded")
+}

commonspace/pubsub/deps.go

@@ -0,0 +1,141 @@
+package pubsub
+
+import (
+	"context"
+	"time"
+
+	"github.com/anyproto/any-sync/commonspace/pubsub/pubsubproto"
+	"github.com/anyproto/any-sync/metric"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/streampool"
+	"github.com/anyproto/any-sync/util/crypto"
+)
+
+// Handler receives a decrypted, signature-verified message on a subscribed topic.
+// Handlers run on a bounded dispatch queue and must not block.
+type Handler func(spaceId, topic string, identity crypto.PubKey, payload []byte)
+
+// MembershipChecker gates subscribe and publish on space membership.
+// Implementations resolve the space's ACL state; any non-nil error rejects.
+type MembershipChecker interface {
+	CheckMember(ctx context.Context, spaceId string, identity crypto.PubKey) error
+}
+
+// Crypto encrypts and decrypts payloads with the space read key.
+// A nil Crypto means plaintext payloads (keyless spaces).
+type Crypto interface {
+	// Encrypt returns the key id used and the ciphertext.
+	Encrypt(spaceId string, payload []byte) (keyId string, encrypted []byte, err error)
+	// Decrypt resolves keyId to a historical read key and decrypts.
+	Decrypt(spaceId, keyId string, encrypted []byte) ([]byte, error)
+}
+
+// PeerProvider resolves the peers a client sends publishes and interest to for a
+// space: the responsible sync node plus any directly connected LAN peers.
+type PeerProvider interface {
+	SpacePeers(ctx context.Context, spaceId string) ([]peer.Peer, error)
+}
+
+// Relay is implemented only on responsible nodes; nil on clients.
+type Relay interface {
+	// IsResponsible reports whether this node is responsible for the space.
+	IsResponsible(spaceId string) bool
+	// IsResponsibleNode reports whether peerId is a responsible node for the space
+	// (used to authorize inbound relayed publishes).
+	IsResponsibleNode(spaceId, peerId string) bool
+	// OtherResponsiblePeers returns the other responsible nodes, excluding self.
+	OtherResponsiblePeers(ctx context.Context, spaceId string) ([]peer.Peer, error)
+}
+
+// StatusHandler observes Status frames received from serving peers (rejections).
+type StatusHandler func(peerId string, status *pubsubproto.Status)
+
+// Deps carries the pluggable pieces wired by the node or client host.
+type Deps struct {
+	Membership MembershipChecker
+	Crypto     Crypto
+	Peers      PeerProvider // client side; may be nil on nodes
+	Relay      Relay        // node side; nil on clients
+	OnStatus   StatusHandler
+	// Metric, if set, registers the private pool's prometheus metrics so a node
+	// relaying at scale is observable. Optional.
+	Metric metric.Metric
+	Config Config
+}
+
+// Config bounds the engine per DESIGN.md §9; zero values take defaults.
+type Config struct {
+	MaxPayloadSize       int
+	MaxPatternsPerStream int
+	MaxPatternsPerSpace  int
+	PublishRps           float64
+	PublishBurst         int
+	WriteQueueSize       int
+	DispatchQueueSize    int
+	DedupSize            int
+	// MaxTimestampSkew bounds how stale (or future) a received message's timestamp
+	// may be before it is dropped, raising the replay bar even after dedup eviction.
+	MaxTimestampSkew time.Duration
+	// ResyncInterval is how often a client re-pushes its interest to space peers,
+	// keeping server-side interest alive across reconnects.
+	ResyncInterval time.Duration
+	// PeerTTL keeps a pubsub stream's peer from being reaped by idle pool GC.
+	PeerTTL time.Duration
+	// DialQueueWorkers/DialQueueSize size the pool's outbound dial pool.
+	DialQueueWorkers int
+	DialQueueSize    int
+}
+
+func (c Config) withDefaults() Config {
+	if c.MaxPayloadSize <= 0 {
+		c.MaxPayloadSize = 64 * 1024
+	}
+	if c.MaxPatternsPerStream <= 0 {
+		c.MaxPatternsPerStream = 1000
+	}
+	if c.MaxPatternsPerSpace <= 0 {
+		c.MaxPatternsPerSpace = 100
+	}
+	if c.PublishRps <= 0 {
+		c.PublishRps = 30
+	}
+	if c.PublishBurst <= 0 {
+		c.PublishBurst = 60
+	}
+	if c.WriteQueueSize <= 0 {
+		c.WriteQueueSize = 100
+	}
+	if c.DispatchQueueSize <= 0 {
+		c.DispatchQueueSize = 100
+	}
+	if c.DedupSize <= 0 {
+		c.DedupSize = 4096
+	}
+	if c.MaxTimestampSkew <= 0 {
+		c.MaxTimestampSkew = 5 * time.Minute
+	}
+	if c.ResyncInterval <= 0 {
+		c.ResyncInterval = 20 * time.Second
+	}
+	if c.PeerTTL <= 0 {
+		c.PeerTTL = time.Hour
+	}
+	if c.DialQueueWorkers <= 0 {
+		c.DialQueueWorkers = 4
+	}
+	if c.DialQueueSize <= 0 {
+		c.DialQueueSize = 100
+	}
+	return c
+}
+
+// streamPoolConfig maps the pubsub config onto the streampool's own config. Only
+// the dial knobs are consumed by the pool; per-stream queue size is passed
+// explicitly to AddStream/ReadStream via WriteQueueSize.
+func (c Config) streamPoolConfig() streampool.StreamConfig {
+	return streampool.StreamConfig{
+		SendQueueSize:    c.WriteQueueSize,
+		DialQueueWorkers: c.DialQueueWorkers,
+		DialQueueSize:    c.DialQueueSize,
+	}
+}

commonspace/pubsub/pubsubproto/errors.go

@@ -0,0 +1,20 @@
+package pubsubproto
+
+import (
+	"errors"
+
+	"github.com/anyproto/any-sync/net/rpc/rpcerr"
+)
+
+var (
+	errGroup = rpcerr.ErrGroup(ErrCodes_ErrorOffset)
+
+	ErrUnexpected     = errGroup.Register(errors.New("unexpected error"), uint64(ErrCodes_Unexpected))
+	ErrNotAMember     = errGroup.Register(errors.New("identity is not a space member"), uint64(ErrCodes_NotAMember))
+	ErrNotResponsible = errGroup.Register(errors.New("peer is not responsible for space"), uint64(ErrCodes_NotResponsible))
+	ErrRateLimited    = errGroup.Register(errors.New("publish rate limit exceeded"), uint64(ErrCodes_RateLimited))
+	ErrTooManyTopics  = errGroup.Register(errors.New("too many topic patterns"), uint64(ErrCodes_TooManyTopics))
+	ErrInvalidMessage = errGroup.Register(errors.New("invalid message"), uint64(ErrCodes_InvalidMessage))
+	ErrTopicNotOwned  = errGroup.Register(errors.New("topic is owned by another account"), uint64(ErrCodes_TopicNotOwned))
+	ErrInvalidTopic   = errGroup.Register(errors.New("invalid topic or pattern"), uint64(ErrCodes_InvalidTopic))
+)

commonspace/pubsub/pubsubproto/protos/pubsub.proto

@@ -0,0 +1,86 @@
+syntax = "proto3";
+package pubsub;
+
+option go_package = "commonspace/pubsub/pubsubproto";
+
+service PubSub {
+    // PubSubStream is a long-lived bidirectional stream multiplexing all spaces and topics between two peers
+    rpc PubSubStream(stream PubSubMessage) returns (stream PubSubMessage);
+}
+
+enum ErrCodes {
+    Unexpected = 0;
+    // NotAMember - identity has no permissions in the space's ACL
+    NotAMember = 1;
+    // NotResponsible - the serving node is not responsible for the space
+    NotResponsible = 2;
+    // RateLimited - the per-peer publish rate limit was exceeded
+    RateLimited = 3;
+    // TooManyTopics - the per-stream or per-space pattern cap was exceeded
+    TooManyTopics = 4;
+    // InvalidMessage - malformed frame, oversized payload, identity mismatch or bad signature
+    InvalidMessage = 5;
+    // TopicNotOwned - publish into the acc/ namespace by an identity other than the topic owner
+    TopicNotOwned = 6;
+    // InvalidTopic - malformed topic or pattern: bad wildcard placement, reserved chars, non-canonical form
+    InvalidTopic = 7;
+    // ErrorOffset - error-code band for this proto package; see docs/rpc-error-offsets.md
+    // (100..900 are the any-sync core bands; pubsub uses 1100)
+    ErrorOffset = 1100;
+}
+
+// PubSubMessage is the single frame type carried by PubSubStream
+message PubSubMessage {
+    oneof content {
+        Subscribe subscribe = 1;
+        Unsubscribe unsubscribe = 2;
+        Publish publish = 3;
+        Status status = 4;
+    }
+}
+
+// Subscribe adds topic patterns to the stream's interest set for the space.
+// Patterns may contain wildcards: '*' matches exactly one segment, '>' matches one-or-more trailing segments (tail-only)
+message Subscribe {
+    string spaceId = 1;
+    repeated string topics = 2;
+}
+
+// Unsubscribe removes patterns from the stream's interest set, matched verbatim (not expanded);
+// empty topics means remove all patterns of the space
+message Unsubscribe {
+    string spaceId = 1;
+    repeated string topics = 2;
+}
+
+// Publish carries one ephemeral fire-and-forget message to a fully-qualified topic
+message Publish {
+    string spaceId = 1;
+    // topic is a fully-qualified '/'-separated topic; wildcards are not allowed
+    string topic = 2;
+    // msgId is 16 random bytes generated by the publisher, used for duplicate suppression
+    bytes msgId = 3;
+    // keyId is the space ReadKey id used to encrypt the payload; empty means plaintext (keyless spaces)
+    string keyId = 4;
+    // payload is ciphertext, or plaintext iff keyId is empty
+    bytes payload = 5;
+    // identity is the sender's marshalled account public key
+    bytes identity = 6;
+    // signature is the account-key signature over "anysync:pubsub:v1" | spaceId | topic | msgId | keyId | le64(timestampMilli) | payload
+    bytes signature = 7;
+    // timestampMilli is the sender's wall clock, informational
+    int64 timestampMilli = 8;
+    // relayed is set by a node when forwarding node-to-node; a relayed message is never forwarded again; excluded from the signature
+    bool relayed = 9;
+}
+
+// Status is sent by the serving peer on a rejected subscribe or publish; success is silent
+message Status {
+    string spaceId = 1;
+    // topics echoes the offending patterns (or the topic of a rejected publish)
+    repeated string topics = 2;
+    ErrCodes code = 3;
+    // msgId echoes a rejected publish's id so the caller can correlate the
+    // rejection to a specific Publish; empty for subscribe rejections
+    bytes msgId = 4;
+}