diff --git a/cmd/any-sync-node.go b/cmd/any-sync-node.go index f6875192..65d22f9f 100644 --- a/cmd/any-sync-node.go +++ b/cmd/any-sync-node.go @@ -54,6 +54,7 @@ import ( "github.com/anyproto/any-sync-node/debug/spacechecker" "github.com/anyproto/any-sync-node/nodespace" "github.com/anyproto/any-sync-node/nodespace/nodecache" + "github.com/anyproto/any-sync-node/nodespace/pubsubrelay" "github.com/anyproto/any-sync-node/nodestorage" ) @@ -150,6 +151,7 @@ func Bootstrap(a *app.App) { Register(secureservice.New()). Register(commonspace.New()). Register(nodespace.New()). + Register(pubsubrelay.New()). Register(spacedeleter.New()). Register(peermanager.New()). Register(debugserver.New()). diff --git a/go.mod b/go.mod index 678954a6..2f99f1cd 100644 --- a/go.mod +++ b/go.mod @@ -1,12 +1,12 @@ module github.com/anyproto/any-sync-node -go 1.25.0 +go 1.25.7 require ( github.com/ahmetb/govvv v0.3.0 github.com/akrylysov/pogreb v0.10.3-0.20240803013244-523613e335e9 - github.com/anyproto/any-store v0.4.6 - github.com/anyproto/any-sync v0.11.20 + github.com/anyproto/any-store v0.4.7 + github.com/anyproto/any-sync v0.12.13-0.20260707065752-f05d5319cff2 github.com/anyproto/go-chash v0.1.0 github.com/aws/aws-sdk-go v1.55.8 github.com/cheggaaa/mb/v3 v3.0.2 @@ -16,12 +16,12 @@ require ( go.uber.org/atomic v1.11.0 go.uber.org/mock v0.6.0 go.uber.org/multierr v1.11.0 - go.uber.org/zap v1.27.1 - golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a - golang.org/x/net v0.52.0 + go.uber.org/zap v1.28.0 + golang.org/x/exp v0.0.0-20260603202125-055de637280b + golang.org/x/net v0.56.0 google.golang.org/protobuf v1.36.11 gopkg.in/yaml.v3 v3.0.1 - storj.io/drpc v0.0.34 + storj.io/drpc v1.0.0 ) require ( @@ -39,8 +39,9 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect - github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect + github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 // indirect github.com/disintegration/imaging v1.6.2 // indirect + github.com/dunglas/httpsfv v1.1.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/flopp/go-findfont v0.1.0 // indirect github.com/fogleman/gg v1.3.0 // indirect @@ -51,21 +52,21 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/hashicorp/yamux v0.1.2 // indirect github.com/huandu/skiplist v1.2.1 // indirect - github.com/ipfs/boxo v0.37.0 // indirect + github.com/ipfs/boxo v0.41.0 // indirect github.com/ipfs/go-block-format v0.2.3 // indirect - github.com/ipfs/go-cid v0.6.0 // indirect + github.com/ipfs/go-cid v0.6.1 // indirect github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect github.com/jmespath/go-jmespath v0.4.0 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/libp2p/go-buffer-pool v0.1.0 // indirect - github.com/libp2p/go-libp2p v0.47.0 // indirect - github.com/mattn/go-isatty v0.0.20 // indirect + github.com/libp2p/go-libp2p v0.48.0 // indirect + github.com/mattn/go-isatty v0.0.22 // indirect github.com/minio/sha256-simd v1.0.1 // indirect - github.com/mr-tron/base58 v1.2.0 // indirect + github.com/mr-tron/base58 v1.3.0 // indirect github.com/multiformats/go-base32 v0.1.0 // indirect github.com/multiformats/go-base36 v0.2.0 // indirect github.com/multiformats/go-multiaddr v0.16.1 // indirect - github.com/multiformats/go-multibase v0.2.0 // indirect + github.com/multiformats/go-multibase v0.3.0 // indirect github.com/multiformats/go-multicodec v0.10.0 // indirect github.com/multiformats/go-multihash v0.2.3 // indirect github.com/multiformats/go-multistream v0.6.1 // indirect @@ -74,22 +75,24 @@ require ( github.com/ncruces/go-strftime v0.1.9 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_model v0.6.2 // indirect - github.com/prometheus/common v0.66.1 // indirect - github.com/prometheus/procfs v0.17.0 // indirect - github.com/quic-go/quic-go v0.59.0 // indirect + github.com/prometheus/common v0.67.5 // indirect + github.com/prometheus/procfs v0.20.1 // indirect + github.com/quic-go/qpack v0.6.0 // indirect + github.com/quic-go/quic-go v0.60.0 // indirect + github.com/quic-go/webtransport-go v0.11.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/tetratelabs/wazero v1.10.1 // indirect - github.com/valyala/fastjson v1.6.7 // indirect + github.com/valyala/fastjson v1.6.10 // indirect github.com/zeebo/blake3 v0.2.4 // indirect github.com/zeebo/errs v1.3.0 // indirect - go.yaml.in/yaml/v2 v2.4.3 // indirect - golang.org/x/crypto v0.49.0 // indirect + go.yaml.in/yaml/v2 v2.4.4 // indirect + golang.org/x/crypto v0.53.0 // indirect golang.org/x/image v0.21.0 // indirect - golang.org/x/sys v0.42.0 // indirect - golang.org/x/text v0.35.0 // indirect + golang.org/x/sys v0.46.0 // indirect + golang.org/x/text v0.38.0 // indirect golang.org/x/time v0.15.0 // indirect - golang.org/x/tools v0.43.0 // indirect + golang.org/x/tools v0.46.0 // indirect lukechampine.com/blake3 v1.4.1 // indirect modernc.org/libc v1.66.8 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/go.sum b/go.sum index b1c42b52..942916cf 100644 --- a/go.sum +++ b/go.sum @@ -7,10 +7,10 @@ github.com/ahmetb/govvv v0.3.0 h1:YGLGwEyiUwHFy5eh/RUhdupbuaCGBYn5T5GWXp+WJB0= github.com/ahmetb/govvv v0.3.0/go.mod h1:4WRFpdWtc/YtKgPFwa1dr5+9hiRY5uKAL08bOlxOR6s= github.com/akrylysov/pogreb v0.10.3-0.20240803013244-523613e335e9 h1:GnBlbnor8geJB4j7GGXqVHiSG0cHNEkIYNmpYDAPa+Y= github.com/akrylysov/pogreb v0.10.3-0.20240803013244-523613e335e9/go.mod h1:fPb3n+7H42SxX84B4/os7POrR+UGRKSRr3kNS5+xq/c= -github.com/anyproto/any-store v0.4.6 h1:opnTUGfuHa/VA9Incu6iLEy3WsQ4mb2o1oAXGDysH0s= -github.com/anyproto/any-store v0.4.6/go.mod h1:Npi35qMUVZ8ouiV4o9AqpZDs6LbDOF+5ZLlVijXofFM= -github.com/anyproto/any-sync v0.11.20 h1:/yv38s5KJh1A9SmYj1zD5k8Fqp1v0UZae3n8KNBd5Z0= -github.com/anyproto/any-sync v0.11.20/go.mod h1:qR5leKtd4j6cnmFiaI4ULe+Az3qjhKpdFtScV7aFCHY= +github.com/anyproto/any-store v0.4.7 h1:329NWY/xUzGdwKSqFgjAFaPAVkTJbR+WdJirPrvTm/w= +github.com/anyproto/any-store v0.4.7/go.mod h1:8cqb52gjZSaYnlybugqpSqSG1RQygY96D2vWMbSJsLo= +github.com/anyproto/any-sync v0.12.13-0.20260707065752-f05d5319cff2 h1:jbzhhga5tjmrYzbBYO8aZWXUPJ5Utqh9WeHg0GKWdyo= +github.com/anyproto/any-sync v0.12.13-0.20260707065752-f05d5319cff2/go.mod h1:3tGmfzxWIP476UwIFG2XnqwEBSk0Uml8OmN/f/m64Oo= github.com/anyproto/go-bip39 v1.0.0 h1:T6/7WowKYDeyuX/QyXtt98ZX0XXaoOh17M/LFF2M5yk= github.com/anyproto/go-bip39 v1.0.0/go.mod h1:l0rcxmXRyiWAYzE1noMAc4qbeNrbhUwxM3rqSO9ILwo= github.com/anyproto/go-chash v0.1.0 h1:I9meTPjXFRfXZHRJzjOHC/XF7Q5vzysKkiT/grsogXY= @@ -57,10 +57,12 @@ github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c h1:pFUpOrbxDR github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c/go.mod h1:6UhI8N9EjYm1c2odKpFpAYeR8dsBeM7PtzQhRgxRr9U= github.com/decred/dcrd/crypto/blake256 v1.1.0 h1:zPMNGQCm0g4QTY27fOCorQW7EryeQ/U0x++OzVrdms8= github.com/decred/dcrd/crypto/blake256 v1.1.0/go.mod h1:2OfgNZ5wDpcsFmHmCK5gZTPcCXqlm2ArzUIkw9czNJo= -github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 h1:NMZiJj8QnKe1LgsbDayM4UoHwbvwDRwnI3hwNaAHRnc= -github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40= +github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 h1:5RVFMOWjMyRy8cARdy79nAmgYw3hK/4HUq48LQ6Wwqo= +github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40= github.com/disintegration/imaging v1.6.2 h1:w1LecBlG2Lnp8B3jk5zSuNqd7b4DXhcjwek1ei82L+c= github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4= +github.com/dunglas/httpsfv v1.1.0 h1:Jw76nAyKWKZKFrpMMcL76y35tOpYHqQPzHQiwDvpe54= +github.com/dunglas/httpsfv v1.1.0/go.mod h1:zID2mqw9mFsnt7YC3vYQ9/cjq30q41W+1AnDwH8TiMg= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/flopp/go-findfont v0.1.0 h1:lPn0BymDUtJo+ZkV01VS3661HL6F4qFlkhcJN55u6mU= @@ -88,12 +90,12 @@ github.com/huandu/go-assert v1.1.5 h1:fjemmA7sSfYHJD7CUqs9qTwwfdNAx7/j2/ZlHXzNB3 github.com/huandu/go-assert v1.1.5/go.mod h1:yOLvuqZwmcHIC5rIzrBhT7D3Q9c3GFnd0JrPVhn/06U= github.com/huandu/skiplist v1.2.1 h1:dTi93MgjwErA/8idWTzIw4Y1kZsMWx35fmI2c8Rij7w= github.com/huandu/skiplist v1.2.1/go.mod h1:7v3iFjLcSAzO4fN5B8dvebvo/qsfumiLiDXMrPiHF9w= -github.com/ipfs/boxo v0.37.0 h1:2E3mZvydMI2t5IkAgtkmZ3sGsld0oS7o3I+xyzDk6uI= -github.com/ipfs/boxo v0.37.0/go.mod h1:8yyiRn54F2CsW13n0zwXEPrVsZix/gFj9SYIRYMZ6KE= +github.com/ipfs/boxo v0.41.0 h1:diKlFosOG2e1mgSO1CXqcMSnHvtn6ubUvaCf9iF8AIY= +github.com/ipfs/boxo v0.41.0/go.mod h1:1Fo36UVVvq3XAZwMDD82Cm4JTUi5x1k3AsJlg9DttOY= github.com/ipfs/go-block-format v0.2.3 h1:mpCuDaNXJ4wrBJLrtEaGFGXkferrw5eqVvzaHhtFKQk= github.com/ipfs/go-block-format v0.2.3/go.mod h1:WJaQmPAKhD3LspLixqlqNFxiZ3BZ3xgqxxoSR/76pnA= -github.com/ipfs/go-cid v0.6.0 h1:DlOReBV1xhHBhhfy/gBNNTSyfOM6rLiIx9J7A4DGf30= -github.com/ipfs/go-cid v0.6.0/go.mod h1:NC4kS1LZjzfhK40UGmpXv5/qD2kcMzACYJNntCUiDhQ= +github.com/ipfs/go-cid v0.6.1 h1:T5TnNb08+ueovG76Z5gx1L4Y7QOaGTXHg1F6raWFxIc= +github.com/ipfs/go-cid v0.6.1/go.mod h1:zrY0SwOhjrrIdfPQ/kf+k1sXyJ0QE7cMxfCployLBs0= github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABoLk/+KKHggpk= github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk= github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI= @@ -115,26 +117,26 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8= github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg= -github.com/libp2p/go-libp2p v0.47.0 h1:qQpBjSCWNQFF0hjBbKirMXE9RHLtSuzTDkTfr1rw0yc= -github.com/libp2p/go-libp2p v0.47.0/go.mod h1:s8HPh7mMV933OtXzONaGFseCg/BE//m1V34p3x4EUOY= +github.com/libp2p/go-libp2p v0.48.0 h1:h2BrLAgrj7X8bEN05K7qmrjpNHYA+6tnsGRdprjTnvo= +github.com/libp2p/go-libp2p v0.48.0/go.mod h1:Q1fBZNdmC2Hf82husCTfkKJVfHm2we5zk+NWmOGEmWk= github.com/libp2p/go-msgio v0.3.0 h1:mf3Z8B1xcFN314sWX+2vOTShIE0Mmn2TXn3YCUQGNj0= github.com/libp2p/go-msgio v0.3.0/go.mod h1:nyRM819GmVaF9LX3l03RMh10QdOroF++NBbxAb0mmDM= github.com/libp2p/go-yamux/v5 v5.0.1 h1:f0WoX/bEF2E8SbE4c/k1Mo+/9z0O4oC/hWEA+nfYRSg= github.com/libp2p/go-yamux/v5 v5.0.1/go.mod h1:en+3cdX51U0ZslwRdRLrvQsdayFt3TSUKvBGErzpWbU= -github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= -github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4= +github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM= github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8= -github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o= -github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc= +github.com/mr-tron/base58 v1.3.0 h1:K6Y13R2h+dku0wOqKtecgRnBUBPrZzLZy5aIj8lCcJI= +github.com/mr-tron/base58 v1.3.0/go.mod h1:2BuubE67DCSWwVfx37JWNG8emOC0sHEU4/HpcYgCLX8= github.com/multiformats/go-base32 v0.1.0 h1:pVx9xoSPqEIQG8o+UbAe7DNi51oej1NtK+aGkbLYxPE= github.com/multiformats/go-base32 v0.1.0/go.mod h1:Kj3tFY6zNr+ABYMqeUNeGvkIC/UYgtWibDcT0rExnbI= github.com/multiformats/go-base36 v0.2.0 h1:lFsAbNOGeKtuKozrtBsAkSVhv1p9D0/qedU9rQyccr0= github.com/multiformats/go-base36 v0.2.0/go.mod h1:qvnKE++v+2MWCfePClUEjE78Z7P2a1UV0xHgWc0hkp4= github.com/multiformats/go-multiaddr v0.16.1 h1:fgJ0Pitow+wWXzN9do+1b8Pyjmo8m5WhGfzpL82MpCw= github.com/multiformats/go-multiaddr v0.16.1/go.mod h1:JSVUmXDjsVFiW7RjIFMP7+Ev+h1DTbiJgVeTV/tcmP0= -github.com/multiformats/go-multibase v0.2.0 h1:isdYCVLvksgWlMW9OZRYJEa9pZETFivncJHmHnnd87g= -github.com/multiformats/go-multibase v0.2.0/go.mod h1:bFBZX4lKCA/2lyOFSAoKH5SS6oPyjtnzK/XTFDPkNuk= +github.com/multiformats/go-multibase v0.3.0 h1:8helZD2+4Db7NNWFiktk2NePbF0boolBe6bDQvM4r68= +github.com/multiformats/go-multibase v0.3.0/go.mod h1:MoBLQPCkRTOL3eveIPO81860j2AQY8JwcnNlRkGRUfI= github.com/multiformats/go-multicodec v0.10.0 h1:UpP223cig/Cx8J76jWt91njpK3GTAO1w02sdcjZDSuc= github.com/multiformats/go-multicodec v0.10.0/go.mod h1:wg88pM+s2kZJEQfRCKBNU+g32F5aWBEjyFHXvZLTcLI= github.com/multiformats/go-multihash v0.2.3 h1:7Lyc8XfX/IY2jWb/gI7JP+o7JEq9hOa7BFvVU9RSh+U= @@ -160,12 +162,18 @@ github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= -github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= -github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= -github.com/prometheus/procfs v0.17.0 h1:FuLQ+05u4ZI+SS/w9+BWEM2TXiHKsUQ9TADiRH7DuK0= -github.com/prometheus/procfs v0.17.0/go.mod h1:oPQLaDAMRbA+u8H5Pbfq+dl3VDAvHxMUOVhe0wYB2zw= -github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= -github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTUGI4= +github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw= +github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc= +github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo= +github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0= +github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.60.0 h1:xcQioE8OM66UQLeUMHltK1CCcOu3JbVB4JAQdDQSB+0= +github.com/quic-go/quic-go v0.60.0/go.mod h1:wpKpjmPpftl30sL6pFh7REVpjbcCVy4zt2vDyK1TuJk= +github.com/quic-go/webtransport-go v0.11.0 h1:3afiZq7MHv3gmKCbMwZ8D5M1u0y/1RdONN9KlWp32J0= +github.com/quic-go/webtransport-go v0.11.0/go.mod h1:SHgEzUFVyj+9WUSuGB1P6Zd351Pww2leWV3SwlTovkA= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= @@ -179,8 +187,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/tetratelabs/wazero v1.10.1 h1:2DugeJf6VVk58KTPszlNfeeN8AhhpwcZqkJj2wwFuH8= github.com/tetratelabs/wazero v1.10.1/go.mod h1:DRm5twOQ5Gr1AoEdSi0CLjDQF1J9ZAuyqFIjl1KKfQU= -github.com/valyala/fastjson v1.6.7 h1:ZE4tRy0CIkh+qDc5McjatheGX2czdn8slQjomexVpBM= -github.com/valyala/fastjson v1.6.7/go.mod h1:CLCAqky6SMuOcxStkYQvblddUtoRxhYMGLrsQns1aXY= +github.com/valyala/fastjson v1.6.10 h1:/yjJg8jaVQdYR3arGxPE2X5z89xrlhS0eGXdv+ADTh4= +github.com/valyala/fastjson v1.6.10/go.mod h1:e6FubmQouUNP73jtMLmcbxS6ydWIpOfhz34TSfO3JaE= github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ= github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= github.com/zeebo/blake3 v0.2.4 h1:KYQPkhpRtcqh0ssGYcKLG1JYvddkEA8QwCM/yBqhaZI= @@ -197,44 +205,45 @@ go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= -go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc= -go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= -go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0= -go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8= +go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= +go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20170930174604-9419663f5a44/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20200115085410-6d4e4cb37c7d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= -golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= -golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a h1:ovFr6Z0MNmU7nH8VaX5xqw+05ST2uO1exVfZPVqRC5o= -golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a/go.mod h1:K79w1Vqn7PoiZn+TkNpx3BUWUQksGO3JcVX6qIjytmA= +golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= +golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= +golang.org/x/exp v0.0.0-20260603202125-055de637280b h1:v1uXiEBHo8QA0LiGCo7UgHMzHT4Kdfpl2zmtH5vaP1Q= +golang.org/x/exp v0.0.0-20260603202125-055de637280b/go.mod h1:d2fgXJLVs4dYDHUk5lwMIfzRzSrWCfGZb0ZqeLa/Vcw= golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= golang.org/x/image v0.21.0 h1:c5qV36ajHpdj4Qi0GnE0jUc/yuo33OLFaa0d+crTD5s= golang.org/x/image v0.21.0/go.mod h1:vUbsLavqK/W303ZroQQVKQ+Af3Yl6Uz1Ppu5J/cLz78= -golang.org/x/mod v0.34.0 h1:xIHgNUUnW6sYkcM5Jleh05DvLOtwc6RitGHbDk4akRI= -golang.org/x/mod v0.34.0/go.mod h1:ykgH52iCZe79kzLLMhyCUzhMci+nQj+0XkbXpNYtVjY= +golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= +golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= -golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= +golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= +golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= +golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200602225109-6fdc65e7d980/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= +golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= -golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= +golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= +golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= -golang.org/x/tools v0.43.0 h1:12BdW9CeB3Z+J/I/wj34VMl8X+fEXBxVR90JeMX5E7s= -golang.org/x/tools v0.43.0/go.mod h1:uHkMso649BX2cZK6+RpuIPXS3ho2hZo4FVwfoy1vIk0= +golang.org/x/tools v0.46.0 h1:7jTurBkPZu4moS/Uy4OQT1M+QBlsj3wejyZwsT8Z7rk= +golang.org/x/tools v0.46.0/go.mod h1:FrD85F8l+NWL+9XWBSyVSHO6Ne4jutsfIFba7AWQ5Ys= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -276,5 +285,5 @@ modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= -storj.io/drpc v0.0.34 h1:q9zlQKfJ5A7x8NQNFk8x7eKUF78FMhmAbZLnFK+og7I= -storj.io/drpc v0.0.34/go.mod h1:Y9LZaa8esL1PW2IDMqJE7CFSNq7d5bQ3RI7mGPtmKMg= +storj.io/drpc v1.0.0 h1:1Xf1KCXXbV1viIfN56eqdJ3cNwpAL7OKwQkpS6ksing= +storj.io/drpc v1.0.0/go.mod h1:Y9LZaa8esL1PW2IDMqJE7CFSNq7d5bQ3RI7mGPtmKMg= diff --git a/nodespace/mock_nodespace/mock_nodespace.go b/nodespace/mock_nodespace/mock_nodespace.go index 2368ae0a..845f00dc 100644 --- a/nodespace/mock_nodespace/mock_nodespace.go +++ b/nodespace/mock_nodespace/mock_nodespace.go @@ -187,6 +187,18 @@ func (mr *MockServiceMockRecorder) Run(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockService)(nil).Run), ctx) } +// SetAclObserver mocks base method. +func (m *MockService) SetAclObserver(observer func(string)) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "SetAclObserver", observer) +} + +// SetAclObserver indicates an expected call of SetAclObserver. +func (mr *MockServiceMockRecorder) SetAclObserver(observer any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetAclObserver", reflect.TypeOf((*MockService)(nil).SetAclObserver), observer) +} + // MockNodeSpace is a mock of NodeSpace interface. type MockNodeSpace struct { ctrl *gomock.Controller @@ -424,6 +436,20 @@ func (mr *MockNodeSpaceMockRecorder) StoredIds() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StoredIds", reflect.TypeOf((*MockNodeSpace)(nil).StoredIds)) } +// SyncHeads mocks base method. +func (m *MockNodeSpace) SyncHeads(ctx context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SyncHeads", ctx) + ret0, _ := ret[0].(error) + return ret0 +} + +// SyncHeads indicates an expected call of SyncHeads. +func (mr *MockNodeSpaceMockRecorder) SyncHeads(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncHeads", reflect.TypeOf((*MockNodeSpace)(nil).SyncHeads), ctx) +} + // SyncStatus mocks base method. func (m *MockNodeSpace) SyncStatus() syncstatus.StatusUpdater { m.ctrl.T.Helper() diff --git a/nodespace/pubsubrelay/pubsubrelay.go b/nodespace/pubsubrelay/pubsubrelay.go new file mode 100644 index 00000000..7425e177 --- /dev/null +++ b/nodespace/pubsubrelay/pubsubrelay.go @@ -0,0 +1,223 @@ +// Package pubsubrelay wires the any-sync commonspace/pubsub engine into the node +// as a relay: it authorizes subscribe/publish against the hosted space ACL, +// forwards client-originated messages to the other responsible nodes, and evicts +// removed members when a space's ACL changes. +package pubsubrelay + +import ( + "context" + "errors" + "slices" + "sync" + "time" + + "github.com/anyproto/any-sync/accountservice" + "github.com/anyproto/any-sync/app" + "github.com/anyproto/any-sync/app/logger" + "github.com/anyproto/any-sync/commonspace/object/accountdata" + "github.com/anyproto/any-sync/commonspace/object/acl/list" + "github.com/anyproto/any-sync/commonspace/pubsub" + "github.com/anyproto/any-sync/metric" + "github.com/anyproto/any-sync/net/peer" + "github.com/anyproto/any-sync/net/pool" + "github.com/anyproto/any-sync/net/rpc/server" + "github.com/anyproto/any-sync/nodeconf" + "github.com/anyproto/any-sync/util/crypto" + "go.uber.org/zap" + + "github.com/anyproto/any-sync-node/nodespace" +) + +const CName = "node.pubsubrelay" + +// revalidateTimeout bounds a single space's member re-check so the worker can't +// wedge on a slow PickSpace/ACL read. +const revalidateTimeout = 30 * time.Second + +var log = logger.NewNamed(CName) + +var errNotMember = errors.New("account is not a space member") + +func New() app.ComponentRunnable { + return &relay{} +} + +// relay implements the pubsub engine's Relay and MembershipChecker deps and owns +// the engine's lifecycle. The engine itself is not registered in the app graph; +// this component drives its Init/Run/Close. +type relay struct { + nodeconf nodeconf.Service + pool pool.Pool + nodeSpace nodespace.Service + account *accountdata.AccountKeys + engine pubsub.Service + selfPeerId string + ctx context.Context + ctxCancel context.CancelFunc + + // ACL-change revalidation is dispatched to a worker goroutine: the observer + // fires from the consensus stream reader while it holds the consensusclient + // mutex, so it must not touch the space cache (GetSpace/PickSpace) inline or + // it deadlocks against Watch/UnWatch. It only enqueues here. + revalMu sync.Mutex + revalPending map[string]struct{} + revalWake chan struct{} + workerWG sync.WaitGroup +} + +func (r *relay) Init(a *app.App) (err error) { + r.nodeconf = a.MustComponent(nodeconf.CName).(nodeconf.Service) + r.pool = a.MustComponent(pool.CName).(pool.Service) + r.nodeSpace = a.MustComponent(nodespace.CName).(nodespace.Service) + r.account = a.MustComponent(accountservice.CName).(accountservice.Service).Account() + r.selfPeerId = r.account.PeerId + r.ctx, r.ctxCancel = context.WithCancel(context.Background()) + r.revalPending = make(map[string]struct{}) + r.revalWake = make(chan struct{}, 1) + + m, _ := a.Component(metric.CName).(metric.Metric) + r.engine = pubsub.New(pubsub.Deps{ + Relay: r, + Membership: r, + Metric: m, + // Crypto is nil: the node relays ciphertext it cannot read. + // Peers is nil: the node routes via Relay, not a client peer provider. + }) + if err = r.engine.Init(a); err != nil { + return err + } + return pubsub.RegisterRpc(a.MustComponent(server.CName).(server.DRPCServer), r.engine) +} + +func (r *relay) Name() string { return CName } + +func (r *relay) Run(ctx context.Context) error { + r.workerWG.Add(1) + go r.revalidateWorker() + // re-check subscriber membership whenever a hosted space's ACL changes + r.nodeSpace.SetAclObserver(r.onAclUpdate) + return r.engine.Run(ctx) +} + +func (r *relay) Close(ctx context.Context) error { + // stop new notifications before tearing down the worker and engine + r.nodeSpace.SetAclObserver(nil) + r.ctxCancel() + r.workerWG.Wait() + return r.engine.Close(ctx) +} + +// +// pubsub.Relay +// + +func (r *relay) IsResponsible(spaceId string) bool { + return r.nodeconf.IsResponsible(spaceId) +} + +func (r *relay) IsResponsibleNode(spaceId, peerId string) bool { + return slices.Contains(r.nodeconf.NodeIds(spaceId), peerId) +} + +func (r *relay) OtherResponsiblePeers(ctx context.Context, spaceId string) ([]peer.Peer, error) { + var peers []peer.Peer + for _, peerId := range r.nodeconf.NodeIds(spaceId) { + if peerId == r.selfPeerId { + continue + } + p, err := r.pool.Get(ctx, peerId) + if err != nil { + log.InfoCtx(ctx, "can't dial responsible node", zap.String("peerId", peerId), zap.Error(err)) + continue + } + peers = append(peers, p) + } + return peers, nil +} + +// +// pubsub.MembershipChecker +// + +func (r *relay) CheckMember(ctx context.Context, spaceId string, identity crypto.PubKey) error { + perms, err := r.permissions(ctx, spaceId, identity) + if err != nil { + return err + } + if perms.NoPermissions() { + return errNotMember + } + return nil +} + +func (r *relay) permissions(ctx context.Context, spaceId string, identity crypto.PubKey) (list.AclPermissions, error) { + sp, err := r.nodeSpace.GetSpace(ctx, spaceId) + if err != nil { + return list.AclPermissionsNone, err + } + acl := sp.Acl() + acl.RLock() + defer acl.RUnlock() + return acl.AclState().Permissions(identity), nil +} + +// +// ACL-change eviction (DESIGN ยง6.4) +// + +// onAclUpdate runs on the consensus stream-reader goroutine while it holds the +// consensusclient mutex. It must be non-blocking and must not touch the space +// cache (that would deadlock against Watch/UnWatch), so it only records the +// spaceId; the worker does the actual revalidation. +func (r *relay) onAclUpdate(spaceId string) { + r.revalMu.Lock() + r.revalPending[spaceId] = struct{}{} + r.revalMu.Unlock() + select { + case r.revalWake <- struct{}{}: + default: + } +} + +func (r *relay) revalidateWorker() { + defer r.workerWG.Done() + for { + select { + case <-r.ctx.Done(): + return + case <-r.revalWake: + } + r.revalMu.Lock() + pending := r.revalPending + r.revalPending = make(map[string]struct{}) + r.revalMu.Unlock() + for spaceId := range pending { + r.revalidateSpace(spaceId) + } + } +} + +func (r *relay) revalidateSpace(spaceId string) { + ctx, cancel := context.WithTimeout(r.ctx, revalidateTimeout) + defer cancel() + // PickSpace never loads: a miss means the space isn't hosted right now, so it + // has no consensus watcher and no interest to revalidate here โ€” members are + // re-checked on the next subscribe/publish. Avoids resurrecting an evicted space. + sp, err := r.nodeSpace.PickSpace(ctx, spaceId) + if err != nil { + return + } + acl := sp.Acl() + acl.RLock() + members := make(map[string]struct{}) + for _, acc := range acl.AclState().CurrentAccounts() { + if !acc.Permissions.NoPermissions() { + members[acc.PubKey.Account()] = struct{}{} + } + } + acl.RUnlock() + r.engine.RevalidateMembers(spaceId, func(account string) bool { + _, ok := members[account] + return ok + }) +} diff --git a/nodespace/pubsubrelay/pubsubrelay_test.go b/nodespace/pubsubrelay/pubsubrelay_test.go new file mode 100644 index 00000000..ef48e812 --- /dev/null +++ b/nodespace/pubsubrelay/pubsubrelay_test.go @@ -0,0 +1,167 @@ +package pubsubrelay + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/anyproto/any-sync/net/pool/mock_pool" + "github.com/anyproto/any-sync/net/rpc/rpctest" + "github.com/anyproto/any-sync/nodeconf/mock_nodeconf" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/anyproto/any-sync-node/nodespace" + "github.com/anyproto/any-sync-node/nodespace/mock_nodespace" +) + +func newRelay(t *testing.T) (*relay, *mock_nodeconf.MockService, *mock_pool.MockService) { + ctrl := gomock.NewController(t) + t.Cleanup(ctrl.Finish) + nc := mock_nodeconf.NewMockService(ctrl) + p := mock_pool.NewMockService(ctrl) + r := &relay{ + nodeconf: nc, + pool: p, + selfPeerId: "self", + } + return r, nc, p +} + +func TestRelayIsResponsible(t *testing.T) { + r, nc, _ := newRelay(t) + nc.EXPECT().IsResponsible("space1").Return(true) + nc.EXPECT().IsResponsible("space2").Return(false) + assert.True(t, r.IsResponsible("space1")) + assert.False(t, r.IsResponsible("space2")) +} + +func TestRelayIsResponsibleNode(t *testing.T) { + r, nc, _ := newRelay(t) + nc.EXPECT().NodeIds("space1").Return([]string{"nodeA", "nodeB"}).AnyTimes() + assert.True(t, r.IsResponsibleNode("space1", "nodeA")) + assert.True(t, r.IsResponsibleNode("space1", "nodeB")) + assert.False(t, r.IsResponsibleNode("space1", "clientX")) +} + +func TestRelayOtherResponsiblePeersExcludesSelf(t *testing.T) { + r, nc, p := newRelay(t) + // nodeconf.NodeIds already excludes self, but the relay also self-skips + // defensively; this stub includes "self" to exercise that guard. + nc.EXPECT().NodeIds("space1").Return([]string{"self", "nodeA", "nodeB"}) + pa := rpctest.MockPeer{} + pb := rpctest.MockPeer{} + p.EXPECT().Get(gomock.Any(), "nodeA").Return(pa, nil) + p.EXPECT().Get(gomock.Any(), "nodeB").Return(pb, nil) + // self is never dialed + + peers, err := r.OtherResponsiblePeers(context.Background(), "space1") + require.NoError(t, err) + require.Len(t, peers, 2) +} + +func TestRelayOtherResponsiblePeersSkipsUndialable(t *testing.T) { + r, nc, p := newRelay(t) + nc.EXPECT().NodeIds("space1").Return([]string{"nodeA", "nodeB"}) + pb := rpctest.MockPeer{} + p.EXPECT().Get(gomock.Any(), "nodeA").Return(nil, errors.New("unreachable")) + p.EXPECT().Get(gomock.Any(), "nodeB").Return(pb, nil) + + peers, err := r.OtherResponsiblePeers(context.Background(), "space1") + require.NoError(t, err) + require.Len(t, peers, 1, "unreachable node is skipped, not fatal") +} + +func TestRelayOtherResponsiblePeersSingleNode(t *testing.T) { + r, nc, _ := newRelay(t) + // only self is responsible (small network): empty result, no error + nc.EXPECT().NodeIds("space1").Return([]string{"self"}) + peers, err := r.OtherResponsiblePeers(context.Background(), "space1") + require.NoError(t, err) + require.Empty(t, peers) +} + +// TestOnAclUpdateNonBlocking is the regression test for the consensus-mutex +// deadlock: onAclUpdate runs on the consensus stream reader while it holds the +// consensusclient mutex, so it must only enqueue โ€” never touch the space cache +// synchronously (which would deadlock against Watch/UnWatch). With no worker +// running, the space service must not be called at all. +func TestOnAclUpdateNonBlocking(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + ns := mock_nodespace.NewMockService(ctrl) + // no PickSpace/GetSpace expectations: any synchronous cache access fails the test + r := &relay{ + nodeSpace: ns, + revalPending: make(map[string]struct{}), + revalWake: make(chan struct{}, 1), + } + + done := make(chan struct{}) + go func() { + r.onAclUpdate("space1") + r.onAclUpdate("space1") // dedup + r.onAclUpdate("space2") + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("onAclUpdate blocked โ€” must be non-blocking on the consensus goroutine") + } + + r.revalMu.Lock() + defer r.revalMu.Unlock() + require.Len(t, r.revalPending, 2) + require.Contains(t, r.revalPending, "space1") + require.Contains(t, r.revalPending, "space2") +} + +// TestRevalidateSpaceMissIsNoop verifies a PickSpace miss (space not hosted / +// closing) is a no-op that never resurrects the space or touches the engine. +func TestRevalidateSpaceMissIsNoop(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + ns := mock_nodespace.NewMockService(ctrl) + ns.EXPECT().PickSpace(gomock.Any(), "space1").Return(nil, errors.New("not found")) + r := &relay{ + nodeSpace: ns, + // engine intentionally nil: a miss must return before touching it + } + r.ctx, r.ctxCancel = context.WithCancel(context.Background()) + defer r.ctxCancel() + require.NotPanics(t, func() { r.revalidateSpace("space1") }) +} + +// TestRevalidateWorkerDrainsQueue verifies the worker drains enqueued spaceIds +// and calls PickSpace off the enqueuing goroutine. +func TestRevalidateWorkerDrainsQueue(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + ns := mock_nodespace.NewMockService(ctrl) + picked := make(chan string, 1) + ns.EXPECT().PickSpace(gomock.Any(), "space1").DoAndReturn( + func(_ context.Context, id string) (nodespace.NodeSpace, error) { + picked <- id + return nil, errors.New("miss") + }) + r := &relay{ + nodeSpace: ns, + revalPending: make(map[string]struct{}), + revalWake: make(chan struct{}, 1), + } + r.ctx, r.ctxCancel = context.WithCancel(context.Background()) + r.workerWG.Add(1) + go r.revalidateWorker() + defer func() { r.ctxCancel(); r.workerWG.Wait() }() + + r.onAclUpdate("space1") + select { + case id := <-picked: + require.Equal(t, "space1", id) + case <-time.After(2 * time.Second): + t.Fatal("worker did not drain the queue") + } +} diff --git a/nodespace/service.go b/nodespace/service.go index b434f66d..0a5ee408 100644 --- a/nodespace/service.go +++ b/nodespace/service.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "sync/atomic" "time" "github.com/anyproto/any-sync/app" @@ -44,6 +45,10 @@ type Service interface { EvictSpace(ctx context.Context, id string) error Cache() ocache.OCache GetStats(ctx context.Context, id string, treeTop int) (nodestorage.SpaceStats, error) + // SetAclObserver registers a callback invoked with a spaceId whenever that + // space's ACL changes (new consensus records applied). Used by the pubsub + // relay to re-check subscriber membership. Only one observer is supported. + SetAclObserver(observer func(spaceId string)) app.ComponentRunnable } @@ -58,6 +63,7 @@ type service struct { nodeHead nodehead.NodeHead metric metric.Metric coordClient coordinatorclient.CoordinatorClient + aclObserver atomic.Pointer[func(spaceId string)] } func (s *service) Init(a *app.App) (err error) { @@ -151,7 +157,7 @@ func (s *service) loadSpace(ctx context.Context, id string) (value ocache.Object } return } - ns, err := newNodeSpace(cc, s.consClient, s.spaceStorageProvider) + ns, err := newNodeSpace(cc, s.consClient, s.spaceStorageProvider, s.onAclUpdate) if err != nil { return } @@ -161,6 +167,22 @@ func (s *service) loadSpace(ctx context.Context, id string) (value ocache.Object return ns, nil } +func (s *service) SetAclObserver(observer func(spaceId string)) { + if observer == nil { + s.aclObserver.Store(nil) + return + } + s.aclObserver.Store(&observer) +} + +// onAclUpdate is passed to each loaded space and invoked when its ACL changes, +// forwarding to the registered observer if any. +func (s *service) onAclUpdate(spaceId string) { + if observer := s.aclObserver.Load(); observer != nil { + (*observer)(spaceId) + } +} + func (s *service) Close(ctx context.Context) (err error) { return s.spaceCache.Close() } diff --git a/nodespace/space.go b/nodespace/space.go index 3365c30f..a54b9b56 100644 --- a/nodespace/space.go +++ b/nodespace/space.go @@ -19,11 +19,12 @@ type NodeSpace interface { commonspace.Space } -func newNodeSpace(cc commonspace.Space, consClient consensusclient.Service, nodeStorage nodestorage.NodeStorage) (*nodeSpace, error) { +func newNodeSpace(cc commonspace.Space, consClient consensusclient.Service, nodeStorage nodestorage.NodeStorage, onAclUpdate func(spaceId string)) (*nodeSpace, error) { return &nodeSpace{ Space: cc, consClient: consClient, nodeStorage: nodeStorage, + onAclUpdate: onAclUpdate, log: log.With(zap.String("spaceId", cc.Id())), }, nil } @@ -32,22 +33,29 @@ type nodeSpace struct { commonspace.Space consClient consensusclient.Service nodeStorage nodestorage.NodeStorage + onAclUpdate func(spaceId string) log logger.CtxLogger } func (s *nodeSpace) AddConsensusRecords(recs []*consensusproto.RawRecordWithId) { log := s.log.With(zap.Int("len(records)", len(recs)), zap.String("firstId", recs[0].Id)) s.Acl().Lock() - defer s.Acl().Unlock() for i := 0; i < len(recs)/2; i++ { recs[i], recs[len(recs)-i-1] = recs[len(recs)-i-1], recs[i] } err := s.Acl().AddRawRecords(recs) + s.Acl().Unlock() if err != nil { log.Warn("failed to add consensus records", zap.Error(err)) } else { log.Debug("added consensus records") } + // notify observers (pubsub relay) outside the acl lock so they can re-read it. + // fire even on error: AddRawRecords applies records one by one, so a partial + // batch may already have changed membership before failing. + if s.onAclUpdate != nil { + s.onAclUpdate(s.Id()) + } } func (s *nodeSpace) AddConsensusError(err error) {