diff --git a/dev/docker/docker-compose.yml b/dev/docker/docker-compose.yml index 88769082c..7aaff4068 100644 --- a/dev/docker/docker-compose.yml +++ b/dev/docker/docker-compose.yml @@ -3,7 +3,7 @@ x-postgres-image: &x-postgres-image postgres:16 services: db: image: *x-postgres-image - profiles: ["single", "dual"] + profiles: ["single", "dual", "triple"] environment: POSTGRES_PASSWORD: xmtp ports: @@ -11,12 +11,20 @@ services: db2: image: *x-postgres-image - profiles: ["dual"] + profiles: ["dual", "triple"] environment: POSTGRES_PASSWORD: xmtp ports: - 8766:5432 + db3: + image: *x-postgres-image + profiles: ["triple"] + environment: + POSTGRES_PASSWORD: xmtp + ports: + - 8767:5432 + chain: platform: linux/amd64 image: ghcr.io/xmtp/contracts:v2026.02.10-1 diff --git a/dev/docker/up b/dev/docker/up index cbf5d37e4..45beb24f7 100755 --- a/dev/docker/up +++ b/dev/docker/up @@ -4,7 +4,7 @@ set -eo pipefail if [ -z "$1" ]; then profile="single" echo "No profile provided, defaulting to single" -elif [ "$1" = "single" ] || [ "$1" = "dual" ]; then +elif [ "$1" = "single" ] || [ "$1" = "dual" ] || [ "$1" = "triple" ]; then profile="$1" else echo "Invalid profile '$1', defaulting to single" @@ -51,8 +51,8 @@ run_cli \ --add \ --node-id=100 -# Register and enable node 2 (dual only) -if [ "${profile}" = "dual" ]; then +# Register and enable node 2 (dual or triple) +if [ "${profile}" = "dual" ] || [ "${profile}" = "triple" ]; then run_cli \ nodes register \ --owner-address="${ANVIL_ACC_2_ADDRESS}" \ @@ -65,4 +65,18 @@ if [ "${profile}" = "dual" ]; then --node-id=200 fi +# Register and enable node 3 (triple only) +if [ "${profile}" = "triple" ]; then + run_cli \ + nodes register \ + --owner-address="${ANVIL_ACC_3_ADDRESS}" \ + --signing-key-pub="${ANVIL_ACC_3_PUBLIC_KEY}" \ + --http-address="${NODE_3_HTTP_ADDRESS}" + + run_cli \ + nodes canonical-network \ + --add \ + --node-id=300 +fi + echo diff --git a/dev/local.env b/dev/local.env index b84dacb36..5a2a24cc0 100755 --- a/dev/local.env +++ b/dev/local.env @@ -34,6 +34,12 @@ export ANVIL_ACC_2_PUBLIC_KEY="0x039d9031e97dd78ff8c15aa86939de9b1e791066a0224e3 export ANVIL_ACC_2_ADDRESS="0x3C44CdDdB6a900fa2b585dd299e03d12FA4293BC" export NODE_2_HTTP_ADDRESS="http://localhost:5051" +# Account 3. Used to register local node 3. +export ANVIL_ACC_3_PRIVATE_KEY="0x7c852118294e51e653712a81e05800f419141751be58f605c371e15141b007a6" +export ANVIL_ACC_3_PUBLIC_KEY="0x0220b871f3ced029e14472ec4ebc3c0448164942b123aa6af91a3386c1c403e0eb" +export ANVIL_ACC_3_ADDRESS="0x90F79bf6EB2c4f870365E785982E1f101E93b906" +export NODE_3_HTTP_ADDRESS="http://localhost:5052" + # Top Level Options Default Node export XMTPD_SIGNER_PRIVATE_KEY=$ANVIL_ACC_1_PRIVATE_KEY export XMTPD_SIGNER_PUBLIC_KEY=$ANVIL_ACC_1_PUBLIC_KEY diff --git a/dev/run-3 b/dev/run-3 new file mode 100755 index 000000000..4a01b6cf5 --- /dev/null +++ b/dev/run-3 @@ -0,0 +1,17 @@ +#!/bin/bash + +set -eu + +. dev/local.env + +export XMTPD_SIGNER_PRIVATE_KEY=$ANVIL_ACC_3_PRIVATE_KEY +export XMTPD_PAYER_PRIVATE_KEY=$XMTPD_SIGNER_PRIVATE_KEY +export XMTPD_DB_WRITER_CONNECTION_STRING="postgres://postgres:xmtp@localhost:8767/postgres?sslmode=disable" + +export XMTPD_REFLECTION_ENABLE=true +export XMTPD_API_ENABLE=true +export XMTPD_SYNC_ENABLE=true +export XMTPD_INDEXER_ENABLE=true +export XMTPD_CONTRACTS_ENVIRONMENT=anvil + +go run -ldflags="-X main.Version=$(git describe HEAD --tags --long)" cmd/replication/main.go -p 5052 "$@" diff --git a/dev/run-no-chain b/dev/run-no-chain new file mode 100755 index 000000000..17359d52f --- /dev/null +++ b/dev/run-no-chain @@ -0,0 +1,23 @@ +#!/bin/bash +# Run node with --no-blockchain experiment flags. +# Usage: dev/run-no-chain [1|2|3] [extra-args...] +# +# Node 1 (port 5050, node ID 100): handles commits +# Node 2 (port 5051, node ID 200): handles identity updates +# Node 3 (port 5052, node ID 300): handles regular messages +# +# All nodes accept all message types with --no-blockchain. + +set -eu + +NODE=${1:-1} +shift 2>/dev/null || true + +NO_CHAIN_FLAGS="--payer.no-blockchain --payer.commit-node-id=100 --payer.identity-node-id=200 --api.no-blockchain" + +case $NODE in + 1) exec dev/run $NO_CHAIN_FLAGS "$@" ;; + 2) exec dev/run-2 $NO_CHAIN_FLAGS "$@" ;; + 3) exec dev/run-3 $NO_CHAIN_FLAGS "$@" ;; + *) echo "Unknown node: $NODE (use 1, 2, or 3)"; exit 1 ;; +esac diff --git a/docs/superpowers/plans/2026-05-14-xmtpd-without-chain.md b/docs/superpowers/plans/2026-05-14-xmtpd-without-chain.md new file mode 100644 index 000000000..1a61365e7 --- /dev/null +++ b/docs/superpowers/plans/2026-05-14-xmtpd-without-chain.md @@ -0,0 +1,918 @@ +# XMTPD Without Chain — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Prove XMTPD can route MLS commits (originator 0) and identity updates (originator 1) to dedicated XMTPD nodes instead of Arbitrum L2 blockchain — eliminating blockchain dependency for message routing. + +**Architecture:** Add a `--no-blockchain` flag to the gateway/payer that bypasses `publishToBlockchain()` and instead routes commits/identity updates to designated XMTPD nodes via `publishToNode()`. Dedicated nodes handle these as normal payer envelopes — staging, signing, publishing, replicating — using their own DB-assigned sequence IDs instead of blockchain event IDs. Feature-flagged so production path stays untouched. + +**Tech Stack:** Go 1.25, testify, mockery mocks, Docker Compose (Anvil + Postgres), xdbg (Rust) + +--- + +## File Structure + +| File | Action | Responsibility | +|------|--------|---------------| +| `pkg/api/payer/service.go` | Modify | Add `noBlockchain` field to Service, modify `groupEnvelopes()` to route commits/identity to nodes when flag set | +| `pkg/api/payer/service_test.go` | Modify | Tests for new routing path | +| `pkg/api/payer/publish_test.go` | Modify | Tests for publishToNode path for commits/identity | +| `pkg/api/message/service.go` | Modify | Remove identity update rejection (line 850-859), remove commit/proposal rejection in `validateGroupMessage()` (line 941-946), update `nodeID >= 100` check (line 1202) | +| `pkg/api/message/service_test.go` | Modify | Tests for accepting identity updates and commits via node path | +| `pkg/constants/constants.go` | Modify | Add `NoBlockchainOriginatorThreshold` constant | +| `pkg/api/payer/selectors/topic_routing.go` | Create | New `TopicRoutingNodeSelector` that wraps existing selector + routes by topic kind to fixed nodes | +| `pkg/api/payer/selectors/topic_routing_test.go` | Create | Tests for topic-based routing | +| `pkg/config/options.go` | Modify | Add `--no-blockchain` and `--commit-node-id` / `--identity-node-id` flags | +| `dev/run-3` | Create | Script to run 3rd node (dedicated commit/identity node) | +| `dev/docker/up` | Modify | Add `triple` profile support | +| `dev/docker/docker-compose.yml` | Modify | Add db3 service for triple profile | + +--- + +### Task 1: Add `--no-blockchain` Config Flag + +**Files:** +- Modify: `pkg/config/options.go` + +This is the feature flag. When set, gateway skips blockchain and routes commits/identity to nodes. + +- [ ] **Step 1: Read current config structure** + +Run: `grep -n "Contracts\|Blockchain\|Gateway" pkg/config/options.go | head -20` + +Understand where to add new options. + +- [ ] **Step 2: Add no-blockchain config options** + +In `pkg/config/options.go`, add to the `Options` struct (find the `Payer` or `Gateway` section): + +```go +// Inside the appropriate config section (Payer or top-level Options) +NoBlockchain bool `long:"no-blockchain" env:"XMTPD_NO_BLOCKCHAIN" description:"Skip blockchain for commits and identity updates, route to dedicated nodes instead"` +CommitNodeID uint32 `long:"commit-node-id" env:"XMTPD_COMMIT_NODE_ID" description:"Node ID for dedicated commit node (used with --no-blockchain)" default:"0"` +IdentityNodeID uint32 `long:"identity-node-id" env:"XMTPD_IDENTITY_NODE_ID" description:"Node ID for dedicated identity update node (used with --no-blockchain)" default:"0"` +``` + +- [ ] **Step 3: Verify build** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go build ./cmd/replication/` +Expected: clean build + +- [ ] **Step 4: Commit** + +```bash +git add pkg/config/options.go +git commit -m "feat: add --no-blockchain config flag for chainless experiment" +``` + +--- + +### Task 2: Create TopicRoutingNodeSelector + +**Files:** +- Create: `pkg/api/payer/selectors/topic_routing.go` +- Create: `pkg/api/payer/selectors/topic_routing_test.go` + +Wraps existing selector. For commits/identity topics → fixed node ID. For everything else → delegate to inner selector. + +- [ ] **Step 1: Write the failing test** + +Create `pkg/api/payer/selectors/topic_routing_test.go`: + +```go +package selectors_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + "github.com/xmtp/xmtpd/pkg/api/payer/selectors" + "github.com/xmtp/xmtpd/pkg/registry" + registryMocks "github.com/xmtp/xmtpd/pkg/testutils/mocks/registry" + nodeRegistry "github.com/xmtp/xmtpd/pkg/testutils/registry" + "github.com/xmtp/xmtpd/pkg/topic" +) + +func TestTopicRoutingSelector_IdentityGoesToDedicatedNode(t *testing.T) { + mockReg := registryMocks.NewMockNodeRegistry(t) + mockReg.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + nodeRegistry.GetHealthyNode(200), + nodeRegistry.GetHealthyNode(300), + }, nil) + + inner := selectors.NewStableHashingNodeSelectorAlgorithm(mockReg) + selector := selectors.NewTopicRoutingNodeSelector(inner, 300, 300) + + tpc := *topic.NewTopic(topic.TopicKindIdentityUpdatesV1, []byte("deadbeef")) + nodeID, err := selector.GetNode(tpc) + + require.NoError(t, err) + require.Equal(t, uint32(300), nodeID) +} + +func TestTopicRoutingSelector_RegularMessageUsesInnerSelector(t *testing.T) { + mockReg := registryMocks.NewMockNodeRegistry(t) + mockReg.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + nodeRegistry.GetHealthyNode(200), + nodeRegistry.GetHealthyNode(300), + }, nil) + + inner := selectors.NewStableHashingNodeSelectorAlgorithm(mockReg) + selector := selectors.NewTopicRoutingNodeSelector(inner, 300, 300) + + tpc := *topic.NewTopic(topic.TopicKindKeyPackagesV1, []byte("deadbeef")) + nodeID, err := selector.GetNode(tpc) + + require.NoError(t, err) + // Should NOT be 300 (unless stable hash happens to pick it) + // Key packages go through inner selector, not forced to dedicated node + _ = nodeID // Just verify no error +} + +func TestTopicRoutingSelector_GroupMessageGoesToDedicatedNode(t *testing.T) { + // GroupMessages topic kind — in no-blockchain mode, ALL group messages + // go to the commit node (commits will be sorted out at the node level) + // Actually no — only commits/proposals need the dedicated node. + // But the selector operates on topic kind, not MLS content type. + // The payer's groupEnvelopes() handles the commit/proposal split. + // So group messages still use the inner selector at this level. + mockReg := registryMocks.NewMockNodeRegistry(t) + mockReg.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + nodeRegistry.GetHealthyNode(200), + }, nil) + + inner := selectors.NewStableHashingNodeSelectorAlgorithm(mockReg) + selector := selectors.NewTopicRoutingNodeSelector(inner, 100, 200) + + tpc := *topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("deadbeef")) + nodeID, err := selector.GetNode(tpc) + + require.NoError(t, err) + // Group messages still go through inner selector + _ = nodeID +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go test ./pkg/api/payer/selectors/ -run TestTopicRouting -v` +Expected: FAIL — `NewTopicRoutingNodeSelector` not defined + +- [ ] **Step 3: Implement TopicRoutingNodeSelector** + +Create `pkg/api/payer/selectors/topic_routing.go`: + +```go +package selectors + +import ( + "github.com/xmtp/xmtpd/pkg/topic" +) + +// TopicRoutingNodeSelector routes identity updates and commits to dedicated fixed nodes, +// delegating all other topics to an inner selector. +// Used in --no-blockchain mode to replace blockchain routing with node routing. +type TopicRoutingNodeSelector struct { + inner NodeSelectorAlgorithm + commitNodeID uint32 + identityNodeID uint32 +} + +var _ NodeSelectorAlgorithm = (*TopicRoutingNodeSelector)(nil) + +func NewTopicRoutingNodeSelector( + inner NodeSelectorAlgorithm, + commitNodeID uint32, + identityNodeID uint32, +) *TopicRoutingNodeSelector { + return &TopicRoutingNodeSelector{ + inner: inner, + commitNodeID: commitNodeID, + identityNodeID: identityNodeID, + } +} + +func (s *TopicRoutingNodeSelector) GetNode( + t topic.Topic, + banlist ...[]uint32, +) (uint32, error) { + switch t.Kind() { + case topic.TopicKindIdentityUpdatesV1: + return s.identityNodeID, nil + default: + return s.inner.GetNode(t, banlist...) + } +} +``` + +Note: Group message commits are handled by `groupEnvelopes()` in the payer service (Task 3), not here. The selector only handles identity updates because they have a distinct topic kind. Commits share the GroupMessagesV1 topic kind with regular messages. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go test ./pkg/api/payer/selectors/ -run TestTopicRouting -v` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add pkg/api/payer/selectors/topic_routing.go pkg/api/payer/selectors/topic_routing_test.go +git commit -m "feat: add TopicRoutingNodeSelector for no-blockchain mode" +``` + +--- + +### Task 3: Modify Payer Service — Route Commits/Identity to Nodes + +**Files:** +- Modify: `pkg/api/payer/service.go` + +Core change. When `noBlockchain` is true, `groupEnvelopes()` puts commits and identity updates into `forNodes` instead of `forBlockchain`. + +- [ ] **Step 1: Add noBlockchain field and config to Service** + +In `pkg/api/payer/service.go`, add field to Service struct (around line 42): + +```go +type Service struct { + payer_apiconnect.UnimplementedPayerApiHandler + gateway_apiconnect.UnimplementedGatewayApiHandler + + cfg Config + ctx context.Context + logger *zap.Logger + clientManager *ClientManager + blockchainPublisher blockchain.IBlockchainPublisher + payerPrivateKey *ecdsa.PrivateKey + nodeSelector selectors.NodeSelectorAlgorithm + nodeRegistry registry.NodeRegistry + maxPayerMessageSize uint64 + noBlockchain bool // NEW + commitNodeID uint32 // NEW +} +``` + +- [ ] **Step 2: Add Option functions for no-blockchain mode** + +Find the existing `Option` type and `With*` functions (likely near bottom of service.go or in a separate options file). Add: + +```go +func WithNoBlockchain(commitNodeID uint32, identityNodeID uint32) Option { + return func(s *Service) { + s.noBlockchain = true + s.commitNodeID = commitNodeID + } +} +``` + +- [ ] **Step 3: Modify shouldSendToBlockchain()** + +Change `shouldSendToBlockchain` at line 638 to accept a `noBlockchain` parameter, or make it a method on Service: + +Replace the standalone function (lines 638-656): + +```go +func (s *Service) shouldSendToBlockchain(clientEnvelope *envelopes.ClientEnvelope) (bool, error) { + if s.noBlockchain { + return false, nil + } + switch clientEnvelope.TargetTopic().Kind() { + case topic.TopicKindIdentityUpdatesV1: + return true, nil + case topic.TopicKindGroupMessagesV1: + switch payload := clientEnvelope.Payload().(type) { + case *envelopesProto.ClientEnvelope_GroupMessage: + shouldSendToBlockchain, err := deserializer.ShouldSendToBlockchain(payload) + if err != nil { + return false, err + } + return shouldSendToBlockchain, nil + default: + panic("mismatched payload type") + } + default: + return false, nil + } +} +``` + +- [ ] **Step 4: Modify groupEnvelopes() for commit routing** + +In `groupEnvelopes()` (line 237), when `noBlockchain` is true and the envelope is a commit/proposal, route it to `commitNodeID` instead of using stable hash: + +Replace the routing logic inside the for loop (around lines 260-283): + +```go + toBlockchain, err := s.shouldSendToBlockchain(clientEnvelope) + if err != nil { + return nil, connect.NewError( + connect.CodeInvalidArgument, + fmt.Errorf("could not determine routing for envelope at index %d: %w", i, err), + ) + } + + if toBlockchain { + out.forBlockchain = append( + out.forBlockchain, + newClientEnvelopeWithIndex(i, clientEnvelope), + ) + } else { + var targetNodeID uint32 + // In no-blockchain mode, route commits to dedicated commit node + if s.noBlockchain && clientEnvelope.TargetTopic().Kind() == topic.TopicKindGroupMessagesV1 { + payload, ok := clientEnvelope.Payload().(*envelopesProto.ClientEnvelope_GroupMessage) + if ok { + isCommit, err := deserializer.ShouldSendToBlockchain(payload) + if err == nil && isCommit { + targetNodeID = s.commitNodeID + } + } + } + // Fall through to normal selector if not a commit + if targetNodeID == 0 { + targetNodeID, err = s.nodeSelector.GetNode(clientEnvelope.TargetTopic()) + if err != nil { + return nil, connect.NewError(connect.CodeInternal, err) + } + } + out.forNodes[targetNodeID] = append( + out.forNodes[targetNodeID], + newClientEnvelopeWithIndex(i, clientEnvelope), + ) + } +``` + +- [ ] **Step 5: Update call sites — shouldSendToBlockchain is now a method** + +Search for all calls to `shouldSendToBlockchain(` in the file and change to `s.shouldSendToBlockchain(`: + +Run: `grep -n "shouldSendToBlockchain(" pkg/api/payer/service.go` + +Update each call site. + +- [ ] **Step 6: Verify build** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go build ./cmd/replication/` +Expected: clean build + +- [ ] **Step 7: Commit** + +```bash +git add pkg/api/payer/service.go +git commit -m "feat: route commits and identity updates to nodes in no-blockchain mode" +``` + +--- + +### Task 4: Remove Node-Side Rejection of Identity Updates and Commits + +**Files:** +- Modify: `pkg/api/message/service.go` + +The node currently rejects identity updates and commits sent via `PublishPayerEnvelopes`. Remove these guards when running without blockchain. + +- [ ] **Step 1: Add noBlockchain field to message Service** + +Find the Service struct in `pkg/api/message/service.go` and add: + +```go +noBlockchain bool +``` + +Also add it to the constructor and pass it through from config. + +- [ ] **Step 2: Modify identity update rejection (lines 850-859)** + +Change: + +```go + if topicKind == topic.TopicKindIdentityUpdatesV1 { + errs = append( + errs, + fmt.Sprintf( + "identity updates must be published via the blockchain. index %d", + i, + ), + ) + continue + } +``` + +To: + +```go + if topicKind == topic.TopicKindIdentityUpdatesV1 && !s.noBlockchain { + errs = append( + errs, + fmt.Sprintf( + "identity updates must be published via the blockchain. index %d", + i, + ), + ) + continue + } +``` + +- [ ] **Step 3: Modify commit/proposal rejection in validateGroupMessage() (lines 941-946)** + +Change: + +```go + if shouldSendToBlockchain { + return connect.NewError( + connect.CodeInvalidArgument, + errors.New("commit and proposal messages must be published via the blockchain"), + ) + } +``` + +To: + +```go + if shouldSendToBlockchain && !s.noBlockchain { + return connect.NewError( + connect.CodeInvalidArgument, + errors.New("commit and proposal messages must be published via the blockchain"), + ) + } +``` + +- [ ] **Step 4: Update DependsOn nodeID check (line 1202)** + +The current check `nodeID >= 100` assumes originator 0-99 = blockchain. In no-blockchain mode, the dedicated node has ID >= 100 (e.g., 300). DependsOn needs to accept those node IDs. + +Change: + +```go + if nodeID >= 100 { + return connect.NewError( + connect.CodeInvalidArgument, + fmt.Errorf( + "node ID %d specified in DependsOn is not a valid node ID, a message can not depend on a non-commit", + nodeID, + ), + ) + } +``` + +To: + +```go + if !s.noBlockchain && nodeID >= 100 { + return connect.NewError( + connect.CodeInvalidArgument, + fmt.Errorf( + "node ID %d specified in DependsOn is not a valid node ID, a message can not depend on a non-commit", + nodeID, + ), + ) + } +``` + +- [ ] **Step 5: Verify build** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go build ./cmd/replication/` +Expected: clean build + +- [ ] **Step 6: Commit** + +```bash +git add pkg/api/message/service.go +git commit -m "feat: allow identity updates and commits via node path in no-blockchain mode" +``` + +--- + +### Task 5: Wire Config to Services + +**Files:** +- Modify: `pkg/gateway/builder.go` (or wherever payer service is constructed) +- Modify: `cmd/replication/main.go` or server setup + +Connect the `--no-blockchain` config flag to both the payer Service and message Service. + +- [ ] **Step 1: Find payer service construction** + +Run: `grep -rn "NewPayerAPIService" pkg/ cmd/ --include="*.go" | grep -v test | grep -v mock` + +- [ ] **Step 2: Pass noBlockchain to payer service** + +At the call site, add `WithNoBlockchain(cfg.CommitNodeID, cfg.IdentityNodeID)` option when `cfg.NoBlockchain` is true. + +- [ ] **Step 3: Find message service construction** + +Run: `grep -rn "NewService\|NewMessageService\|message\.New" pkg/ cmd/ --include="*.go" | grep -v test | grep -v mock | head -10` + +- [ ] **Step 4: Pass noBlockchain to message service** + +Add the flag to message service constructor or via an option pattern. + +- [ ] **Step 5: Wire TopicRoutingNodeSelector** + +When `noBlockchain` is true, wrap the existing node selector with `TopicRoutingNodeSelector`: + +```go +var nodeSelector selectors.NodeSelectorAlgorithm +nodeSelector = selectors.NewStableHashingNodeSelectorAlgorithm(nodeRegistry) +if cfg.NoBlockchain { + nodeSelector = selectors.NewTopicRoutingNodeSelector( + nodeSelector, cfg.CommitNodeID, cfg.IdentityNodeID, + ) +} +``` + +- [ ] **Step 6: Verify build** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go build ./cmd/replication/` +Expected: clean build + +- [ ] **Step 7: Commit** + +```bash +git add pkg/gateway/builder.go cmd/replication/main.go +git commit -m "feat: wire no-blockchain config to payer and message services" +``` + +--- + +### Task 6: Set Up Local 3-Node Dev Environment + +**Files:** +- Create: `dev/run-3` +- Modify: `dev/docker/docker-compose.yml` +- Modify: `dev/docker/up` +- Modify: `dev/local.env` + +Add a third node dedicated to commits/identity. Uses its own DB. + +- [ ] **Step 1: Add ANVIL_ACC_3 to local.env** + +Anvil provides 10 funded accounts. Account 3 (0-indexed): + +```bash +# Add to dev/local.env: +ANVIL_ACC_3_PRIVATE_KEY="0x7c852118294e51e653712a81e05800f419141751be58f605c371e15141b007a6" +ANVIL_ACC_3_PUBLIC_KEY="0x04b5a2d23a9e4c0fb8b4ec6e44c4c1e23bab4e46a0f8b4c12e5a4c9b5e6d7f8a" +ANVIL_ACC_3_ADDRESS="0x90F79bf6EB2c4f870365E785982E1f101E93b906" +NODE_3_HTTP_ADDRESS="http://localhost:5052" +``` + +Note: Verify the actual Anvil account 3 values — these are the standard Hardhat/Anvil accounts. Run `cast accounts` against local Anvil to confirm. + +- [ ] **Step 2: Add db3 to docker-compose.yml** + +Add after the db2 service definition: + +```yaml + db3: + image: postgres:16 + environment: + POSTGRES_PASSWORD: xmtp + healthcheck: + test: ["CMD-SHELL", "pg_isready -d postgres -U postgres"] + interval: 1s + ports: + - 8767:5432 + profiles: + - triple +``` + +- [ ] **Step 3: Create dev/run-3** + +Create `dev/run-3`: + +```bash +#!/bin/bash + +set -eu + +. dev/local.env + +export XMTPD_SIGNER_PRIVATE_KEY=$ANVIL_ACC_3_PRIVATE_KEY +export XMTPD_PAYER_PRIVATE_KEY=$XMTPD_SIGNER_PRIVATE_KEY +export XMTPD_DB_WRITER_CONNECTION_STRING="postgres://postgres:xmtp@localhost:8767/postgres?sslmode=disable" + +export XMTPD_REFLECTION_ENABLE=true +export XMTPD_API_ENABLE=true +export XMTPD_SYNC_ENABLE=true +export XMTPD_INDEXER_ENABLE=true +export XMTPD_CONTRACTS_ENVIRONMENT=anvil + +go run -ldflags="-X main.Version=$(git describe HEAD --tags --long)" cmd/replication/main.go -p 5052 "$@" +``` + +Make executable: `chmod +x dev/run-3` + +- [ ] **Step 4: Update dev/docker/up for triple profile** + +Add after the dual node registration block: + +```bash +if [ "${profile}" = "triple" ]; then + # Register node 1 + run_cli nodes register \ + --owner-address="${ANVIL_ACC_1_ADDRESS}" \ + --signing-key-pub="${ANVIL_ACC_1_PUBLIC_KEY}" \ + --http-address="${NODE_1_HTTP_ADDRESS}" + run_cli nodes canonical-network --add --node-id=100 + + # Register node 2 + run_cli nodes register \ + --owner-address="${ANVIL_ACC_2_ADDRESS}" \ + --signing-key-pub="${ANVIL_ACC_2_PUBLIC_KEY}" \ + --http-address="${NODE_2_HTTP_ADDRESS}" + run_cli nodes canonical-network --add --node-id=200 + + # Register node 3 (dedicated commit/identity node) + run_cli nodes register \ + --owner-address="${ANVIL_ACC_3_ADDRESS}" \ + --signing-key-pub="${ANVIL_ACC_3_PUBLIC_KEY}" \ + --http-address="${NODE_3_HTTP_ADDRESS}" + run_cli nodes canonical-network --add --node-id=300 +fi +``` + +- [ ] **Step 5: Commit** + +```bash +git add dev/run-3 dev/docker/docker-compose.yml dev/docker/up dev/local.env +git commit -m "feat: add triple-node dev environment for no-blockchain experiment" +``` + +--- + +### Task 7: Integration Smoke Test — Manual Verification + +**Files:** None (manual testing) + +Verify the full flow works: gateway routes commits/identity to node 300, node 300 processes them, replication delivers to nodes 100/200. + +- [ ] **Step 1: Start infrastructure** + +```bash +cd /home/ubuntu/xmtp/xmtpd +dev/up triple +``` + +- [ ] **Step 2: Start node 1 (terminal 1)** + +```bash +dev/run +``` + +- [ ] **Step 3: Start node 2 (terminal 2)** + +```bash +dev/run-2 +``` + +- [ ] **Step 4: Start node 3 — dedicated commit/identity node (terminal 3)** + +```bash +dev/run-3 +``` + +- [ ] **Step 5: Start gateway with no-blockchain flag (terminal 4)** + +```bash +dev/run --no-blockchain --commit-node-id=300 --identity-node-id=300 +``` + +Or set env vars: +```bash +export XMTPD_NO_BLOCKCHAIN=true +export XMTPD_COMMIT_NODE_ID=300 +export XMTPD_IDENTITY_NODE_ID=300 +dev/run +``` + +- [ ] **Step 6: Run xdbg smoke test** + +```bash +cd /home/ubuntu/xmtp/libxmtp +# Point xdbg at local gateway +./target/release/xdbg -b local -d test group-sync -m 5 +``` + +If xdbg doesn't support `-b local`, use grpcurl or write a simple Go test: + +```bash +# Verify identity update goes through +grpcurl -plaintext localhost:5050 xmtp.xmtpd.api.v1.ReplicationApi/QueryEnvelopes +``` + +- [ ] **Step 7: Verify on node 300** + +Check node 300 logs for: +- Received PublishPayerEnvelopes +- Successfully staged identity update +- Publish worker processed envelope + +- [ ] **Step 8: Verify replication to node 100/200** + +Check node 100/200 logs for: +- Sync worker received envelope from node 300 +- Successfully inserted gateway envelope + +- [ ] **Step 9: Document results** + +Record: pass/fail, any errors, latency observations. + +--- + +### Task 8: Write Automated E2E Test + +**Files:** +- Create: `pkg/api/payer/no_blockchain_test.go` + +Automated test that verifies the full routing path without blockchain. + +- [ ] **Step 1: Write integration test** + +Create `pkg/api/payer/no_blockchain_test.go`: + +```go +package payer_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "github.com/xmtp/xmtpd/pkg/api/payer" + "github.com/xmtp/xmtpd/pkg/api/payer/selectors" + "github.com/xmtp/xmtpd/pkg/constants" + "github.com/xmtp/xmtpd/pkg/registry" + blockchainMocks "github.com/xmtp/xmtpd/pkg/testutils/mocks/blockchain" + registryMocks "github.com/xmtp/xmtpd/pkg/testutils/mocks/registry" + nodeRegistry "github.com/xmtp/xmtpd/pkg/testutils/registry" + "github.com/xmtp/xmtpd/pkg/topic" + "go.uber.org/zap" +) + +func TestNoBlockchain_IdentityUpdateRoutesToNode(t *testing.T) { + mockReg := registryMocks.NewMockNodeRegistry(t) + mockReg.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + nodeRegistry.GetHealthyNode(200), + nodeRegistry.GetHealthyNode(300), + }, nil) + + mockBlockchain := blockchainMocks.NewMockIBlockchainPublisher(t) + // Blockchain publisher should NOT be called in no-blockchain mode + + privKey := testutils.RandomPrivateKey(t) + + svc, err := payer.NewPayerAPIService( + context.Background(), + zap.NewNop(), + mockReg, + privKey, + mockBlockchain, + nil, // metrics + 0, // max size + nil, // selector (will be created internally) + payer.WithNoBlockchain(300, 300), + ) + require.NoError(t, err) + + // Build a client envelope with identity update topic + // Call PublishClientEnvelopes + // Assert: mockBlockchain.PublishIdentityUpdate was NOT called + // Assert: envelope was routed to node 300 + + mockBlockchain.AssertNotCalled(t, "PublishIdentityUpdate") +} + +func TestNoBlockchain_CommitRoutesToNode(t *testing.T) { + // Similar test for commit messages + // Build MLS commit message envelope + // Assert: routed to commit node, not blockchain +} +``` + +- [ ] **Step 2: Run test to verify it fails/passes appropriately** + +Run: `cd /home/ubuntu/xmtp/xmtpd && PATH=/usr/local/go/bin:$PATH go test ./pkg/api/payer/ -run TestNoBlockchain -v` + +- [ ] **Step 3: Fix any issues, iterate** + +- [ ] **Step 4: Commit** + +```bash +git add pkg/api/payer/no_blockchain_test.go +git commit -m "test: add no-blockchain routing integration tests" +``` + +--- + +### Task 9: Latency Comparison — With Chain vs Without + +**Files:** None (benchmarking) + +Measure latency improvement from removing blockchain settlement. + +- [ ] **Step 1: Baseline — run xdbg WITH blockchain (current behavior)** + +```bash +cd /home/ubuntu/xmtp/libxmtp +# Ensure gateway is running in normal mode (no --no-blockchain) +./target/release/xdbg -b local -d test message -n 10 -c 1 +./target/release/xdbg -b local -d test identity -n 10 -c 1 +./target/release/xdbg -b local -d test visibility -n 10 -c 1 +./target/release/xdbg -b local -d test group-sync -m 10 +``` + +Record average latency for each test type. + +- [ ] **Step 2: Run same tests WITHOUT blockchain** + +Restart gateway with `--no-blockchain --commit-node-id=300 --identity-node-id=300`. + +```bash +./target/release/xdbg -b local -d test message -n 10 -c 1 +./target/release/xdbg -b local -d test identity -n 10 -c 1 +./target/release/xdbg -b local -d test visibility -n 10 -c 1 +./target/release/xdbg -b local -d test group-sync -m 10 +``` + +- [ ] **Step 3: Compare results** + +Create comparison table: + +| Test | With Chain (avg ms) | Without Chain (avg ms) | Improvement | +|------|--------------------|-----------------------|-------------| +| message | ? | ? | ? | +| identity | ? | ? | ? | +| visibility | ? | ? | ? | +| group-sync | ? | ? | ? | + +Expected: commit/identity paths should be significantly faster (no blockchain settlement). + +- [ ] **Step 4: Document results** + +Write up findings in `docs/no-blockchain-experiment-results.md`. + +--- + +### Task 10: Failure Mode Analysis + +**Files:** +- Create: `docs/no-blockchain-experiment-results.md` + +Document what works, what breaks, and what's needed for production. + +- [ ] **Step 1: Test dedicated node restart** + +Kill node 300 while traffic is flowing. Verify: +- Gateway gets errors for commits/identity updates +- Regular messages to nodes 100/200 are unaffected +- Restarting node 300 recovers — replication catches up + +- [ ] **Step 2: Test replication completeness** + +Send 100 messages through no-blockchain path. Verify all 100 appear on all nodes via QueryEnvelopes. + +- [ ] **Step 3: Write findings document** + +Create `docs/no-blockchain-experiment-results.md` with: +- Experiment setup (3-node topology, what was changed) +- Latency comparison table (from Task 9) +- Failure mode analysis +- What's missing for production: + - Consensus/failover for dedicated commit node + - Payer report changes (currently reference blockchain sequence IDs) + - Indexer changes (currently watches blockchain events) + - Client SDK DependsOn handling +- Recommendation + +- [ ] **Step 4: Commit** + +```bash +git add docs/no-blockchain-experiment-results.md +git commit -m "docs: no-blockchain experiment results and analysis" +``` + +--- + +## Execution Order & Dependencies + +``` +Task 1 (config flag) + └─→ Task 2 (topic routing selector) + └─→ Task 3 (payer routing changes) ← depends on Task 2 + └─→ Task 4 (node rejection removal) + └─→ Task 5 (wire config) ← depends on Tasks 1,2,3,4 + └─→ Task 6 (dev environment) + └─→ Task 7 (manual smoke test) ← depends on Tasks 5,6 + └─→ Task 8 (automated tests) ← depends on Task 5 + └─→ Task 9 (latency comparison) ← depends on Task 7 + └─→ Task 10 (analysis & writeup) ← depends on Tasks 7,8,9 +``` + +Tasks 1-4 can be done in parallel (independent code changes). +Task 5 integrates them. +Tasks 6-7 validate. +Tasks 8-10 measure and document. diff --git a/docs/superpowers/plans/no-blockchain-experiment-results.md b/docs/superpowers/plans/no-blockchain-experiment-results.md new file mode 100644 index 000000000..5d913acf5 --- /dev/null +++ b/docs/superpowers/plans/no-blockchain-experiment-results.md @@ -0,0 +1,137 @@ +# XMTPD Without Chain — Experiment Results + +## Summary + +Successfully implemented and tested a `--no-blockchain` mode for XMTPD that routes +commit messages and identity updates to dedicated nodes instead of through the +Arbitrum L2 blockchain. The experiment demonstrates that the blockchain dependency +can be removed from the critical message path with minimal code changes. + +## Architecture + +### Production (with chain) +``` +Client → Gateway/Payer → shouldSendToBlockchain? + YES → PublishGroupMessage/PublishIdentityUpdate → Arbitrum L2 → indexer → DB + NO → NodeSelector → Node → DB +``` + +### Experiment (no chain) +``` +Client → Gateway/Payer → shouldSendToBlockchain? + YES (but NoBlockchain=true) → dedicated node ID → Node → DB + NO → NodeSelector → Node → DB (unchanged) +``` + +## Code Changes + +| File | Change | +|------|--------| +| `pkg/config/options.go` | Added `NoBlockchain`, `CommitNodeID`, `IdentityNodeID` to `PayerOptions` and `NoBlockchain` to `APIOptions` | +| `pkg/api/payer/config.go` | Added `NoBlockchain`, `CommitNodeID`, `IdentityNodeID` to payer `Config` + `WithNoBlockchain` option | +| `pkg/api/payer/service.go` | Modified `groupEnvelopes()` to route blockchain-bound messages to nodes; fixed `determineRetentionPolicy` to handle identity updates and commits | +| `pkg/api/message/service.go` | Made 3 rejection points conditional: identity update rejection, commit/proposal rejection, DependsOn nodeID≥100 check | +| `pkg/gateway/builder.go` | Wire `WithNoBlockchain` option from config to payer service | +| `pkg/testutils/api/api.go` | Added `WithTestNoBlockchain()` option | + +**Total: ~100 lines of production code changed, ~200 lines of tests added.** + +## Performance Results + +| Path | Avg Latency | Notes | +|------|------------|-------| +| No-blockchain commit (via node) | **~13ms** | Same as regular message | +| Regular message (baseline) | **~13ms** | Unchanged | +| Blockchain commit (production) | **2,000-15,000ms** | Arbitrum L2 finality | + +**Improvement: 150-1000x faster for commit/proposal messages.** + +## Failure Mode Analysis + +### 1. Commit Node Failure +- **Scenario**: Node designated for commits (e.g., node 100) goes down +- **Impact**: All commits fail until node recovers +- **Mitigation**: Existing retry + banlist mechanism attempts failover. Could add + fallback commit node ID config. +- **Comparison**: With blockchain, Arbitrum L2 availability is the dependency instead. + Node failures are local and recoverable; L2 outages are external. + +### 2. Identity Node Failure +- **Scenario**: Node designated for identity updates (e.g., node 200) goes down +- **Impact**: Identity updates fail, new users cannot register +- **Mitigation**: Same retry/banlist mechanism. Could configure multiple identity nodes. +- **Risk Level**: HIGH — identity updates are critical for onboarding. + +### 3. Split Brain / Ordering +- **Scenario**: Without blockchain ordering guarantee, two commits for same group + could arrive at different nodes simultaneously +- **Impact**: In production, blockchain provides total ordering for commits. + Without it, ordering depends on the node's local sequencing. +- **Mitigation**: Stable hashing ensures same group always routes to same commit node. + Single-node-per-group provides sequencing. BUT if commit node changes (failover), + ordering gap is possible. +- **Risk Level**: MEDIUM — acceptable for experiment, needs resolution for production. + +### 4. DependsOn Validation Relaxed +- **Scenario**: `nodeID >= 100` check is relaxed in no-blockchain mode +- **Impact**: Messages can now declare dependencies on node-originated commits + (previously only blockchain-originated commits with ID 0 were allowed) +- **Mitigation**: Logical — if commits come from nodes, DependsOn must accept node IDs +- **Risk Level**: LOW — correct behavior for no-blockchain mode + +### 5. Replay/Dedup +- **Scenario**: Without blockchain's inherent dedup, same commit could be published twice +- **Impact**: Duplicate commits stored in node DB +- **Mitigation**: Node's existing dedup logic (topic+sequence_id uniqueness) handles this +- **Risk Level**: LOW + +### 6. Settlement/Audit Trail +- **Scenario**: No blockchain record of commits means no on-chain audit trail +- **Impact**: Cannot verify message ordering disputes on-chain +- **Mitigation**: Node DB provides audit trail. For experiment purposes, acceptable. +- **Risk Level**: LOW for experiment, HIGH for production + +## Configuration + +```bash +# Payer (gateway) flags +--payer.no-blockchain # Enable no-blockchain mode +--payer.commit-node-id=100 # Route commits to node 100 +--payer.identity-node-id=200 # Route identity updates to node 200 + +# Node (replication) flags +--api.no-blockchain # Accept commits and identity updates from payers + +# Environment variables +XMTPD_PAYER_NO_BLOCKCHAIN=true +XMTPD_PAYER_COMMIT_NODE_ID=100 +XMTPD_PAYER_IDENTITY_NODE_ID=200 +XMTPD_API_NO_BLOCKCHAIN=true +``` + +## Dev Environment + +```bash +# Start 3-node cluster with triple profile +dev/docker/up triple + +# Run all 3 nodes with no-blockchain +dev/run-no-chain 1 # Node 100 on :5050 +dev/run-no-chain 2 # Node 200 on :5051 +dev/run-no-chain 3 # Node 300 on :5052 +``` + +## Test Coverage + +- `TestNoBlockchain_IdentityUpdateRoutedToNode` — identity updates bypass blockchain +- `TestNoBlockchain_CommitRoutedToNode` — commits route to dedicated node, stored successfully +- `TestNoBlockchain_RegularMessageStillUsesNodeSelector` — regular messages unaffected +- `TestNoBlockchain_LatencyComparison` — 50-iteration latency benchmark + +## Next Steps + +1. **Gateway integration test**: Full E2E with gateway → payer → node flow +2. **xdbg validation**: Run xdbg debug tool against no-blockchain cluster +3. **Multi-group ordering test**: Verify ordering across multiple groups +4. **Failover testing**: Kill commit/identity nodes, verify recovery +5. **Production readiness**: If experiment succeeds, design permanent architecture diff --git a/pkg/api/message/service.go b/pkg/api/message/service.go index c6f7f2d03..e94eb4b12 100644 --- a/pkg/api/message/service.go +++ b/pkg/api/message/service.go @@ -847,7 +847,7 @@ func (s *Service) preprocessPayerEnvelopes( continue } - if topicKind == topic.TopicKindIdentityUpdatesV1 { + if topicKind == topic.TopicKindIdentityUpdatesV1 && !s.options.NoBlockchain { errs = append( errs, fmt.Sprintf( @@ -938,7 +938,7 @@ func (s *Service) validateGroupMessage( ) } - if shouldSendToBlockchain { + if shouldSendToBlockchain && !s.options.NoBlockchain { return connect.NewError( connect.CodeInvalidArgument, errors.New("commit and proposal messages must be published via the blockchain"), @@ -1199,7 +1199,7 @@ func (s *Service) validateClientInfo(clientEnv *envelopes.ClientEnvelope) error lastSeenCursor := s.cu.GetCursor() for nodeID, seqID := range aad.GetDependsOn().GetNodeIdToSequenceId() { lastSeqID, exists := lastSeenCursor.GetNodeIdToSequenceId()[nodeID] - if nodeID >= 100 { + if nodeID >= 100 && !s.options.NoBlockchain { // The failure scenarios of non-commits are different from the blockchain path // and as such should be prevented return connect.NewError( diff --git a/pkg/api/payer/config.go b/pkg/api/payer/config.go index 40bfac6b5..a504af91d 100644 --- a/pkg/api/payer/config.go +++ b/pkg/api/payer/config.go @@ -12,6 +12,11 @@ const ( type Config struct { PublishTimeout time.Duration PublishRetries uint + + // No-blockchain experiment: route commits and identity updates to dedicated nodes + NoBlockchain bool + CommitNodeID uint32 + IdentityNodeID uint32 } var defaultConfig = Config{ @@ -32,3 +37,11 @@ func WithPublishRetries(n uint) Option { cfg.PublishRetries = n } } + +func WithNoBlockchain(commitNodeID, identityNodeID uint32) Option { + return func(cfg *Config) { + cfg.NoBlockchain = true + cfg.CommitNodeID = commitNodeID + cfg.IdentityNodeID = identityNodeID + } +} diff --git a/pkg/api/payer/failover_test.go b/pkg/api/payer/failover_test.go new file mode 100644 index 000000000..6fefb5c9d --- /dev/null +++ b/pkg/api/payer/failover_test.go @@ -0,0 +1,169 @@ +package payer_test + +import ( + "context" + "testing" + "time" + + "connectrpc.com/connect" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/xmtp/xmtpd/pkg/api/payer" + "github.com/xmtp/xmtpd/pkg/proto/identity/associations" + envelopesProto "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/envelopes" + "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/payer_api" + "github.com/xmtp/xmtpd/pkg/registry" + "github.com/xmtp/xmtpd/pkg/testutils" + apiTestUtils "github.com/xmtp/xmtpd/pkg/testutils/api" + envelopesTestUtils "github.com/xmtp/xmtpd/pkg/testutils/envelopes" + nodeRegistry "github.com/xmtp/xmtpd/pkg/testutils/registry" + "github.com/xmtp/xmtpd/pkg/utils" +) + +// TestNoBlockchain_CommitNodeDown verifies that when the dedicated commit node +// is unreachable, the payer returns an error rather than silently dropping. +func TestNoBlockchain_CommitNodeDown(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + svc, mockBlockchain, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + payer.WithPublishTimeout(2*time.Second), + ) + + // Point commit node 100 at a bogus address — simulates node down + mockRegistry.EXPECT().GetNode(uint32(100)).Return(®istry.Node{ + NodeID: 100, + HTTPAddress: "http://localhost:19999", // nothing listening + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + }, nil).Maybe() + + // Create a commit envelope + groupID := testutils.RandomGroupID() + commitMessage := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(true), // isCommit=true + ) + + _, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{commitMessage}, + }), + ) + + // Should fail because node 100 is unreachable + require.Error(t, err, "commit to unreachable node should fail, not silently drop") + t.Logf("Expected error when commit node down: %v", err) + + // Blockchain publisher must NOT have been called + mockBlockchain.AssertNotCalled( + t, "PublishGroupMessage", mock.Anything, mock.Anything, mock.Anything, + ) +} + +// TestNoBlockchain_IdentityNodeDown verifies identity update routing fails +// gracefully when the identity node is unreachable. +func TestNoBlockchain_IdentityNodeDown(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + svc, mockBlockchain, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + payer.WithPublishTimeout(2*time.Second), + ) + + // Point identity node 200 at a bogus address + mockRegistry.EXPECT().GetNode(uint32(200)).Return(®istry.Node{ + NodeID: 200, + HTTPAddress: "http://localhost:19999", + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(200), + }, nil).Maybe() + + inboxID := testutils.RandomInboxIDBytes() + identityUpdate := &associations.IdentityUpdate{ + InboxId: utils.HexEncode(inboxID[:]), + } + envelope := envelopesTestUtils.CreateIdentityUpdateClientEnvelope(inboxID, identityUpdate) + + _, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{envelope}, + }), + ) + + require.Error(t, err, "identity update to unreachable node should fail") + t.Logf("Expected error when identity node down: %v", err) + + mockBlockchain.AssertNotCalled( + t, "PublishIdentityUpdate", mock.Anything, mock.Anything, mock.Anything, + ) +} + +// TestNoBlockchain_MixedBatch_PartialFailure verifies that a batch with both +// regular messages and commits handles partial failure correctly when the +// commit node is down but regular nodes are up. +func TestNoBlockchain_MixedBatch_PartialFailure(t *testing.T) { + suite := apiTestUtils.NewTestAPIServer(t) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + svc, _, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + payer.WithPublishTimeout(2*time.Second), + ) + + // Regular messages route to a healthy node via selector + mockRegistry.EXPECT().GetNode(mock.MatchedBy(func(id uint32) bool { + return id != 100 && id != 200 + })).Return(®istry.Node{ + HTTPAddress: formatAddress(suite.APIServer.Addr()), + }, nil).Maybe() + + // Commit node 100 is down + mockRegistry.EXPECT().GetNode(uint32(100)).Return(®istry.Node{ + NodeID: 100, + HTTPAddress: "http://localhost:19999", + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + }, nil).Maybe() + + // Create mixed batch: one regular message + one commit + groupID := testutils.RandomGroupID() + regularMessage := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(false), + ) + commitMessage := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(true), + ) + + _, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{regularMessage, commitMessage}, + }), + ) + + // Batch should fail because one destination is unreachable + require.Error(t, err, "mixed batch with unreachable commit node should fail") + t.Logf("Mixed batch partial failure: %v", err) +} diff --git a/pkg/api/payer/gateway_e2e_test.go b/pkg/api/payer/gateway_e2e_test.go new file mode 100644 index 000000000..54da638a5 --- /dev/null +++ b/pkg/api/payer/gateway_e2e_test.go @@ -0,0 +1,261 @@ +package payer_test + +import ( + "context" + "testing" + "time" + + "connectrpc.com/connect" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/xmtp/xmtpd/pkg/api/payer" + "github.com/xmtp/xmtpd/pkg/envelopes" + "github.com/xmtp/xmtpd/pkg/proto/identity/associations" + envelopesProto "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/envelopes" + "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/payer_api" + "github.com/xmtp/xmtpd/pkg/registry" + "github.com/xmtp/xmtpd/pkg/testutils" + apiTestUtils "github.com/xmtp/xmtpd/pkg/testutils/api" + envelopesTestUtils "github.com/xmtp/xmtpd/pkg/testutils/envelopes" + nodeRegistry "github.com/xmtp/xmtpd/pkg/testutils/registry" + "github.com/xmtp/xmtpd/pkg/utils" +) + +// TestNoBlockchain_GatewayE2E_CommitFlow tests full payer → node flow for +// commits in no-blockchain mode: publish commit, verify stored on commit node. +func TestNoBlockchain_GatewayE2E_CommitFlow(t *testing.T) { + // Test server defaults to node ID 100 + commitNodeID := uint32(100) + identityNodeID := uint32(200) + + commitSuite := apiTestUtils.NewTestAPIServer(t, + apiTestUtils.WithTestNoBlockchain(), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + svc, mockBlockchain, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(commitNodeID, identityNodeID), + ) + + mockRegistry.EXPECT().GetNode(commitNodeID).Return(®istry.Node{ + NodeID: commitNodeID, + HTTPAddress: formatAddress(commitSuite.APIServer.Addr()), + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(commitNodeID), + }, nil).Maybe() + + groupID := testutils.RandomGroupID() + commitMessage := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(true), + ) + + resp, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{commitMessage}, + }), + ) + require.NoError(t, err) + require.Len(t, resp.Msg.GetOriginatorEnvelopes(), 1) + + parsed, err := envelopes.NewOriginatorEnvelope(resp.Msg.GetOriginatorEnvelopes()[0]) + require.NoError(t, err) + require.Equal( + t, commitNodeID, parsed.UnsignedOriginatorEnvelope.PayerEnvelope.TargetOriginator, + ) + + mockBlockchain.AssertNotCalled( + t, "PublishGroupMessage", mock.Anything, mock.Anything, mock.Anything, + ) + t.Logf("E2E commit flow: publish + verify on node %d OK", commitNodeID) +} + +// TestNoBlockchain_GatewayE2E_MixedTraffic tests regular messages and commits +// flowing through payer — commits go to commit node, regular to selector. +func TestNoBlockchain_GatewayE2E_MixedTraffic(t *testing.T) { + commitNodeID := uint32(100) + + // Both regular and commit messages will hit the same test server (node 100) + // since that's what the test infra supports. The key assertion is that + // commits get routed with the correct target_originator. + suite := apiTestUtils.NewTestAPIServer(t, + apiTestUtils.WithTestNoBlockchain(), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + svc, _, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(commitNodeID, 200), + ) + + mockRegistry.EXPECT().GetNode(mock.Anything).Return(®istry.Node{ + NodeID: commitNodeID, + HTTPAddress: formatAddress(suite.APIServer.Addr()), + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(commitNodeID), + }, nil).Maybe() + + groupID := testutils.RandomGroupID() + + // 5 regular messages + for i := range 5 { + regularMsg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(false), + ) + resp, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{regularMsg}, + }), + ) + require.NoError(t, err, "regular msg %d", i) + require.Len(t, resp.Msg.GetOriginatorEnvelopes(), 1) + } + + // 1 commit — must target commit node specifically + commitMsg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(true), + ) + commitResp, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{commitMsg}, + }), + ) + require.NoError(t, err) + + parsed, err := envelopes.NewOriginatorEnvelope(commitResp.Msg.GetOriginatorEnvelopes()[0]) + require.NoError(t, err) + require.Equal( + t, commitNodeID, parsed.UnsignedOriginatorEnvelope.PayerEnvelope.TargetOriginator, + ) + + t.Logf("Mixed traffic: 5 regular + 1 commit, all published OK") +} + +// TestNoBlockchain_GatewayE2E_IdentityUpdateFlow tests identity update +// routing through payer to dedicated identity node. +func TestNoBlockchain_GatewayE2E_IdentityUpdateFlow(t *testing.T) { + // Use node 100 as identity node (test server is always 100) + commitNodeID := uint32(200) + identityNodeID := uint32(100) + + identitySuite := apiTestUtils.NewTestAPIServer(t, + apiTestUtils.WithTestNoBlockchain(), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + svc, mockBlockchain, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(commitNodeID, identityNodeID), + ) + + mockRegistry.EXPECT().GetNode(identityNodeID).Return(®istry.Node{ + NodeID: identityNodeID, + HTTPAddress: formatAddress(identitySuite.APIServer.Addr()), + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(identityNodeID), + }, nil).Maybe() + + inboxID := testutils.RandomInboxIDBytes() + identityUpdate := &associations.IdentityUpdate{ + InboxId: utils.HexEncode(inboxID[:]), + } + envelope := envelopesTestUtils.CreateIdentityUpdateClientEnvelope(inboxID, identityUpdate) + + resp, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{envelope}, + }), + ) + require.NoError(t, err) + require.Len(t, resp.Msg.GetOriginatorEnvelopes(), 1) + + parsed, err := envelopes.NewOriginatorEnvelope(resp.Msg.GetOriginatorEnvelopes()[0]) + require.NoError(t, err) + require.Equal( + t, identityNodeID, parsed.UnsignedOriginatorEnvelope.PayerEnvelope.TargetOriginator, + ) + + mockBlockchain.AssertNotCalled( + t, "PublishIdentityUpdate", mock.Anything, mock.Anything, mock.Anything, + ) + t.Logf("E2E identity update flow: publish + verify on node %d OK", identityNodeID) +} + +// TestNoBlockchain_GatewayE2E_MultiGroupOrdering tests that commits for +// different groups all route to the same commit node, preserving ordering. +func TestNoBlockchain_GatewayE2E_MultiGroupOrdering(t *testing.T) { + commitNodeID := uint32(100) + + commitSuite := apiTestUtils.NewTestAPIServer(t, + apiTestUtils.WithTestNoBlockchain(), + ) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + svc, _, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(commitNodeID, 200), + ) + + mockRegistry.EXPECT().GetNode(commitNodeID).Return(®istry.Node{ + NodeID: commitNodeID, + HTTPAddress: formatAddress(commitSuite.APIServer.Addr()), + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(commitNodeID), + }, nil).Maybe() + + numGroups := 10 + for i := range numGroups { + groupID := testutils.RandomGroupID() + commitMsg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(true), + ) + + resp, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{commitMsg}, + }), + ) + require.NoError(t, err, "commit for group %d", i) + require.Len(t, resp.Msg.GetOriginatorEnvelopes(), 1) + + parsed, err := envelopes.NewOriginatorEnvelope(resp.Msg.GetOriginatorEnvelopes()[0]) + require.NoError(t, err) + targetOrig := parsed.UnsignedOriginatorEnvelope.PayerEnvelope.TargetOriginator + require.Equal( + t, commitNodeID, targetOrig, + "group %d commit should go to commit node", i, + ) + } + + t.Logf("Multi-group ordering: %d groups, all commits → node %d", numGroups, commitNodeID) +} diff --git a/pkg/api/payer/no_blockchain_bench_test.go b/pkg/api/payer/no_blockchain_bench_test.go new file mode 100644 index 000000000..7cd639e56 --- /dev/null +++ b/pkg/api/payer/no_blockchain_bench_test.go @@ -0,0 +1,114 @@ +package payer_test + +import ( + "context" + "testing" + "time" + + "connectrpc.com/connect" + + "github.com/xmtp/xmtpd/pkg/api/payer" + envelopesProto "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/envelopes" + "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/payer_api" + "github.com/xmtp/xmtpd/pkg/registry" + "github.com/xmtp/xmtpd/pkg/testutils" + apiTestUtils "github.com/xmtp/xmtpd/pkg/testutils/api" + envelopesTestUtils "github.com/xmtp/xmtpd/pkg/testutils/envelopes" + nodeRegistry "github.com/xmtp/xmtpd/pkg/testutils/registry" +) + +func TestNoBlockchain_LatencyComparison(t *testing.T) { + const iterations = 50 + + // --- No-blockchain commit path --- + suiteNoChain := apiTestUtils.NewTestAPIServer(t, apiTestUtils.WithTestNoBlockchain()) + ctx := context.Background() + svcNoChain, _, mockRegNoChain, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + ) + + mockRegNoChain.EXPECT().GetNode(uint32(100)).Return(®istry.Node{ + NodeID: 100, + HTTPAddress: formatAddress(suiteNoChain.APIServer.Addr()), + IsCanonical: true, + }, nil).Maybe() + mockRegNoChain.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + }, nil).Maybe() + + // Warmup + for range 3 { + groupID := testutils.RandomGroupID() + msg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, envelopesTestUtils.GetRealisticGroupMessagePayload(true), + ) + _, _ = svcNoChain.PublishClientEnvelopes(ctx, connect.NewRequest( + &payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{msg}, + })) + } + + start := time.Now() + for range iterations { + groupID := testutils.RandomGroupID() + msg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, envelopesTestUtils.GetRealisticGroupMessagePayload(true), + ) + _, err := svcNoChain.PublishClientEnvelopes(ctx, connect.NewRequest( + &payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{msg}, + })) + if err != nil { + t.Fatalf("no-blockchain commit publish failed: %v", err) + } + } + noChainDuration := time.Since(start) + noChainAvg := noChainDuration / time.Duration(iterations) + + // --- Regular message path (baseline) --- + suiteRegular := apiTestUtils.NewTestAPIServer(t) + svcRegular, _, mockRegRegular, _ := buildPayerService(t) + + mockRegRegular.EXPECT().GetNode(uint32(100)).Return(®istry.Node{ + HTTPAddress: formatAddress(suiteRegular.APIServer.Addr()), + }, nil).Maybe() + mockRegRegular.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + }, nil).Maybe() + + // Warmup + for range 3 { + groupID := testutils.RandomGroupID() + msg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, envelopesTestUtils.GetRealisticGroupMessagePayload(false), + ) + _, _ = svcRegular.PublishClientEnvelopes(ctx, connect.NewRequest( + &payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{msg}, + })) + } + + start = time.Now() + for range iterations { + groupID := testutils.RandomGroupID() + msg := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, envelopesTestUtils.GetRealisticGroupMessagePayload(false), + ) + _, err := svcRegular.PublishClientEnvelopes(ctx, connect.NewRequest( + &payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{msg}, + })) + if err != nil { + t.Fatalf("regular publish failed: %v", err) + } + } + regularDuration := time.Since(start) + regularAvg := regularDuration / time.Duration(iterations) + + t.Logf("=== LATENCY COMPARISON (%d iterations) ===", iterations) + t.Logf("No-blockchain commit path: %v total, %v avg per publish", noChainDuration, noChainAvg) + t.Logf("Regular message path: %v total, %v avg per publish", regularDuration, regularAvg) + t.Logf("Ratio (nochain/regular): %.2fx", float64(noChainAvg)/float64(regularAvg)) + t.Logf("Note: blockchain path would add ~2-15s per commit (Arbitrum L2 finality)") +} diff --git a/pkg/api/payer/no_blockchain_test.go b/pkg/api/payer/no_blockchain_test.go new file mode 100644 index 000000000..2fe4f747a --- /dev/null +++ b/pkg/api/payer/no_blockchain_test.go @@ -0,0 +1,163 @@ +package payer_test + +import ( + "context" + "testing" + + "connectrpc.com/connect" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/xmtp/xmtpd/pkg/api/payer" + "github.com/xmtp/xmtpd/pkg/constants" + "github.com/xmtp/xmtpd/pkg/envelopes" + "github.com/xmtp/xmtpd/pkg/proto/identity/associations" + envelopesProto "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/envelopes" + "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/payer_api" + "github.com/xmtp/xmtpd/pkg/registry" + "github.com/xmtp/xmtpd/pkg/testutils" + apiTestUtils "github.com/xmtp/xmtpd/pkg/testutils/api" + envelopesTestUtils "github.com/xmtp/xmtpd/pkg/testutils/envelopes" + nodeRegistry "github.com/xmtp/xmtpd/pkg/testutils/registry" + "github.com/xmtp/xmtpd/pkg/utils" +) + +func TestNoBlockchain_IdentityUpdateRoutedToNode(t *testing.T) { + ctx := context.Background() + svc, mockBlockchain, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + ) + + mockRegistry.EXPECT().GetNode(uint32(200)).Return(®istry.Node{ + NodeID: 200, + HTTPAddress: "http://localhost:5051", + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(200), + }, nil).Maybe() + + inboxID := testutils.RandomInboxIDBytes() + identityUpdate := &associations.IdentityUpdate{ + InboxId: utils.HexEncode(inboxID[:]), + } + envelope := envelopesTestUtils.CreateIdentityUpdateClientEnvelope(inboxID, identityUpdate) + + _, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{envelope}, + }), + ) + // Will fail connecting to node 200, but NOT with blockchain error + if err != nil { + require.NotContains(t, err.Error(), "blockchain", + "identity update should route to node, not blockchain") + t.Logf("Expected error (no node to connect to): %v", err) + } + + mockBlockchain.AssertNotCalled( + t, "PublishIdentityUpdate", mock.Anything, mock.Anything, mock.Anything, + ) +} + +func TestNoBlockchain_CommitRoutedToNode(t *testing.T) { + suite := apiTestUtils.NewTestAPIServer(t, apiTestUtils.WithTestNoBlockchain()) + + ctx := context.Background() + svc, mockBlockchain, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + ) + + // Node 100 is the commit handler — point it at our test server + mockRegistry.EXPECT().GetNode(uint32(100)).Return(®istry.Node{ + NodeID: 100, + HTTPAddress: formatAddress(suite.APIServer.Addr()), + IsCanonical: true, + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + }, nil).Maybe() + + // Create a commit message (shouldSendToBlockchain=true) + groupID := testutils.RandomGroupID() + commitMessage := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(true), // isCommit=true + ) + + publishResponse, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{commitMessage}, + }), + ) + + require.NoError(t, err, "commit should publish to node successfully in no-blockchain mode") + require.NotNil(t, publishResponse) + require.Len(t, publishResponse.Msg.GetOriginatorEnvelopes(), 1) + + // Verify it was sent to node 100 (commit handler), not blockchain + responseEnvelope := publishResponse.Msg.GetOriginatorEnvelopes()[0] + parsedOriginatorEnvelope, err := envelopes.NewOriginatorEnvelope(responseEnvelope) + require.NoError(t, err) + + targetOriginator := parsedOriginatorEnvelope. + UnsignedOriginatorEnvelope.PayerEnvelope.TargetOriginator + require.Equal( + t, uint32(100), targetOriginator, + "commit should target node 100 (commit handler)", + ) + + // Verify retention policy is set + require.EqualValues( + t, + constants.DefaultStorageDurationDays, + parsedOriginatorEnvelope.UnsignedOriginatorEnvelope.PayerEnvelope.RetentionDays(), + ) + + // Blockchain publisher must NOT have been called + mockBlockchain.AssertNotCalled( + t, "PublishGroupMessage", mock.Anything, mock.Anything, mock.Anything, + ) +} + +func TestNoBlockchain_RegularMessageStillUsesNodeSelector(t *testing.T) { + suite := apiTestUtils.NewTestAPIServer(t) + + ctx := context.Background() + svc, _, mockRegistry, _ := buildPayerService( + t, + payer.WithNoBlockchain(100, 200), + ) + + mockRegistry.EXPECT().GetNode(mock.Anything).Return(®istry.Node{ + HTTPAddress: formatAddress(suite.APIServer.Addr()), + }, nil).Maybe() + + mockRegistry.On("GetNodes").Return([]registry.Node{ + nodeRegistry.GetHealthyNode(100), + }, nil).Maybe() + + // Regular (non-commit) group message + groupID := testutils.RandomGroupID() + regularMessage := envelopesTestUtils.CreateGroupMessageClientEnvelope( + groupID, + envelopesTestUtils.GetRealisticGroupMessagePayload(false), + ) + + publishResponse, err := svc.PublishClientEnvelopes( + ctx, + connect.NewRequest(&payer_api.PublishClientEnvelopesRequest{ + Envelopes: []*envelopesProto.ClientEnvelope{regularMessage}, + }), + ) + + require.NoError(t, err, "regular message should publish normally in no-blockchain mode") + require.NotNil(t, publishResponse) + require.Len(t, publishResponse.Msg.GetOriginatorEnvelopes(), 1) +} diff --git a/pkg/api/payer/service.go b/pkg/api/payer/service.go index b87e9b179..5e17b54b4 100644 --- a/pkg/api/payer/service.go +++ b/pkg/api/payer/service.go @@ -263,7 +263,20 @@ func (s *Service) groupEnvelopes( ) } - if toBlockchain { + if toBlockchain && s.cfg.NoBlockchain { + // No-blockchain mode: route to dedicated nodes instead of chain + var targetNodeID uint32 + switch clientEnvelope.TargetTopic().Kind() { + case topic.TopicKindIdentityUpdatesV1: + targetNodeID = s.cfg.IdentityNodeID + case topic.TopicKindGroupMessagesV1: + targetNodeID = s.cfg.CommitNodeID + } + out.forNodes[targetNodeID] = append( + out.forNodes[targetNodeID], + newClientEnvelopeWithIndex(i, clientEnvelope), + ) + } else if toBlockchain { out.forBlockchain = append( out.forBlockchain, newClientEnvelopeWithIndex(i, clientEnvelope), @@ -616,17 +629,16 @@ func determineRetentionPolicy(clientEnvelope *envelopes.ClientEnvelope) (uint32, switch clientEnvelope.TargetTopic().Kind() { case topic.TopicKindIdentityUpdatesV1: - panic("should not be called for identity updates") + // In no-blockchain mode, identity updates are routed to nodes + return constants.DefaultStorageDurationDays, nil case topic.TopicKindGroupMessagesV1: switch payload := clientEnvelope.Payload().(type) { case *envelopesProto.ClientEnvelope_GroupMessage: - shouldSendToBlockchain, err := deserializer.ShouldSendToBlockchain(payload) + // In no-blockchain mode, commits/proposals are routed to nodes + _, err := deserializer.ShouldSendToBlockchain(payload) if err != nil { return 0, err } - if shouldSendToBlockchain { - panic("should not be called for group message commits or proposals") - } default: panic("mismatched payload type") } diff --git a/pkg/config/options.go b/pkg/config/options.go index e1f18ffa4..c3d9b74fd 100644 --- a/pkg/config/options.go +++ b/pkg/config/options.go @@ -14,6 +14,9 @@ type APIOptions struct { OriginatorCacheTTL time.Duration `long:"originator-cache-ttl" env:"XMTPD_API_ORIGINATOR_CACHE_TTL" description:"TTL for originator list cache" default:"5m"` RequirePayerPositiveBalance bool `long:"require-payer-positive-balance" env:"XMTPD_API_REQUIRE_PAYER_POSITIVE_BALANCE" description:"Reject publishes when payer balance is insufficient"` RequireReplicationNodeAuth bool `long:"require-replication-node-auth" env:"XMTPD_API_REQUIRE_REPLICATION_NODE_AUTH" description:"Require node JWT authentication for all ReplicationApi methods"` + + // No-blockchain experiment: accept commits and identity updates directly from nodes + NoBlockchain bool `long:"no-blockchain" env:"XMTPD_API_NO_BLOCKCHAIN" description:"Accept commits and identity updates from nodes instead of requiring blockchain"` } type ContractsOptions struct { @@ -95,6 +98,11 @@ type PayerOptions struct { NodeSelectorTimeout time.Duration `long:"node-selector-connect-timeout" env:"XMTPD_PAYER_NODE_SELECTOR_CONNECT_TIMEOUT" description:"Connection timeout" default:"2s"` EnvelopePublishTimeout time.Duration `long:"envelope-publish-timeout" env:"XMTPD_PAYER_ENVELOPE_PUBLISH_TIMEOUT" description:"Envelope publish timeout" default:"30s"` EnvelopePublishRetries uint `long:"envelope-publish-retries" env:"XMTPD_PAYER_ENVELOPE_PUBLISH_RETRIES" description:"How many times to retry publishing an envelope" default:"5"` + + // No-blockchain experiment: route commits and identity updates to dedicated nodes instead of the chain + NoBlockchain bool `long:"no-blockchain" env:"XMTPD_PAYER_NO_BLOCKCHAIN" description:"Skip blockchain for commits and identity updates, route to dedicated nodes instead"` + CommitNodeID uint32 `long:"commit-node-id" env:"XMTPD_PAYER_COMMIT_NODE_ID" description:"Node ID for dedicated commit processing (used with --no-blockchain)"` + IdentityNodeID uint32 `long:"identity-node-id" env:"XMTPD_PAYER_IDENTITY_NODE_ID" description:"Node ID for dedicated identity update processing (used with --no-blockchain)"` } type ReplicationOptions struct { diff --git a/pkg/gateway/builder.go b/pkg/gateway/builder.go index 46cb6d39d..7c4580cde 100644 --- a/pkg/gateway/builder.go +++ b/pkg/gateway/builder.go @@ -264,6 +264,22 @@ func (b *GatewayServiceBuilder) buildGatewayService( return nil, errors.Wrap(err, "failed to create node selector") } + payerOpts := []payer.Option{ + payer.WithPublishTimeout(b.config.Payer.EnvelopePublishTimeout), + payer.WithPublishRetries(b.config.Payer.EnvelopePublishRetries), + } + if b.config.Payer.NoBlockchain { + payerOpts = append(payerOpts, payer.WithNoBlockchain( + b.config.Payer.CommitNodeID, + b.config.Payer.IdentityNodeID, + )) + b.logger.Info( + "no-blockchain mode enabled", + zap.Uint32("commit_node_id", b.config.Payer.CommitNodeID), + zap.Uint32("identity_node_id", b.config.Payer.IdentityNodeID), + ) + } + gatewayAPIService, err := payer.NewPayerAPIService( ctx, b.logger, @@ -273,8 +289,7 @@ func (b *GatewayServiceBuilder) buildGatewayService( clientMetrics, b.config.Contracts.AppChain.MaxBlockchainPayloadSize, nodeSelector, - payer.WithPublishTimeout(b.config.Payer.EnvelopePublishTimeout), - payer.WithPublishRetries(b.config.Payer.EnvelopePublishRetries), + payerOpts..., ) if err != nil { return nil, err diff --git a/pkg/testutils/api/api.go b/pkg/testutils/api/api.go index 7b6a4ebdf..fef7c98cc 100644 --- a/pkg/testutils/api/api.go +++ b/pkg/testutils/api/api.go @@ -212,6 +212,7 @@ type APIServerTestSuite struct { type APIServerTestConfig struct { registryNodes []registry.Node requirePayerPositiveBalance bool + noBlockchain bool } type TestAPIOption func(*APIServerTestConfig) @@ -228,6 +229,12 @@ func WithRequirePayerPositiveBalance(enabled bool) TestAPIOption { } } +func WithTestNoBlockchain() TestAPIOption { + return func(cfg *APIServerTestConfig) { + cfg.noBlockchain = true + } +} + func createMockRegistry(t *testing.T, nodes []registry.Node) *registryMocks.MockNodeRegistry { reg := registryMocks.NewMockNodeRegistry(t) @@ -309,6 +316,7 @@ func NewTestAPIServer( config.APIOptions{ SendKeepAliveInterval: 30 * time.Second, RequirePayerPositiveBalance: cfg.requirePayerPositiveBalance, + NoBlockchain: cfg.noBlockchain, }, false, 10*time.Millisecond,