8
"net/http/httptest"
9
"os"
10
"path/filepath"
11
+ "regexp"
12
+ "strconv"
13
"strings"
14
"sync/atomic"
15
"testing"
23
24
const (
25
timeStep = 20 * time.Millisecond
24
- timeout = time.Second
26
+ timeout = 30 * time.Second
27
)
28
29
type cfgApplier func(*harness.Node)
30
29
-func runProviderSuite(t *testing.T, reprovide bool, apply cfgApplier) {
31
+// uniq appends a nanosecond timestamp to s, ensuring unique CIDs
32
+// across test runs and parallel subtests.
33
+func uniq(s string) string {
34
+ return s + " " + strconv.FormatInt(time.Now().UnixNano(), 10)
35
+}
36
+
37
+// awaitReprovideFunc waits until at least minCIDs have been provided
38
+// and returns the total number of CIDs provided so far. The returned
39
+// count can be passed as minCIDs to a subsequent call to wait for the
40
+// next reprovide cycle.
41
+type awaitReprovideFunc func(t *testing.T, n *harness.Node, minCIDs int64) int64
42
+
43
+func runProviderSuite(t *testing.T, sweep bool, apply cfgApplier, awaitReprovide awaitReprovideFunc) {
44
t.Helper()
45
46
initNodes := func(t *testing.T, n int, fn func(n *harness.Node)) harness.Nodes {
33
- nodes := harness.NewT(t).NewNodes(n).Init()
47
+ h := harness.NewT(t)
48
+ nodes := h.NewNodes(n).Init()
49
nodes.ForEachPar(apply)
50
nodes.ForEachPar(fn)
51
+ h.BootstrapWithStubDHT(nodes)
52
nodes = nodes.StartDaemons().Connect()
53
time.Sleep(500 * time.Millisecond) // wait for DHT clients to be bootstrapped
54
return nodes
55
}
56
57
initNodesWithoutStart := func(t *testing.T, n int, fn func(n *harness.Node)) harness.Nodes {
42
- nodes := harness.NewT(t).NewNodes(n).Init()
58
+ h := harness.NewT(t)
59
+ nodes := h.NewNodes(n).Init()
60
nodes.ForEachPar(apply)
61
nodes.ForEachPar(fn)
62
+ h.BootstrapWithStubDHT(nodes)
63
return nodes
64
}
65
248
expectNoProviders(t, cid, nodes[1:]...)
249
})
250
233
- // It is a lesser evil - forces users to fix their config and have some sort of interval
234
- t.Run("Manual Reprovide trigger does not work when periodic reprovide is disabled", func(t *testing.T) {
235
- t.Parallel()
251
+ // `routing reprovide` is only available with the legacy provider.
252
+ // Sweep provider reprovides automatically on schedule.
253
+ if !sweep {
254
+ t.Run("Manual Reprovide trigger does not work when periodic reprovide is disabled", func(t *testing.T) {
255
+ t.Parallel()
256
237
- nodes := initNodes(t, 2, func(n *harness.Node) {
238
- n.SetIPFSConfig("Provide.DHT.Interval", "0")
239
- })
240
- defer nodes.StopDaemons()
257
+ nodes := initNodes(t, 2, func(n *harness.Node) {
258
+ n.SetIPFSConfig("Provide.DHT.Interval", "0")
259
+ })
260
+ defer nodes.StopDaemons()
261
242
- cid := nodes[0].IPFSAddStr(time.Now().String())
262
+ cid := nodes[0].IPFSAddStr(time.Now().String())
263
244
- expectNoProviders(t, cid, nodes[1:]...)
264
+ expectNoProviders(t, cid, nodes[1:]...)
265
246
- res := nodes[0].RunIPFS("routing", "reprovide")
247
- assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.DHT.Interval is set to '0'")
248
- assert.Equal(t, 1, res.ExitCode())
266
+ res := nodes[0].RunIPFS("routing", "reprovide")
267
+ assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.DHT.Interval is set to '0'")
268
+ assert.Equal(t, 1, res.ExitCode())
269
250
- expectNoProviders(t, cid, nodes[1:]...)
251
- })
270
+ expectNoProviders(t, cid, nodes[1:]...)
271
+ })
272
253
- // It is a lesser evil - forces users to fix their config and have some sort of interval
254
- t.Run("Manual Reprovide trigger does not work when Provide system is disabled", func(t *testing.T) {
255
- t.Parallel()
273
+ t.Run("Manual Reprovide trigger does not work when Provide system is disabled", func(t *testing.T) {
274
+ t.Parallel()
275
257
- nodes := initNodes(t, 2, func(n *harness.Node) {
258
- n.SetIPFSConfig("Provide.Enabled", false)
259
- })
260
- defer nodes.StopDaemons()
276
+ nodes := initNodes(t, 2, func(n *harness.Node) {
277
+ n.SetIPFSConfig("Provide.Enabled", false)
278
+ })
279
+ defer nodes.StopDaemons()
280
262
- cid := nodes[0].IPFSAddStr(time.Now().String())
281
+ cid := nodes[0].IPFSAddStr(time.Now().String())
282
264
- expectNoProviders(t, cid, nodes[1:]...)
283
+ expectNoProviders(t, cid, nodes[1:]...)
284
266
- res := nodes[0].RunIPFS("routing", "reprovide")
267
- assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.Enabled is set to 'false'")
268
- assert.Equal(t, 1, res.ExitCode())
285
+ res := nodes[0].RunIPFS("routing", "reprovide")
286
+ assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.Enabled is set to 'false'")
287
+ assert.Equal(t, 1, res.ExitCode())
288
270
- expectNoProviders(t, cid, nodes[1:]...)
271
- })
289
+ expectNoProviders(t, cid, nodes[1:]...)
290
+ })
291
+ }
292
293
t.Run("Provide with 'all' strategy", func(t *testing.T) {
294
t.Parallel()
297
n.SetIPFSConfig("Provide.Strategy", "all")
298
})
299
defer nodes.StopDaemons()
300
+ publisher := nodes[0]
301
281
- cid := nodes[0].IPFSAddStr("all strategy")
282
- expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
302
+ cid := publisher.IPFSAddStr(uniq("all strategy"))
303
+ expectProviders(t, cid, publisher.PeerID().String(), nodes[1:]...)
304
})
305
306
t.Run("Provide with 'pinned' strategy", func(t *testing.T) {
310
n.SetIPFSConfig("Provide.Strategy", "pinned")
311
})
312
defer nodes.StopDaemons()
313
+ publisher := nodes[0]
314
315
// Add a non-pinned CID (should not be provided)
294
- cid := nodes[0].IPFSAddStr("pinned strategy", "--pin=false")
316
+ cid := publisher.IPFSAddStr(uniq("pinned strategy"), "--pin=false")
317
expectNoProviders(t, cid, nodes[1:]...)
318
319
// Pin the CID (should now be provided)
298
- nodes[0].IPFS("pin", "add", cid)
299
- expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
320
+ publisher.IPFS("pin", "add", cid)
321
+ expectProviders(t, cid, publisher.PeerID().String(), nodes[1:]...)
322
})
323
324
t.Run("Provide with 'pinned+mfs' strategy", func(t *testing.T) {
328
n.SetIPFSConfig("Provide.Strategy", "pinned+mfs")
329
})
330
defer nodes.StopDaemons()
331
+ publisher := nodes[0]
332
310
- // Add a pinned CID (should be provided)
311
- cidPinned := nodes[0].IPFSAddStr("pinned content")
312
- cidUnpinned := nodes[0].IPFSAddStr("unpinned content", "--pin=false")
313
- cidMFS := nodes[0].IPFSAddStr("mfs content", "--pin=false")
314
- nodes[0].IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
333
+ cidPinned := publisher.IPFSAddStr(uniq("pinned content"))
334
+ cidUnpinned := publisher.IPFSAddStr(uniq("unpinned content"), "--pin=false")
335
+ cidMFS := publisher.IPFSAddStr(uniq("mfs content"), "--pin=false")
336
+ publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
337
316
- n0pid := nodes[0].PeerID().String()
317
- expectProviders(t, cidPinned, n0pid, nodes[1:]...)
338
+ expectProviders(t, cidPinned, publisher.PeerID().String(), nodes[1:]...)
339
expectNoProviders(t, cidUnpinned, nodes[1:]...)
319
- expectProviders(t, cidMFS, n0pid, nodes[1:]...)
340
+ expectProviders(t, cidMFS, publisher.PeerID().String(), nodes[1:]...)
341
+ })
342
+
343
+ // addLargeFileInSubdir adds a 2 MiB file inside /subdir/ in MFS and
344
+ // returns the MFS root CID, the file root CID, and a chunk CID.
345
+ // The file is large enough to be split into multiple blocks.
346
+ // The resulting DAG: root-dir/subdir/largefile (2+ chunks).
347
+ addLargeFileInSubdir := func(t *testing.T, publisher *harness.Node) (cidRoot, cidSubdir, cidFile, cidChunk string) {
348
+ t.Helper()
349
+ largeData := random.Bytes(2 * 1024 * 1024) // 2 MiB = 2 chunks at 1 MiB
350
+
351
+ // Add file without pinning, then build directory structure in MFS
352
+ cidFile = publisher.IPFSAdd(bytes.NewReader(largeData), "-Q", "--pin=false")
353
+ publisher.IPFS("files", "mkdir", "-p", "/subdir")
354
+ publisher.IPFS("files", "cp", "/ipfs/"+cidFile, "/subdir/largefile")
355
+
356
+ // Get CIDs for the directory structure
357
+ cidRoot = publisher.IPFS("files", "stat", "--hash", "/").Stdout.Trimmed()
358
+ cidSubdir = publisher.IPFS("files", "stat", "--hash", "/subdir").Stdout.Trimmed()
359
+
360
+ // Get a chunk CID from the file's DAG links
361
+ dagOut := publisher.IPFS("dag", "get", cidFile)
362
+ var dagNode struct {
363
+ Links []struct {
364
+ Hash map[string]string `json:"Hash"`
365
+ } `json:"Links"`
366
+ }
367
+ require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
368
+ require.Greater(t, len(dagNode.Links), 1, "file should have multiple chunks")
369
+ cidChunk = dagNode.Links[0].Hash["/"]
370
+ require.NotEmpty(t, cidChunk)
371
+
372
+ return cidRoot, cidSubdir, cidFile, cidChunk
373
+ }
374
+
375
+ // +unique and +entities tests verify which CIDs end up in the DHT
376
+ // (strategy scope). Bloom filter deduplication correctness and
377
+ // entity type detection are tested in boxo/dag/walker/*_test.go.
378
+
379
+ t.Run("Provide with 'pinned+mfs+unique' strategy", func(t *testing.T) {
380
+ t.Parallel()
381
+
382
+ nodes := initNodes(t, 2, func(n *harness.Node) {
383
+ n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+unique")
384
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
385
+ })
386
+ defer nodes.StopDaemons()
387
+ publisher, peers := nodes[0], nodes[1:]
388
+
389
+ // +unique provides all blocks in pinned DAGs (same scope as
390
+ // pinned+mfs but with bloom filter dedup across pins).
391
+ // Use --fast-provide-dag and --fast-provide-wait on pin add
392
+ // so we can verify which blocks the strategy includes.
393
+ cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
394
+ publisher.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidRoot)
395
+ cidUnpinned := publisher.IPFSAddStr(uniq("unpinned content"), "--pin=false")
396
+
397
+ pid := publisher.PeerID().String()
398
+ // All blocks in the pinned DAG should be provided (including chunks)
399
+ expectProviders(t, cidRoot, pid, peers...)
400
+ expectProviders(t, cidSubdir, pid, peers...)
401
+ expectProviders(t, cidFile, pid, peers...)
402
+ expectProviders(t, cidChunk, pid, peers...)
403
+ expectNoProviders(t, cidUnpinned, peers...)
404
+ })
405
+
406
+ t.Run("Provide with 'pinned+mfs+entities' strategy", func(t *testing.T) {
407
+ t.Parallel()
408
+
409
+ nodes := initNodes(t, 2, func(n *harness.Node) {
410
+ n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+entities")
411
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
412
+ })
413
+ defer nodes.StopDaemons()
414
+ publisher, peers := nodes[0], nodes[1:]
415
+
416
+ // +entities provides only entity roots (files, directories,
417
+ // HAMT shards) and skips internal file chunks.
418
+ // Use --fast-provide-dag and --fast-provide-wait on pin add
419
+ // so we can verify which blocks the strategy skips.
420
+ cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
421
+ publisher.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidRoot)
422
+
423
+ pid := publisher.PeerID().String()
424
+ // Entity roots: directories and file root
425
+ expectProviders(t, cidRoot, pid, peers...)
426
+ expectProviders(t, cidSubdir, pid, peers...)
427
+ expectProviders(t, cidFile, pid, peers...)
428
+ // Internal chunk should NOT be provided (+entities skips chunks)
429
+ expectNoProviders(t, cidChunk, peers...)
430
+ })
431
+
432
+ t.Run("ipfs add --fast-provide-dag honors +entities (no chunk providing)", func(t *testing.T) {
433
+ t.Parallel()
434
+
435
+ nodes := initNodes(t, 2, func(n *harness.Node) {
436
+ n.SetIPFSConfig("Provide.Strategy", "pinned+entities")
437
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
438
+ })
439
+ defer nodes.StopDaemons()
440
+ publisher, peers := nodes[0], nodes[1:]
441
+
442
+ // Regression test for the providingDagService double-providing
443
+ // path. Before the fix, ipfs add --pin --fast-provide-dag wrapped
444
+ // the DAGService with providingDagService, which announced every
445
+ // block as it was written -- including chunks -- regardless of
446
+ // the +entities modifier. The post-add ExecuteFastProvideDAG
447
+ // walk then ran in parallel, so chunks ended up in the DHT
448
+ // despite +entities saying they should be skipped.
449
+ //
450
+ // After the fix, ExecuteFastProvideDAG is the single mechanism
451
+ // for --fast-provide-dag and respects the active strategy.
452
+ largeData := random.Bytes(2 * 1024 * 1024) // 2 MiB = 2 chunks
453
+ cidFile := publisher.IPFSAdd(bytes.NewReader(largeData),
454
+ "--fast-provide-dag", "--fast-provide-wait")
455
+
456
+ // Get a chunk CID from the file's DAG links
457
+ dagOut := publisher.IPFS("dag", "get", cidFile)
458
+ var dagNode struct {
459
+ Links []struct {
460
+ Hash map[string]string `json:"Hash"`
461
+ } `json:"Links"`
462
+ }
463
+ require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
464
+ require.Greater(t, len(dagNode.Links), 1, "file should have multiple chunks")
465
+ cidChunk := dagNode.Links[0].Hash["/"]
466
+ require.NotEmpty(t, cidChunk)
467
+
468
+ pid := publisher.PeerID().String()
469
+ // File root (entity) should be provided
470
+ expectProviders(t, cidFile, pid, peers...)
471
+ // Chunk should NOT be provided (+entities skips chunks)
472
+ expectNoProviders(t, cidChunk, peers...)
473
})
474
475
t.Run("Provide with 'roots' strategy", func(t *testing.T) {
479
n.SetIPFSConfig("Provide.Strategy", "roots")
480
})
481
defer nodes.StopDaemons()
482
+ publisher := nodes[0]
483
330
- // Add a root CID (should be provided)
331
- cidRoot := nodes[0].IPFSAddStr("roots strategy", "-w", "-Q")
332
- // the same without wrapping should give us a child node.
333
- cidChild := nodes[0].IPFSAddStr("root strategy", "--pin=false")
484
+ // Add with -w: the wrapper directory is the recursive pin root,
485
+ // the file inside is a child block of that pin (not a root).
486
+ // Use --only-hash first to learn the child CID without providing.
487
+ data := random.Bytes(1000)
488
+ cidChild := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "--only-hash")
489
+ cidRoot := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "-w")
490
335
- expectProviders(t, cidRoot, nodes[0].PeerID().String(), nodes[1:]...)
491
+ // 'roots' strategy provides only pin roots, not child blocks.
492
+ expectProviders(t, cidRoot, publisher.PeerID().String(), nodes[1:]...)
493
expectNoProviders(t, cidChild, nodes[1:]...)
494
})
495
500
n.SetIPFSConfig("Provide.Strategy", "mfs")
501
})
502
defer nodes.StopDaemons()
503
+ publisher := nodes[0]
504
347
- // Add a file to MFS (should be provided)
348
- data := random.Bytes(1000)
349
- cid := nodes[0].IPFSAdd(bytes.NewReader(data), "-Q")
505
+ // 'mfs' only provides content in MFS. Pinned content outside
506
+ // MFS should NOT be provided (mfs excludes pinned by default).
507
+ cidPinned := publisher.IPFSAddStr(uniq("pinned but not mfs"))
508
+ expectNoProviders(t, cidPinned, nodes[1:]...)
509
351
- // not yet in MFS
352
- expectNoProviders(t, cid, nodes[1:]...)
510
+ // Add to MFS (should be provided)
511
+ data := random.Bytes(1000)
512
+ cidMFS := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "--pin=false")
513
+ publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
514
+ expectProviders(t, cidMFS, publisher.PeerID().String(), nodes[1:]...)
515
354
- nodes[0].IPFS("files", "cp", "/ipfs/"+cid, "/myfile")
355
- expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
516
+ // Pinned CID still not provided (mfs strategy ignores pins)
517
+ expectNoProviders(t, cidPinned, nodes[1:]...)
518
})
519
358
- if reprovide {
520
+ // Reprovide tests: add content offline, start daemon, wait for reprovide.
521
+ //
522
+ // Each test waits for TWO reprovide cycles to confirm the schedule
523
+ // works repeatedly, not just on the initial bootstrap. The second
524
+ // cycle also catches bugs where state isn't persisted across cycles.
525
+ //
526
+ // Legacy: `routing reprovide` blocks until the reprovide cycle finishes,
527
+ // so we call it and check results immediately after.
528
+ //
529
+ // Sweep: no manual trigger exists. Instead, we set
530
+ // Provide.DHT.Interval=30s on the importing node and poll
531
+ // `provide stat` until the cycle completes.
532
+
533
+ // verifyReprovide waits for two reprovide cycles and asserts which
534
+ // CIDs are/aren't findable after each. minCIDs is the expected
535
+ // number of provided CIDs per cycle.
536
+ verifyReprovide := func(
537
+ t *testing.T,
538
+ publisher *harness.Node,
539
+ queriers harness.Nodes,
540
+ minCIDs int64,
541
+ provided []string,
542
+ notProvided []string,
543
+ ) {
544
+ t.Helper()
545
+ pid := publisher.PeerID().String()
546
+ check := func() {
547
+ for _, c := range provided {
548
+ expectProviders(t, c, pid, queriers...)
549
+ }
550
+ for _, c := range notProvided {
551
+ expectNoProviders(t, c, queriers...)
552
+ }
553
+ }
554
+
555
+ after1 := awaitReprovide(t, publisher, minCIDs)
556
+ check()
557
+ // Second cycle: confirms the schedule runs repeatedly.
558
+ awaitReprovide(t, publisher, after1+minCIDs)
559
+ check()
560
+ }
561
+
562
+ {
563
564
t.Run("Reprovides with 'all' strategy when strategy is '' (empty)", func(t *testing.T) {
565
t.Parallel()
567
nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
568
n.SetIPFSConfig("Provide.Strategy", "")
569
})
570
+ publisher := nodes[0]
571
+ if sweep {
572
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
573
+ }
574
367
- cid := nodes[0].IPFSAddStr(time.Now().String())
575
+ cid := publisher.IPFSAddStr(time.Now().String())
576
577
nodes = nodes.StartDaemons().Connect()
578
defer nodes.StopDaemons()
371
- expectNoProviders(t, cid, nodes[1:]...)
372
-
373
- nodes[0].IPFS("routing", "reprovide")
579
+ peers := nodes[1:]
580
375
- expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
581
+ verifyReprovide(t, publisher, peers, 1, // 1 block added
582
+ []string{cid}, nil)
583
})
584
585
t.Run("Reprovides with 'all' strategy", func(t *testing.T) {
588
nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
589
n.SetIPFSConfig("Provide.Strategy", "all")
590
})
591
+ publisher := nodes[0]
592
+ if sweep {
593
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
594
+ }
595
385
- cid := nodes[0].IPFSAddStr(time.Now().String())
596
+ cid := publisher.IPFSAddStr(time.Now().String())
597
598
nodes = nodes.StartDaemons().Connect()
599
defer nodes.StopDaemons()
389
- expectNoProviders(t, cid, nodes[1:]...)
600
+ peers := nodes[1:]
601
391
- nodes[0].IPFS("routing", "reprovide")
392
-
393
- expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
602
+ verifyReprovide(t, publisher, peers, 1, // 1 block added
603
+ []string{cid}, nil)
604
})
605
606
t.Run("Reprovides with 'pinned' strategy", func(t *testing.T) {
612
nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
613
n.SetIPFSConfig("Provide.Strategy", "pinned")
614
})
615
+ publisher := nodes[0]
616
+ if sweep {
617
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
618
+ }
619
406
- // Add a pin while offline so it cannot be provided
407
- cidBarDir := nodes[0].IPFSAdd(bytes.NewReader(bar), "-Q", "-w")
620
+ // Add a pin while offline
621
+ cidBarDir := publisher.IPFSAdd(bytes.NewReader(bar), "-Q", "-w")
622
623
nodes = nodes.StartDaemons().Connect()
624
defer nodes.StopDaemons()
625
+ peers := nodes[1:]
626
412
- // Add content without pinning while daemon line
413
- cidFoo := nodes[0].IPFSAdd(bytes.NewReader(foo), "--pin=false")
414
- cidBar := nodes[0].IPFSAdd(bytes.NewReader(bar), "--pin=false")
415
-
416
- // Nothing should have been provided. The pin was offline, and
417
- // the others should not be provided per the strategy.
418
- expectNoProviders(t, cidFoo, nodes[1:]...)
419
- expectNoProviders(t, cidBar, nodes[1:]...)
420
- expectNoProviders(t, cidBarDir, nodes[1:]...)
627
+ // Add content without pinning while daemon is online
628
+ cidFoo := publisher.IPFSAdd(bytes.NewReader(foo), "--pin=false")
629
+ cidBar := publisher.IPFSAdd(bytes.NewReader(bar), "--pin=false")
630
422
- nodes[0].IPFS("routing", "reprovide")
423
-
424
- // cidFoo is not pinned so should not be provided.
425
- expectNoProviders(t, cidFoo, nodes[1:]...)
426
- // cidBar gets provided by being a child from cidBarDir even though we added with pin=false.
427
- expectProviders(t, cidBar, nodes[0].PeerID().String(), nodes[1:]...)
428
- expectProviders(t, cidBarDir, nodes[0].PeerID().String(), nodes[1:]...)
631
+ verifyReprovide(t, publisher, peers, 2, // cidBar + cidBarDir (bar is child of the wrapped dir pin)
632
+ []string{cidBar, cidBarDir},
633
+ []string{cidFoo}) // cidFoo not pinned
634
})
635
636
t.Run("Reprovides with 'roots' strategy", func(t *testing.T) {
637
t.Parallel()
638
434
- foo := random.Bytes(1000)
639
bar := random.Bytes(1000)
640
641
nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
642
n.SetIPFSConfig("Provide.Strategy", "roots")
643
})
440
- n0pid := nodes[0].PeerID().String()
644
+ publisher := nodes[0]
645
+ if sweep {
646
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
647
+ }
648
442
- // Add a pin. Only root should get pinned but not provided
443
- // because node not started
444
- cidBarDir := nodes[0].IPFSAdd(bytes.NewReader(bar), "-Q", "-w")
649
+ // Compute the child CID without storing anything (safe
650
+ // offline, daemon not started yet).
651
+ cidChild := publisher.IPFSAdd(bytes.NewReader(bar), "-Q", "--only-hash")
652
+ // Add with -w: pins the wrapper directory as root. The file
653
+ // inside is a child block of that pin, not a root.
654
+ cidRoot := publisher.IPFSAdd(bytes.NewReader(bar), "-Q", "-w")
655
656
nodes = nodes.StartDaemons().Connect()
657
defer nodes.StopDaemons()
658
+ peers := nodes[1:]
659
449
- cidFoo := nodes[0].IPFSAdd(bytes.NewReader(foo))
450
- cidBar := nodes[0].IPFSAdd(bytes.NewReader(bar), "--pin=false")
451
-
452
- // cidFoo will get provided per the strategy but cidBar will not.
453
- expectProviders(t, cidFoo, n0pid, nodes[1:]...)
454
- expectNoProviders(t, cidBar, nodes[1:]...)
455
-
456
- nodes[0].IPFS("routing", "reprovide")
457
-
458
- expectProviders(t, cidFoo, n0pid, nodes[1:]...)
459
- expectNoProviders(t, cidBar, nodes[1:]...)
460
- expectProviders(t, cidBarDir, n0pid, nodes[1:]...)
660
+ verifyReprovide(t, publisher, peers, 1, // cidRoot (only pin root)
661
+ []string{cidRoot},
662
+ []string{cidChild}) // child of pin, not a root
663
})
664
665
t.Run("Reprovides with 'mfs' strategy", func(t *testing.T) {
670
nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
671
n.SetIPFSConfig("Provide.Strategy", "mfs")
672
})
471
- n0pid := nodes[0].PeerID().String()
673
+ publisher := nodes[0]
674
+ if sweep {
675
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
676
+ }
677
473
- // add something and lets put it in MFS
474
- cidBar := nodes[0].IPFSAdd(bytes.NewReader(bar), "--pin=false", "-Q")
475
- nodes[0].IPFS("files", "cp", "/ipfs/"+cidBar, "/myfile")
678
+ // Add to MFS (should be provided)
679
+ cidMFS := publisher.IPFSAdd(bytes.NewReader(bar), "--pin=false", "-Q")
680
+ publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
681
+ // Pin something NOT in MFS (should NOT be provided)
682
+ cidPinned := publisher.IPFSAddStr(uniq("pinned but not mfs"))
683
684
nodes = nodes.StartDaemons().Connect()
685
defer nodes.StopDaemons()
686
+ peers := nodes[1:]
687
480
- // cidBar is in MFS but not provided
481
- expectNoProviders(t, cidBar, nodes[1:]...)
482
-
483
- nodes[0].IPFS("routing", "reprovide")
484
-
485
- // And now is provided
486
- expectProviders(t, cidBar, n0pid, nodes[1:]...)
688
+ verifyReprovide(t, publisher, peers, 1, // cidMFS only
689
+ []string{cidMFS},
690
+ []string{cidPinned}) // mfs strategy ignores pinned content outside MFS
691
})
692
693
t.Run("Reprovides with 'pinned+mfs' strategy", func(t *testing.T) {
696
nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
697
n.SetIPFSConfig("Provide.Strategy", "pinned+mfs")
698
})
495
- n0pid := nodes[0].PeerID().String()
699
+ publisher := nodes[0]
700
+ if sweep {
701
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
702
+ }
703
704
// Add a pinned CID (should be provided)
498
- cidPinned := nodes[0].IPFSAddStr("pinned content", "--pin=true")
705
+ cidPinned := publisher.IPFSAddStr(uniq("pinned content"), "--pin=true")
706
// Add a CID to MFS (should be provided)
500
- cidMFS := nodes[0].IPFSAddStr("mfs content")
501
- nodes[0].IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
707
+ cidMFS := publisher.IPFSAddStr(uniq("mfs content"))
708
+ publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
709
// Add a CID that is neither pinned nor in MFS (should not be provided)
503
- cidNeither := nodes[0].IPFSAddStr("neither content", "--pin=false")
710
+ cidNeither := publisher.IPFSAddStr(uniq("neither content"), "--pin=false")
711
+
712
+ nodes = nodes.StartDaemons().Connect()
713
+ defer nodes.StopDaemons()
714
+ peers := nodes[1:]
715
+
716
+ verifyReprovide(t, publisher, peers, 2, // cidPinned + cidMFS
717
+ []string{cidPinned, cidMFS},
718
+ []string{cidNeither}) // neither pinned nor in MFS
719
+ })
720
+
721
+ t.Run("Reprovides with 'pinned+mfs+unique' strategy", func(t *testing.T) {
722
+ t.Parallel()
723
+
724
+ nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
725
+ n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+unique")
726
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
727
+ })
728
+ publisher := nodes[0]
729
+ if sweep {
730
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
731
+ }
732
+
733
+ // Build a directory DAG with a multi-chunk file in MFS, then pin it.
734
+ cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
735
+ publisher.IPFS("pin", "add", cidRoot)
736
+ cidUnpinned := publisher.IPFSAddStr(uniq("unpinned content"), "--pin=false")
737
738
nodes = nodes.StartDaemons().Connect()
739
defer nodes.StopDaemons()
740
+ peers := nodes[1:]
741
+
742
+ // +unique provides all blocks in pinned DAGs (same as pinned+mfs)
743
+ verifyReprovide(t, publisher, peers, 4, // root + subdir + file + chunks
744
+ []string{cidRoot, cidSubdir, cidFile, cidChunk},
745
+ []string{cidUnpinned})
746
+ })
747
+
748
+ t.Run("Reprovides with 'pinned+mfs+entities' strategy", func(t *testing.T) {
749
+ t.Parallel()
750
+
751
+ nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
752
+ n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+entities")
753
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
754
+ })
755
+ publisher := nodes[0]
756
+ if sweep {
757
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
758
+ }
759
508
- // Trigger reprovide
509
- nodes[0].IPFS("routing", "reprovide")
760
+ // Build a directory DAG with a multi-chunk file in MFS, then pin it.
761
+ cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
762
+ publisher.IPFS("pin", "add", cidRoot)
763
511
- // Check that pinned CID is provided
512
- expectProviders(t, cidPinned, n0pid, nodes[1:]...)
513
- // Check that MFS CID is provided
514
- expectProviders(t, cidMFS, n0pid, nodes[1:]...)
515
- // Check that neither CID is not provided
516
- expectNoProviders(t, cidNeither, nodes[1:]...)
764
+ nodes = nodes.StartDaemons().Connect()
765
+ defer nodes.StopDaemons()
766
+ peers := nodes[1:]
767
+
768
+ // Entity roots: directories and file root (not chunks)
769
+ verifyReprovide(t, publisher, peers, 3, // root + subdir + file (not chunks)
770
+ []string{cidRoot, cidSubdir, cidFile},
771
+ []string{cidChunk}) // chunks skipped by +entities
772
})
773
}
774
969
Schedule struct {
970
NextReprovidePrefix string `json:"next_reprovide_prefix"`
971
} `json:"schedule"`
972
+ Operations struct {
973
+ Ongoing struct {
974
+ KeyReprovides int `json:"key_reprovides"`
975
+ } `json:"ongoing"`
976
+ Past struct {
977
+ KeysProvided int64 `json:"keys_provided"`
978
+ } `json:"past"`
979
+ } `json:"operations"`
980
+ Queues struct {
981
+ PendingKeyProvides int64 `json:"pending_key_provides"`
982
+ } `json:"queues"`
983
} `json:"Sweep"`
984
}
985
986
// parseProvideStatJSON extracts timing and schedule information from
987
// the JSON output of 'ipfs provide stat --enc=json'.
722
-// Note: prefix is unused in current tests but kept for potential future use.
988
func parseProvideStatJSON(output string) (offset time.Duration, prefix string, err error) {
989
var stat provideStatJSON
990
if err := json.Unmarshal([]byte(output), &stat); err != nil {
995
return offset, prefix, nil
996
}
997
998
+// waitForSweepReprovide polls `provide stat --enc=json` until the
999
+// sweep provider has provided at least minCIDs and no work is pending.
1000
+// Pass 0 for minCIDs to just wait for any provide activity to finish.
1001
+// Returns the total CIDs provided so far (for use as minCIDs in a
1002
+// subsequent call to wait for the next cycle).
1003
+// The importing node must have a short Provide.DHT.Interval so the
1004
+// reprovide cycle completes within the timeout.
1005
+func waitForSweepReprovide(t *testing.T, n *harness.Node, timeout time.Duration, minCIDs int64) int64 {
1006
+ t.Helper()
1007
+ if minCIDs == 0 {
1008
+ minCIDs = 1
1009
+ }
1010
+ deadline := time.Now().Add(timeout)
1011
+ for time.Now().Before(deadline) {
1012
+ res := n.RunIPFS("provide", "stat", "--enc=json")
1013
+ if res.ExitCode() == 0 {
1014
+ var stat provideStatJSON
1015
+ if err := json.Unmarshal(res.Stdout.Bytes(), &stat); err == nil {
1016
+ s := stat.Sweep
1017
+ if s.Operations.Past.KeysProvided >= minCIDs &&
1018
+ s.Queues.PendingKeyProvides == 0 &&
1019
+ s.Operations.Ongoing.KeyReprovides == 0 {
1020
+ return s.Operations.Past.KeysProvided
1021
+ }
1022
+ }
1023
+ }
1024
+ time.Sleep(500 * time.Millisecond)
1025
+ }
1026
+ t.Fatalf("sweep reprovide: expected at least %d CIDs provided within %s", minCIDs, timeout)
1027
+ return 0
1028
+}
1029
+
1030
func TestProvider(t *testing.T) {
1031
t.Parallel()
1032
1033
variants := []struct {
737
- name string
738
- reprovide bool
739
- apply cfgApplier
1034
+ name string
1035
+ sweep bool
1036
+ apply cfgApplier
1037
+ awaitReprovide awaitReprovideFunc
1038
}{
1039
{
742
- name: "LegacyProvider",
743
- reprovide: true,
1040
+ name: "LegacyProvider",
1041
+ sweep: false,
1042
apply: func(n *harness.Node) {
1043
n.SetIPFSConfig("Provide.DHT.SweepEnabled", false)
1044
},
1045
+ // `routing reprovide` blocks until the cycle finishes.
1046
+ // minCIDs is ignored (legacy has no stat counter).
1047
+ awaitReprovide: func(t *testing.T, n *harness.Node, minCIDs int64) int64 {
1048
+ n.IPFS("routing", "reprovide")
1049
+ return minCIDs
1050
+ },
1051
},
1052
{
749
- name: "SweepingProvider",
750
- reprovide: false,
1053
+ name: "SweepingProvider",
1054
+ sweep: true,
1055
apply: func(n *harness.Node) {
1056
n.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1057
},
1058
+ // No manual trigger exists for sweep. Poll `provide stat`
1059
+ // until the reprovide cycle completes.
1060
+ awaitReprovide: func(t *testing.T, n *harness.Node, minCIDs int64) int64 {
1061
+ // 90s accounts for provider bootstrap time (connecting
1062
+ // to ephemeral peers, measuring prefix length) before
1063
+ // the 30s reprovide cycle starts. On CI with parallel
1064
+ // tests, bootstrap can take 20-30s.
1065
+ return waitForSweepReprovide(t, n, 90*time.Second, minCIDs)
1066
+ },
1067
},
1068
}
1069
1070
for _, v := range variants {
1071
t.Run(v.name, func(t *testing.T) {
1072
// t.Parallel()
760
- runProviderSuite(t, v.reprovide, v.apply)
1073
+ runProviderSuite(t, v.sweep, v.apply, v.awaitReprovide)
1074
1075
// Resume tests only apply to SweepingProvider
763
- if v.name == "SweepingProvider" {
1076
+ if v.sweep {
1077
runResumeTests(t, v.apply)
1078
}
1079
})
1080
}
1081
}
1082
1083
+// TestProviderUniqueDedupLogging verifies that the +unique bloom filter
1084
+// deduplication produces a "skippedBranches" log with a value > 0 when
1085
+// two pins share content. Tests both the fast-provide-dag path (immediate
1086
+// provide on pin add) and the reprovide cycle path.
1087
+func TestProviderUniqueDedupLogging(t *testing.T) {
1088
+ t.Parallel()
1089
+
1090
+ // Shared data that both pins will reference. Two pins containing
1091
+ // the same file block give the bloom something to dedup.
1092
+ sharedData := random.Bytes(10 * 1024) // 10 KiB, single block
1093
+
1094
+ t.Run("fast-provide-dag dedup across pins in single call", func(t *testing.T) {
1095
+ t.Parallel()
1096
+
1097
+ h := harness.NewT(t)
1098
+ node := h.NewNode().Init()
1099
+ node.SetIPFSConfig("Provide.Strategy", "pinned+unique")
1100
+ node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1101
+ node.SetIPFSConfig("Import.UnixFSChunker", "size-5120") // 5 KiB chunks
1102
+ h.BootstrapWithStubDHT(harness.Nodes{node})
1103
+
1104
+ node.StartDaemonWithReq(harness.RunRequest{
1105
+ CmdOpts: []harness.CmdOpt{
1106
+ harness.RunWithEnv(map[string]string{
1107
+ // dagwalker: bloom creation log
1108
+ // core/commands/cmdenv: fast-provide-dag finished log
1109
+ "GOLOG_LOG_LEVEL": "error,dagwalker=info,core/commands/cmdenv=info",
1110
+ }),
1111
+ },
1112
+ }, "")
1113
+ defer node.StopDaemon()
1114
+
1115
+ // 10 KiB file with 5 KiB chunks = 1 file root + 2 chunks = 3 blocks.
1116
+ // Two dirs each containing the file under different names:
1117
+ // dirA/fileA → same 3 blocks
1118
+ // dirB/fileB → same 3 blocks
1119
+ // Pinning both in a single `pin add` shares one bloom tracker.
1120
+ // Walking dirA: dirA + file root + chunk1 + chunk2 = 4 provided.
1121
+ // Walking dirB: dirB + file root (bloom hit, skip subtree) = 1 provided, 1 skipped.
1122
+ // Total: 5 provided, 1 skipped branch (file root in dirB; its
1123
+ // 2 chunks are never visited because the parent was skipped).
1124
+ cidFile := node.IPFSAdd(bytes.NewReader(sharedData), "-Q", "--pin=false")
1125
+ node.IPFS("files", "mkdir", "-p", "/dirA")
1126
+ node.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirA/fileA")
1127
+ cidDirA := node.IPFS("files", "stat", "--hash", "/dirA").Stdout.Trimmed()
1128
+ node.IPFS("files", "mkdir", "-p", "/dirB")
1129
+ node.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirB/fileB")
1130
+ cidDirB := node.IPFS("files", "stat", "--hash", "/dirB").Stdout.Trimmed()
1131
+ require.NotEqual(t, cidDirA, cidDirB, "dirs must differ to test dedup")
1132
+ // Single pin add with both CIDs shares one bloom.
1133
+ node.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidDirA, cidDirB)
1134
+
1135
+ daemonLog := node.Daemon.Stderr.String()
1136
+ require.Contains(t, daemonLog, "bloom tracker created")
1137
+ require.NotContains(t, daemonLog, "bloom tracker autoscaled")
1138
+ require.Contains(t, daemonLog, `"providedCIDs": 5`)
1139
+ require.Contains(t, daemonLog, `"skippedBranches": 1`)
1140
+ })
1141
+
1142
+ t.Run("reprovide cycle dedup across pins", func(t *testing.T) {
1143
+ t.Parallel()
1144
+
1145
+ h := harness.NewT(t)
1146
+ nodes := h.NewNodes(2).Init()
1147
+ for _, n := range nodes {
1148
+ n.SetIPFSConfig("Provide.Strategy", "pinned+unique")
1149
+ n.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1150
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-5120") // 5 KiB chunks
1151
+ }
1152
+ publisher := nodes[0]
1153
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
1154
+ h.BootstrapWithStubDHT(nodes)
1155
+
1156
+ // Same file structure as fast-provide-dag test above.
1157
+ // The reprovide cycle walks all recursive pins:
1158
+ // pin dirA: dirA + file root + chunk1 + chunk2 = 4 provided
1159
+ // pin empty MFS root (always present): 1 provided
1160
+ // pin dirB: dirB + file root (bloom hit, skip subtree) = 1 provided, 1 skipped
1161
+ // Total: 6 provided, 1 skipped branch.
1162
+ cidFile := publisher.IPFSAdd(bytes.NewReader(sharedData), "-Q", "--pin=false")
1163
+ publisher.IPFS("files", "mkdir", "-p", "/dirA")
1164
+ publisher.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirA/fileA")
1165
+ cidDirA := publisher.IPFS("files", "stat", "--hash", "/dirA").Stdout.Trimmed()
1166
+ publisher.IPFS("pin", "add", cidDirA)
1167
+ publisher.IPFS("files", "mkdir", "-p", "/dirB")
1168
+ publisher.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirB/fileB")
1169
+ cidDirB := publisher.IPFS("files", "stat", "--hash", "/dirB").Stdout.Trimmed()
1170
+ require.NotEqual(t, cidDirA, cidDirB, "dirs must differ to test dedup")
1171
+ publisher.IPFS("pin", "add", cidDirB)
1172
+
1173
+ nodes[0].StartDaemonWithReq(harness.RunRequest{
1174
+ CmdOpts: []harness.CmdOpt{
1175
+ harness.RunWithEnv(map[string]string{
1176
+ "GOLOG_LOG_LEVEL": "error,dagwalker=info,core:constructor=info",
1177
+ }),
1178
+ },
1179
+ }, "")
1180
+ nodes[1].StartDaemon()
1181
+ defer nodes.StopDaemons()
1182
+ nodes.Connect()
1183
+
1184
+ waitForSweepReprovide(t, publisher, 90*time.Second, 6)
1185
+
1186
+ daemonLog := publisher.Daemon.Stderr.String()
1187
+ require.Contains(t, daemonLog, "bloom tracker created")
1188
+ require.NotContains(t, daemonLog, "bloom tracker autoscaled")
1189
+ require.Contains(t, daemonLog, `"providedCIDs": 6`)
1190
+ require.Contains(t, daemonLog, `"skippedBranches": 1`)
1191
+ })
1192
+}
1193
+
1194
+// TestProviderFastProvideDAGAsyncSurvives verifies that
1195
+// --fast-provide-dag without --fast-provide-wait runs a background
1196
+// DAG walk that outlives the command handler and publishes every
1197
+// block of the newly added DAG to the routing system.
1198
+//
1199
+// The async walk runs in a goroutine parented on the IpfsNode
1200
+// lifetime context (not req.Context), so it keeps running after
1201
+// `ipfs add` returns and is only cancelled on daemon shutdown.
1202
+//
1203
+// Provide.DHT.Interval is set high so the scheduled reprovide
1204
+// cycle cannot fire during the test window. That makes the async
1205
+// walk the only path that can publish non-root block CIDs.
1206
+func TestProviderFastProvideDAGAsyncSurvives(t *testing.T) {
1207
+ t.Parallel()
1208
+
1209
+ h := harness.NewT(t)
1210
+ nodes := h.NewNodes(2).Init()
1211
+ for _, n := range nodes {
1212
+ n.SetIPFSConfig("Provide.Strategy", "pinned")
1213
+ n.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1214
+ // Small chunks so a modest file produces many leaf blocks.
1215
+ n.SetIPFSConfig("Import.UnixFSChunker", "size-1024")
1216
+ }
1217
+ publisher, peers := nodes[0], nodes[1:]
1218
+ publisher.SetIPFSConfig("Provide.DHT.Interval", "1h")
1219
+ h.BootstrapWithStubDHT(nodes)
1220
+
1221
+ publisher.StartDaemonWithReq(harness.RunRequest{
1222
+ CmdOpts: []harness.CmdOpt{
1223
+ harness.RunWithEnv(map[string]string{
1224
+ "GOLOG_LOG_LEVEL": "error,core/commands/cmdenv=info",
1225
+ }),
1226
+ },
1227
+ }, "")
1228
+ nodes[1].StartDaemon()
1229
+ defer nodes.StopDaemons()
1230
+ nodes.Connect()
1231
+
1232
+ // 16 KiB + 1 KiB chunks yields a file root plus many leaf
1233
+ // blocks, so the providedCIDs count after the walk is
1234
+ // unambiguous.
1235
+ data := random.Bytes(16 * 1024)
1236
+ cidFile := publisher.IPFSAdd(bytes.NewReader(data), "-Q",
1237
+ "--pin=true",
1238
+ "--fast-provide-dag=true",
1239
+ // --fast-provide-wait deliberately omitted: the walk
1240
+ // runs in the background after `ipfs add` returns.
1241
+ )
1242
+
1243
+ // Pull a chunk CID out of the file DAG. Chunks are not pin
1244
+ // roots, so fast-provide-root does not touch them; only the
1245
+ // DAG walk can announce them.
1246
+ dagOut := publisher.IPFS("dag", "get", cidFile)
1247
+ var dagNode struct {
1248
+ Links []struct {
1249
+ Hash map[string]string `json:"Hash"`
1250
+ } `json:"Links"`
1251
+ }
1252
+ require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
1253
+ require.Greater(t, len(dagNode.Links), 1, "file should have multiple chunks")
1254
+ cidChunk := dagNode.Links[0].Hash["/"]
1255
+ require.NotEmpty(t, cidChunk)
1256
+
1257
+ // The async walk logs "fast-provide-dag: finished" with a
1258
+ // providedCIDs count on completion. A full walk of this file
1259
+ // visits the root plus every leaf chunk, so the count is much
1260
+ // larger than 2.
1261
+ providedRe := regexp.MustCompile(`"providedCIDs": (\d+)`)
1262
+ var providedCount int
1263
+ require.Eventually(t, func() bool {
1264
+ m := providedRe.FindStringSubmatch(publisher.Daemon.Stderr.String())
1265
+ if len(m) != 2 {
1266
+ return false
1267
+ }
1268
+ n, err := strconv.Atoi(m[1])
1269
+ if err != nil {
1270
+ return false
1271
+ }
1272
+ providedCount = n
1273
+ return true
1274
+ }, 30*time.Second, 200*time.Millisecond, "async fast-provide-dag walk did not log 'finished'")
1275
+
1276
+ require.Greater(t, providedCount, 2,
1277
+ "providedCIDs=%d is too small for a full walk of the file DAG", providedCount)
1278
+
1279
+ // End-to-end: the peer can find the publisher as a provider
1280
+ // for a chunk CID, which only the async walk could have
1281
+ // announced within the test window.
1282
+ pid := publisher.PeerID().String()
1283
+ var found bool
1284
+ for _, peer := range peers {
1285
+ for i := time.Duration(0); i*timeStep < timeout; i++ {
1286
+ res := peer.IPFS("routing", "findprovs", "-n=1", cidChunk)
1287
+ if res.Stdout.Trimmed() == pid {
1288
+ found = true
1289
+ break
1290
+ }
1291
+ }
1292
+ }
1293
+ require.True(t, found, "chunk %s not announced by the async walk", cidChunk)
1294
+}
1295
+
1296
// TestHTTPOnlyProviderWithSweepEnabled tests that provider records are correctly
1297
// sent to HTTP routers when Routing.Type="custom" with only HTTP routers configured,
1298
// even when Provide.DHT.SweepEnabled=true (the default since v0.39).