| 1 | package cli |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "slices" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/ipfs/kubo/test/cli/harness" |
| 11 | "github.com/stretchr/testify/assert" |
| 12 | "github.com/stretchr/testify/require" |
| 13 | ) |
| 14 | |
| 15 | // waitForSubscription waits until the node has a subscription to the given topic. |
| 16 | func waitForSubscription(t *testing.T, node *harness.Node, topic string) { |
| 17 | t.Helper() |
| 18 | require.Eventually(t, func() bool { |
| 19 | res := node.RunIPFS("pubsub", "ls") |
| 20 | if res.Err != nil { |
| 21 | return false |
| 22 | } |
| 23 | return slices.Contains(res.Stdout.Lines(), topic) |
| 24 | }, 5*time.Second, 100*time.Millisecond, "expected subscription to topic %s", topic) |
| 25 | } |
| 26 | |
| 27 | // waitForMessagePropagation waits for pubsub messages to propagate through the network |
| 28 | // and for seqno state to be persisted to the datastore. |
| 29 | func waitForMessagePropagation(t *testing.T) { |
| 30 | t.Helper() |
| 31 | time.Sleep(1 * time.Second) |
| 32 | } |
| 33 | |
| 34 | // publishMessages publishes n messages from publisher to the given topic with |
| 35 | // a small delay between each to allow for ordered delivery. |
| 36 | func publishMessages(t *testing.T, publisher *harness.Node, topic string, n int) { |
| 37 | t.Helper() |
| 38 | for range n { |
| 39 | publisher.PipeStrToIPFS("msg", "pubsub", "pub", topic) |
| 40 | time.Sleep(50 * time.Millisecond) |
| 41 | } |
| 42 | } |
| 43 | |
| 44 | // TestPubsub tests pubsub functionality and the persistent seqno validator. |
| 45 | // |
| 46 | // Pubsub has two deduplication layers: |
| 47 | // |
| 48 | // Layer 1: MessageID-based TimeCache (in-memory) |
| 49 | // - Controlled by Pubsub.SeenMessagesTTL config (default 120s) |
| 50 | // - Tested in go-libp2p-pubsub (see timecache in github.com/libp2p/go-libp2p-pubsub) |
| 51 | // - Only tested implicitly here via message delivery (timing-sensitive, not practical for CLI tests) |
| 52 | // |
| 53 | // Layer 2: Per-peer seqno validator (persistent in datastore) |
| 54 | // - Stores max seen seqno per peer at /pubsub/seqno/<peerid> |
| 55 | // - Tested directly below: persistence, updates, reset, survives restart |
| 56 | // - Validator: go-libp2p-pubsub BasicSeqnoValidator |
| 57 | func TestPubsub(t *testing.T) { |
| 58 | t.Parallel() |
| 59 | |
| 60 | // enablePubsub configures a node with pubsub enabled |
| 61 | enablePubsub := func(n *harness.Node) { |
| 62 | n.SetIPFSConfig("Pubsub.Enabled", true) |
| 63 | n.SetIPFSConfig("Routing.Type", "none") // simplify test setup |
| 64 | } |
| 65 | |
| 66 | t.Run("basic pub/sub message delivery", func(t *testing.T) { |
| 67 | t.Parallel() |
| 68 | h := harness.NewT(t) |
| 69 | |
| 70 | // Create two connected nodes with pubsub enabled |
| 71 | nodes := h.NewNodes(2).Init() |
| 72 | nodes.ForEachPar(enablePubsub) |
| 73 | nodes = nodes.StartDaemons().Connect() |
| 74 | defer nodes.StopDaemons() |
| 75 | |
| 76 | subscriber := nodes[0] |
| 77 | publisher := nodes[1] |
| 78 | |
| 79 | const topic = "test-topic" |
| 80 | const message = "hello pubsub" |
| 81 | |
| 82 | // Start subscriber in background |
| 83 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 84 | defer cancel() |
| 85 | |
| 86 | // Use a channel to receive the message |
| 87 | msgChan := make(chan string, 1) |
| 88 | go func() { |
| 89 | // Subscribe and wait for one message |
| 90 | res := subscriber.RunIPFS("pubsub", "sub", "--enc=json", topic) |
| 91 | if res.Err == nil { |
| 92 | // Parse JSON output to get message data |
| 93 | lines := res.Stdout.Lines() |
| 94 | if len(lines) > 0 { |
| 95 | var msg struct { |
| 96 | Data []byte `json:"data"` |
| 97 | } |
| 98 | if json.Unmarshal([]byte(lines[0]), &msg) == nil { |
| 99 | msgChan <- string(msg.Data) |
| 100 | } |
| 101 | } |
| 102 | } |
| 103 | }() |
| 104 | |
| 105 | // Wait for subscriber to be ready |
| 106 | waitForSubscription(t, subscriber, topic) |
| 107 | |
| 108 | // Publish message |
| 109 | publisher.PipeStrToIPFS(message, "pubsub", "pub", topic) |
| 110 | |
| 111 | // Wait for message or timeout |
| 112 | select { |
| 113 | case received := <-msgChan: |
| 114 | assert.Equal(t, message, received) |
| 115 | case <-ctx.Done(): |
| 116 | // Subscriber may not receive in time due to test timing - that's OK |
| 117 | // The main goal is to test the seqno validator state persistence |
| 118 | t.Log("subscriber did not receive message in time (this is acceptable)") |
| 119 | } |
| 120 | }) |
| 121 | |
| 122 | t.Run("seqno validator state is persisted", func(t *testing.T) { |
| 123 | t.Parallel() |
| 124 | h := harness.NewT(t) |
| 125 | |
| 126 | // Create two connected nodes with pubsub |
| 127 | nodes := h.NewNodes(2).Init() |
| 128 | nodes.ForEachPar(enablePubsub) |
| 129 | nodes = nodes.StartDaemons().Connect() |
| 130 | |
| 131 | node1 := nodes[0] |
| 132 | node2 := nodes[1] |
| 133 | node2PeerID := node2.PeerID().String() |
| 134 | |
| 135 | const topic = "seqno-test" |
| 136 | |
| 137 | // Start subscriber on node1 |
| 138 | go func() { |
| 139 | node1.RunIPFS("pubsub", "sub", topic) |
| 140 | }() |
| 141 | waitForSubscription(t, node1, topic) |
| 142 | |
| 143 | // Publish multiple messages from node2 to trigger seqno validation |
| 144 | publishMessages(t, node2, topic, 3) |
| 145 | |
| 146 | // Wait for messages to propagate and seqno to be stored |
| 147 | waitForMessagePropagation(t) |
| 148 | |
| 149 | // Stop daemons to check datastore (diag datastore requires daemon to be stopped) |
| 150 | nodes.StopDaemons() |
| 151 | |
| 152 | // Check that seqno state exists |
| 153 | count := node1.DatastoreCount("/pubsub/seqno/") |
| 154 | t.Logf("seqno entries count: %d", count) |
| 155 | |
| 156 | // There should be at least one seqno entry (from node2) |
| 157 | assert.NotEqual(t, int64(0), count, "expected seqno state to be persisted") |
| 158 | |
| 159 | // Verify the specific peer's key exists and test --hex output format |
| 160 | key := "/pubsub/seqno/" + node2PeerID |
| 161 | res := node1.RunIPFS("diag", "datastore", "get", "--hex", key) |
| 162 | if res.Err == nil { |
| 163 | t.Logf("seqno for peer %s:\n%s", node2PeerID, res.Stdout.String()) |
| 164 | assert.Contains(t, res.Stdout.String(), "Hex Dump:") |
| 165 | } else { |
| 166 | // Key might not exist if messages didn't propagate - log but don't fail |
| 167 | t.Logf("seqno key not found for peer %s (messages may not have propagated)", node2PeerID) |
| 168 | } |
| 169 | }) |
| 170 | |
| 171 | t.Run("seqno updates when receiving multiple messages", func(t *testing.T) { |
| 172 | t.Parallel() |
| 173 | h := harness.NewT(t) |
| 174 | |
| 175 | // Create two connected nodes with pubsub |
| 176 | nodes := h.NewNodes(2).Init() |
| 177 | nodes.ForEachPar(enablePubsub) |
| 178 | nodes = nodes.StartDaemons().Connect() |
| 179 | |
| 180 | node1 := nodes[0] |
| 181 | node2 := nodes[1] |
| 182 | node2PeerID := node2.PeerID().String() |
| 183 | |
| 184 | const topic = "seqno-update-test" |
| 185 | seqnoKey := "/pubsub/seqno/" + node2PeerID |
| 186 | |
| 187 | // Start subscriber on node1 |
| 188 | go func() { |
| 189 | node1.RunIPFS("pubsub", "sub", topic) |
| 190 | }() |
| 191 | waitForSubscription(t, node1, topic) |
| 192 | |
| 193 | // Send first message |
| 194 | node2.PipeStrToIPFS("msg1", "pubsub", "pub", topic) |
| 195 | time.Sleep(500 * time.Millisecond) |
| 196 | |
| 197 | // Stop daemons to check seqno (diag datastore requires daemon to be stopped) |
| 198 | nodes.StopDaemons() |
| 199 | |
| 200 | // Get seqno after first message |
| 201 | res1 := node1.RunIPFS("diag", "datastore", "get", seqnoKey) |
| 202 | var seqno1 []byte |
| 203 | if res1.Err == nil { |
| 204 | seqno1 = res1.Stdout.Bytes() |
| 205 | t.Logf("seqno after first message: %d bytes", len(seqno1)) |
| 206 | } else { |
| 207 | t.Logf("seqno not found after first message (message may not have propagated)") |
| 208 | } |
| 209 | |
| 210 | // Restart daemons for second message |
| 211 | nodes = nodes.StartDaemons().Connect() |
| 212 | |
| 213 | // Resubscribe |
| 214 | go func() { |
| 215 | node1.RunIPFS("pubsub", "sub", topic) |
| 216 | }() |
| 217 | waitForSubscription(t, node1, topic) |
| 218 | |
| 219 | // Send second message |
| 220 | node2.PipeStrToIPFS("msg2", "pubsub", "pub", topic) |
| 221 | time.Sleep(500 * time.Millisecond) |
| 222 | |
| 223 | // Stop daemons to check seqno |
| 224 | nodes.StopDaemons() |
| 225 | |
| 226 | // Get seqno after second message |
| 227 | res2 := node1.RunIPFS("diag", "datastore", "get", seqnoKey) |
| 228 | var seqno2 []byte |
| 229 | if res2.Err == nil { |
| 230 | seqno2 = res2.Stdout.Bytes() |
| 231 | t.Logf("seqno after second message: %d bytes", len(seqno2)) |
| 232 | } else { |
| 233 | t.Logf("seqno not found after second message") |
| 234 | } |
| 235 | |
| 236 | // If both messages were received, seqno should have been updated |
| 237 | // The seqno is a uint64 that should increase with each message |
| 238 | if len(seqno1) > 0 && len(seqno2) > 0 { |
| 239 | // seqno2 should be >= seqno1 (it's the max seen seqno) |
| 240 | // We just verify they're both non-empty and potentially different |
| 241 | t.Logf("seqno1: %x", seqno1) |
| 242 | t.Logf("seqno2: %x", seqno2) |
| 243 | // The seqno validator stores the max seqno seen, so seqno2 >= seqno1 |
| 244 | // We can't do a simple byte comparison due to potential endianness |
| 245 | // but both should be valid uint64 values (8 bytes) |
| 246 | assert.Equal(t, 8, len(seqno2), "seqno should be 8 bytes (uint64)") |
| 247 | } |
| 248 | }) |
| 249 | |
| 250 | t.Run("pubsub reset clears seqno state", func(t *testing.T) { |
| 251 | t.Parallel() |
| 252 | h := harness.NewT(t) |
| 253 | |
| 254 | // Create two connected nodes |
| 255 | nodes := h.NewNodes(2).Init() |
| 256 | nodes.ForEachPar(enablePubsub) |
| 257 | nodes = nodes.StartDaemons().Connect() |
| 258 | |
| 259 | node1 := nodes[0] |
| 260 | node2 := nodes[1] |
| 261 | |
| 262 | const topic = "reset-test" |
| 263 | |
| 264 | // Start subscriber and exchange messages |
| 265 | go func() { |
| 266 | node1.RunIPFS("pubsub", "sub", topic) |
| 267 | }() |
| 268 | waitForSubscription(t, node1, topic) |
| 269 | |
| 270 | publishMessages(t, node2, topic, 3) |
| 271 | waitForMessagePropagation(t) |
| 272 | |
| 273 | // Stop daemons to check initial count |
| 274 | nodes.StopDaemons() |
| 275 | |
| 276 | // Verify there is state before resetting |
| 277 | initialCount := node1.DatastoreCount("/pubsub/seqno/") |
| 278 | t.Logf("initial seqno count: %d", initialCount) |
| 279 | |
| 280 | // Restart node1 to run pubsub reset |
| 281 | node1.StartDaemon() |
| 282 | |
| 283 | // Reset all seqno state (while daemon is running) |
| 284 | res := node1.IPFS("pubsub", "reset") |
| 285 | assert.NoError(t, res.Err) |
| 286 | t.Logf("reset output: %s", res.Stdout.String()) |
| 287 | |
| 288 | // Stop daemon to verify state was cleared |
| 289 | node1.StopDaemon() |
| 290 | |
| 291 | // Verify state was cleared |
| 292 | finalCount := node1.DatastoreCount("/pubsub/seqno/") |
| 293 | t.Logf("final seqno count: %d", finalCount) |
| 294 | assert.Equal(t, int64(0), finalCount, "seqno state should be cleared after reset") |
| 295 | }) |
| 296 | |
| 297 | t.Run("pubsub reset with peer flag", func(t *testing.T) { |
| 298 | t.Parallel() |
| 299 | h := harness.NewT(t) |
| 300 | |
| 301 | // Create three connected nodes |
| 302 | nodes := h.NewNodes(3).Init() |
| 303 | nodes.ForEachPar(enablePubsub) |
| 304 | nodes = nodes.StartDaemons().Connect() |
| 305 | |
| 306 | node1 := nodes[0] |
| 307 | node2 := nodes[1] |
| 308 | node3 := nodes[2] |
| 309 | node2PeerID := node2.PeerID().String() |
| 310 | node3PeerID := node3.PeerID().String() |
| 311 | |
| 312 | const topic = "peer-reset-test" |
| 313 | |
| 314 | // Start subscriber on node1 |
| 315 | go func() { |
| 316 | node1.RunIPFS("pubsub", "sub", topic) |
| 317 | }() |
| 318 | waitForSubscription(t, node1, topic) |
| 319 | |
| 320 | // Publish from both node2 and node3 |
| 321 | for range 3 { |
| 322 | node2.PipeStrToIPFS("msg2", "pubsub", "pub", topic) |
| 323 | node3.PipeStrToIPFS("msg3", "pubsub", "pub", topic) |
| 324 | time.Sleep(50 * time.Millisecond) |
| 325 | } |
| 326 | waitForMessagePropagation(t) |
| 327 | |
| 328 | // Stop node2 and node3 |
| 329 | node2.StopDaemon() |
| 330 | node3.StopDaemon() |
| 331 | |
| 332 | // Reset only node2's state (while node1 daemon is running) |
| 333 | res := node1.IPFS("pubsub", "reset", "--peer", node2PeerID) |
| 334 | require.NoError(t, res.Err) |
| 335 | t.Logf("reset output: %s", res.Stdout.String()) |
| 336 | |
| 337 | // Stop node1 daemon to check datastore |
| 338 | node1.StopDaemon() |
| 339 | |
| 340 | // Check that node2's key is gone |
| 341 | res = node1.RunIPFS("diag", "datastore", "get", "/pubsub/seqno/"+node2PeerID) |
| 342 | assert.Error(t, res.Err, "node2's seqno key should be deleted") |
| 343 | |
| 344 | // Check that node3's key still exists (if it was created) |
| 345 | res = node1.RunIPFS("diag", "datastore", "get", "/pubsub/seqno/"+node3PeerID) |
| 346 | // Note: node3's key might not exist if messages didn't propagate |
| 347 | // So we just log the result without asserting |
| 348 | if res.Err == nil { |
| 349 | t.Logf("node3's seqno key still exists (as expected)") |
| 350 | } else { |
| 351 | t.Logf("node3's seqno key not found (messages may not have propagated)") |
| 352 | } |
| 353 | }) |
| 354 | |
| 355 | t.Run("seqno state survives daemon restart", func(t *testing.T) { |
| 356 | t.Parallel() |
| 357 | h := harness.NewT(t) |
| 358 | |
| 359 | // Create and start single node |
| 360 | node := h.NewNode().Init() |
| 361 | enablePubsub(node) |
| 362 | node.StartDaemon() |
| 363 | |
| 364 | // We need another node to publish messages |
| 365 | node2 := h.NewNode().Init() |
| 366 | enablePubsub(node2) |
| 367 | node2.StartDaemon() |
| 368 | node.Connect(node2) |
| 369 | |
| 370 | const topic = "restart-test" |
| 371 | |
| 372 | // Start subscriber and exchange messages |
| 373 | go func() { |
| 374 | node.RunIPFS("pubsub", "sub", topic) |
| 375 | }() |
| 376 | waitForSubscription(t, node, topic) |
| 377 | |
| 378 | publishMessages(t, node2, topic, 3) |
| 379 | waitForMessagePropagation(t) |
| 380 | |
| 381 | // Stop daemons to check datastore |
| 382 | node.StopDaemon() |
| 383 | node2.StopDaemon() |
| 384 | |
| 385 | // Get count before restart |
| 386 | beforeCount := node.DatastoreCount("/pubsub/seqno/") |
| 387 | t.Logf("seqno count before restart: %d", beforeCount) |
| 388 | |
| 389 | // Restart node (simulate restart scenario) |
| 390 | node.StartDaemon() |
| 391 | time.Sleep(500 * time.Millisecond) |
| 392 | |
| 393 | // Stop daemon to check datastore again |
| 394 | node.StopDaemon() |
| 395 | |
| 396 | // Get count after restart |
| 397 | afterCount := node.DatastoreCount("/pubsub/seqno/") |
| 398 | t.Logf("seqno count after restart: %d", afterCount) |
| 399 | |
| 400 | // Count should be the same (state persisted) |
| 401 | assert.Equal(t, beforeCount, afterCount, "seqno state should survive daemon restart") |
| 402 | }) |
| 403 | } |