@cryptotaxi247 / kubo
Code Commits Issues Pulls Sessions CI/CD
master
master
go 1,941 lines 69.3 KB
Raw
1 package cli
2
3 import (
4 "bytes"
5 "encoding/json"
6 "fmt"
7 "net/http"
8 "net/http/httptest"
9 "os"
10 "path/filepath"
11 "regexp"
12 "strconv"
13 "strings"
14 "sync/atomic"
15 "testing"
16 "time"
17
18 "github.com/ipfs/go-test/random"
19 "github.com/ipfs/kubo/test/cli/harness"
20 "github.com/stretchr/testify/assert"
21 "github.com/stretchr/testify/require"
22 )
23
24 const (
25 timeStep = 20 * time.Millisecond
26 timeout = 30 * time.Second
27 )
28
29 type cfgApplier func(*harness.Node)
30
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 {
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 {
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
66 expectNoProviders := func(t *testing.T, cid string, nodes ...*harness.Node) {
67 for _, node := range nodes {
68 res := node.IPFS("routing", "findprovs", "-n=1", cid)
69 require.Empty(t, res.Stdout.String())
70 }
71 }
72
73 expectProviders := func(t *testing.T, cid, expectedProvider string, nodes ...*harness.Node) {
74 outerLoop:
75 for _, node := range nodes {
76 for i := time.Duration(0); i*timeStep < timeout; i++ {
77 res := node.IPFS("routing", "findprovs", "-n=1", cid)
78 if res.Stdout.Trimmed() == expectedProvider {
79 continue outerLoop
80 }
81 }
82 require.FailNowf(t, "found no providers", "expected a provider for %s", cid)
83 }
84 }
85
86 t.Run("Provide.Enabled=true announces new CIDs created by ipfs add", func(t *testing.T) {
87 t.Parallel()
88
89 nodes := initNodes(t, 2, func(n *harness.Node) {
90 n.SetIPFSConfig("Provide.Enabled", true)
91 })
92 defer nodes.StopDaemons()
93
94 cid := nodes[0].IPFSAddStr(time.Now().String())
95 expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
96 })
97
98 t.Run("Provide.Enabled=true announces new CIDs created by ipfs add --pin=false with default strategy", func(t *testing.T) {
99 t.Parallel()
100
101 nodes := initNodes(t, 2, func(n *harness.Node) {
102 n.SetIPFSConfig("Provide.Enabled", true)
103 // Default strategy is "all" which should provide even unpinned content
104 })
105 defer nodes.StopDaemons()
106
107 cid := nodes[0].IPFSAddStr(time.Now().String(), "--pin=false")
108 expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
109 })
110
111 t.Run("Provide.Enabled=true announces new CIDs created by ipfs block put --pin=false with default strategy", func(t *testing.T) {
112 t.Parallel()
113
114 nodes := initNodes(t, 2, func(n *harness.Node) {
115 n.SetIPFSConfig("Provide.Enabled", true)
116 // Default strategy is "all" which should provide unpinned content from block put
117 })
118 defer nodes.StopDaemons()
119
120 data := random.Bytes(256)
121 cid := nodes[0].IPFSBlockPut(bytes.NewReader(data), "--pin=false")
122 expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
123 })
124
125 t.Run("Provide.Enabled=true announces new CIDs created by ipfs dag put --pin=false with default strategy", func(t *testing.T) {
126 t.Parallel()
127
128 nodes := initNodes(t, 2, func(n *harness.Node) {
129 n.SetIPFSConfig("Provide.Enabled", true)
130 // Default strategy is "all" which should provide unpinned content from dag put
131 })
132 defer nodes.StopDaemons()
133
134 dagData := `{"hello": "world", "timestamp": "` + time.Now().String() + `"}`
135 cid := nodes[0].IPFSDAGPut(bytes.NewReader([]byte(dagData)), "--pin=false")
136 expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
137 })
138
139 t.Run("Provide.Enabled=false disables announcement of new CID from ipfs add", func(t *testing.T) {
140 t.Parallel()
141
142 nodes := initNodes(t, 2, func(n *harness.Node) {
143 n.SetIPFSConfig("Provide.Enabled", false)
144 })
145 defer nodes.StopDaemons()
146
147 cid := nodes[0].IPFSAddStr(time.Now().String())
148 expectNoProviders(t, cid, nodes[1:]...)
149 })
150
151 t.Run("Provide.Enabled=false disables manual announcement via RPC command", func(t *testing.T) {
152 t.Parallel()
153
154 nodes := initNodes(t, 2, func(n *harness.Node) {
155 n.SetIPFSConfig("Provide.Enabled", false)
156 })
157 defer nodes.StopDaemons()
158
159 cid := nodes[0].IPFSAddStr(time.Now().String())
160 res := nodes[0].RunIPFS("routing", "provide", cid)
161 assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.Enabled is set to 'false'")
162 assert.Equal(t, 1, res.ExitCode())
163
164 expectNoProviders(t, cid, nodes[1:]...)
165 })
166
167 t.Run("manual provide fails when no libp2p peers and no custom HTTP router", func(t *testing.T) {
168 t.Parallel()
169
170 h := harness.NewT(t)
171 node := h.NewNode().Init()
172 apply(node)
173 node.SetIPFSConfig("Provide.Enabled", true)
174 node.StartDaemon()
175 defer node.StopDaemon()
176
177 cid := node.IPFSAddStr(time.Now().String())
178 res := node.RunIPFS("routing", "provide", cid)
179 assert.Contains(t, res.Stderr.Trimmed(), "cannot provide, no connected peers")
180 assert.Equal(t, 1, res.ExitCode())
181 })
182
183 t.Run("manual provide succeeds via custom HTTP router when no libp2p peers", func(t *testing.T) {
184 t.Parallel()
185
186 // Create a mock HTTP server that accepts provide requests.
187 // This simulates the undocumented API behavior described in
188 // https://discuss.ipfs.tech/t/only-peers-found-from-dht-seem-to-be-getting-used-as-relays-so-cant-use-http-routers/19545/9
189 // Note: This is NOT IPIP-378, which was not implemented.
190 mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
191 // Accept both PUT and POST requests to /routing/v1/providers and /routing/v1/ipns
192 if (r.Method == http.MethodPut || r.Method == http.MethodPost) &&
193 (strings.HasPrefix(r.URL.Path, "/routing/v1/providers") || strings.HasPrefix(r.URL.Path, "/routing/v1/ipns")) {
194 // Return HTTP 200 to indicate successful publishing
195 w.WriteHeader(http.StatusOK)
196 } else {
197 w.WriteHeader(http.StatusNotFound)
198 }
199 }))
200 defer mockServer.Close()
201
202 h := harness.NewT(t)
203 node := h.NewNode().Init()
204 apply(node)
205 node.SetIPFSConfig("Provide.Enabled", true)
206 // Configure a custom HTTP router for providing.
207 // Using our mock server that will accept the provide requests.
208 routingConf := map[string]any{
209 "Type": "custom", // https://github.com/ipfs/kubo/blob/master/docs/delegated-routing.md#configuration-file-example
210 "Methods": map[string]any{
211 "provide": map[string]any{"RouterName": "MyCustomRouter"},
212 "get-ipns": map[string]any{"RouterName": "MyCustomRouter"},
213 "put-ipns": map[string]any{"RouterName": "MyCustomRouter"},
214 "find-peers": map[string]any{"RouterName": "MyCustomRouter"},
215 "find-providers": map[string]any{"RouterName": "MyCustomRouter"},
216 },
217 "Routers": map[string]any{
218 "MyCustomRouter": map[string]any{
219 "Type": "http",
220 "Parameters": map[string]any{
221 // Use the mock server URL
222 "Endpoint": mockServer.URL,
223 },
224 },
225 },
226 }
227 node.SetIPFSConfig("Routing", routingConf)
228 node.StartDaemon()
229 defer node.StopDaemon()
230
231 cid := node.IPFSAddStr(time.Now().String())
232 // The command should successfully provide via HTTP even without libp2p peers
233 res := node.RunIPFS("routing", "provide", cid)
234 assert.Empty(t, res.Stderr.String(), "Should have no errors when providing via HTTP router")
235 assert.Equal(t, 0, res.ExitCode(), "Should succeed with exit code 0")
236 })
237
238 t.Run("ipfs provide once works when Provide.DHT.Interval=0", func(t *testing.T) {
239 t.Parallel()
240
241 nodes := initNodes(t, 2, func(n *harness.Node) {
242 n.SetIPFSConfig("Provide.Enabled", true)
243 // No periodic reprovide schedule; provide once is the only
244 // way new content reaches peers in this configuration.
245 n.SetIPFSConfig("Provide.DHT.Interval", "0")
246 n.SetIPFSConfig("Provide.Strategy", "roots")
247 })
248 defer nodes.StopDaemons()
249
250 publisher := nodes[0]
251 cid := publisher.IPFSAddStr(uniq("interval=0"), "--pin=false")
252 expectNoProviders(t, cid, nodes[1:]...)
253
254 res := publisher.RunIPFS("provide", "once", cid)
255 assert.Equal(t, 0, res.ExitCode(), "provide once should succeed with Interval=0")
256 expectProviders(t, cid, publisher.PeerID().String(), nodes[1:]...)
257 })
258
259 t.Run("Provide.Enabled=false disables ipfs provide once", func(t *testing.T) {
260 t.Parallel()
261
262 nodes := initNodes(t, 2, func(n *harness.Node) {
263 n.SetIPFSConfig("Provide.Enabled", false)
264 })
265 defer nodes.StopDaemons()
266
267 cid := nodes[0].IPFSAddStr(time.Now().String())
268 res := nodes[0].RunIPFS("provide", "once", cid)
269 assert.Contains(t, res.Stderr.Trimmed(), "cannot provide: Provide.Enabled is false")
270 assert.Equal(t, 1, res.ExitCode())
271
272 expectNoProviders(t, cid, nodes[1:]...)
273 })
274
275 t.Run("ipfs provide once announces a CID and finds providers", func(t *testing.T) {
276 t.Parallel()
277
278 nodes := initNodes(t, 2, func(n *harness.Node) {
279 n.SetIPFSConfig("Provide.Enabled", true)
280 // "roots" so add-time providing is skipped and we know the
281 // announcement comes from `provide once`, not from ipfs add.
282 n.SetIPFSConfig("Provide.Strategy", "roots")
283 })
284 defer nodes.StopDaemons()
285
286 cid := nodes[0].IPFSAddStr(uniq("provide once"), "--pin=false")
287 expectNoProviders(t, cid, nodes[1:]...)
288
289 res := nodes[0].RunIPFS("provide", "once", cid)
290 assert.Equal(t, 0, res.ExitCode(), "provide once should succeed")
291 expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
292 })
293
294 t.Run("ipfs provide once errors when CID is not in local blockstore", func(t *testing.T) {
295 t.Parallel()
296
297 nodes := initNodes(t, 1, func(n *harness.Node) {
298 n.SetIPFSConfig("Provide.Enabled", true)
299 })
300 defer nodes.StopDaemons()
301
302 // CID for content the node has never seen.
303 missing := "bafkreigh2akiscaildcqabsyg3dfr6chu3fgpregiymsck7e7aqa4s52zy"
304 res := nodes[0].RunIPFS("provide", "once", missing)
305 assert.Contains(t, res.Stderr.Trimmed(), "not found locally, cannot provide")
306 assert.Equal(t, 1, res.ExitCode())
307 })
308
309 t.Run("ipfs provide once --recursive announces every block in the DAG", func(t *testing.T) {
310 t.Parallel()
311
312 nodes := initNodes(t, 2, func(n *harness.Node) {
313 n.SetIPFSConfig("Provide.Enabled", true)
314 // Selective strategy + --pin=false below means nothing is
315 // auto-provided; everything findable comes from `provide once`.
316 n.SetIPFSConfig("Provide.Strategy", "roots")
317 // 1 MiB chunks so a 2 MiB file produces multiple leaf blocks.
318 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576")
319 })
320 defer nodes.StopDaemons()
321
322 publisher := nodes[0]
323 data := random.Bytes(2 * 1024 * 1024)
324 cidRoot := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "--pin=false")
325
326 // Discover a chunk CID via the root's DAG links.
327 dagOut := publisher.IPFS("dag", "get", cidRoot)
328 var dagNode struct {
329 Links []struct {
330 Hash map[string]string `json:"Hash"`
331 } `json:"Links"`
332 }
333 require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
334 require.Greater(t, len(dagNode.Links), 1, "2 MiB file with 1 MiB chunker should have multiple chunks")
335 cidChunk := dagNode.Links[0].Hash["/"]
336 require.NotEmpty(t, cidChunk)
337
338 // Recursive provide should announce both the root and every chunk.
339 res := publisher.RunIPFS("provide", "once", "-r", cidRoot)
340 assert.Equal(t, 0, res.ExitCode(), "provide once -r should succeed")
341 expectProviders(t, cidRoot, publisher.PeerID().String(), nodes[1:]...)
342 expectProviders(t, cidChunk, publisher.PeerID().String(), nodes[1:]...)
343 })
344
345 t.Run("ipfs provide once accepts multiple CIDs and reports count", func(t *testing.T) {
346 t.Parallel()
347
348 nodes := initNodes(t, 2, func(n *harness.Node) {
349 n.SetIPFSConfig("Provide.Enabled", true)
350 n.SetIPFSConfig("Provide.Strategy", "roots")
351 })
352 defer nodes.StopDaemons()
353
354 publisher := nodes[0]
355 c1 := publisher.IPFSAddStr(uniq("multi 1"), "--pin=false")
356 c2 := publisher.IPFSAddStr(uniq("multi 2"), "--pin=false")
357 c3 := publisher.IPFSAddStr(uniq("multi 3"), "--pin=false")
358
359 res := publisher.RunIPFS("provide", "once", c1, c2, c3)
360 assert.Equal(t, 0, res.ExitCode(), "provide once with multiple CIDs should succeed")
361 assert.Contains(t, res.Stdout.Trimmed(), "queued 3 CID(s) for immediate provide")
362
363 expectProviders(t, c1, publisher.PeerID().String(), nodes[1:]...)
364 expectProviders(t, c2, publisher.PeerID().String(), nodes[1:]...)
365 expectProviders(t, c3, publisher.PeerID().String(), nodes[1:]...)
366 })
367
368 t.Run("ipfs provide once reads CIDs streamed from stdin", func(t *testing.T) {
369 t.Parallel()
370
371 nodes := initNodes(t, 2, func(n *harness.Node) {
372 n.SetIPFSConfig("Provide.Enabled", true)
373 n.SetIPFSConfig("Provide.Strategy", "roots")
374 })
375 defer nodes.StopDaemons()
376
377 publisher := nodes[0]
378 c1 := publisher.IPFSAddStr(uniq("stdin 1"), "--pin=false")
379 c2 := publisher.IPFSAddStr(uniq("stdin 2"), "--pin=false")
380 c3 := publisher.IPFSAddStr(uniq("stdin 3"), "--pin=false")
381
382 res := publisher.Runner.Run(harness.RunRequest{
383 Path: publisher.IPFSBin,
384 Args: []string{"provide", "once"},
385 CmdOpts: []harness.CmdOpt{
386 harness.RunWithStdinStr(c1 + "\n" + c2 + "\n" + c3 + "\n"),
387 },
388 })
389 assert.Equal(t, 0, res.ExitCode(), "provide once with stdin should succeed")
390 assert.Contains(t, res.Stdout.Trimmed(), "queued 3 CID(s) for immediate provide")
391
392 expectProviders(t, c1, publisher.PeerID().String(), nodes[1:]...)
393 expectProviders(t, c2, publisher.PeerID().String(), nodes[1:]...)
394 expectProviders(t, c3, publisher.PeerID().String(), nodes[1:]...)
395 })
396
397 t.Run("ipfs provide once deduplicates repeated CIDs", func(t *testing.T) {
398 t.Parallel()
399
400 nodes := initNodes(t, 1, func(n *harness.Node) {
401 n.SetIPFSConfig("Provide.Enabled", true)
402 n.SetIPFSConfig("Provide.Strategy", "roots")
403 })
404 defer nodes.StopDaemons()
405
406 publisher := nodes[0]
407 c1 := publisher.IPFSAddStr(uniq("dedup 1"), "--pin=false")
408 c2 := publisher.IPFSAddStr(uniq("dedup 2"), "--pin=false")
409
410 // 4 args, 2 unique CIDs. The repeated ones should not produce
411 // extra events on the wire.
412 res := publisher.RunIPFS("provide", "once", "--enc=json", c1, c2, c1, c2)
413 assert.Equal(t, 0, res.ExitCode())
414
415 var queued []string
416 for line := range strings.Lines(res.Stdout.String()) {
417 line = strings.TrimSpace(line)
418 if line == "" {
419 continue
420 }
421 var ev struct{ Queued string }
422 require.NoError(t, json.Unmarshal([]byte(line), &ev))
423 queued = append(queued, ev.Queued)
424 }
425 assert.ElementsMatch(t, []string{c1, c2}, queued, "duplicates should be filtered")
426 })
427
428 t.Run("ipfs provide once --enc=json streams one event per CID", func(t *testing.T) {
429 t.Parallel()
430
431 nodes := initNodes(t, 1, func(n *harness.Node) {
432 n.SetIPFSConfig("Provide.Enabled", true)
433 n.SetIPFSConfig("Provide.Strategy", "roots")
434 })
435 defer nodes.StopDaemons()
436
437 publisher := nodes[0]
438 c1 := publisher.IPFSAddStr(uniq("json 1"), "--pin=false")
439 c2 := publisher.IPFSAddStr(uniq("json 2"), "--pin=false")
440
441 res := publisher.RunIPFS("provide", "once", "--enc=json", c1, c2)
442 assert.Equal(t, 0, res.ExitCode(), "provide once --enc=json should succeed")
443
444 // Parse one JSON object per non-empty line.
445 var queued []string
446 for line := range strings.Lines(res.Stdout.String()) {
447 line = strings.TrimSpace(line)
448 if line == "" {
449 continue
450 }
451 var ev struct{ Queued string }
452 require.NoError(t, json.Unmarshal([]byte(line), &ev), "each line must parse as JSON: %q", line)
453 queued = append(queued, ev.Queued)
454 }
455 assert.ElementsMatch(t, []string{c1, c2}, queued)
456 })
457
458 t.Run("Provide.DHT.Interval=0 keeps announcing new CIDs (fast-provide-root)", func(t *testing.T) {
459 t.Parallel()
460
461 nodes := initNodes(t, 2, func(n *harness.Node) {
462 // Required: Interval=0 alone is rejected by the validator
463 // since the new semantic only disables the schedule.
464 n.SetIPFSConfig("Provide.Enabled", true)
465 n.SetIPFSConfig("Provide.DHT.Interval", "0")
466 })
467 defer nodes.StopDaemons()
468
469 cid := nodes[0].IPFSAddStr(time.Now().String())
470 expectProviders(t, cid, nodes[0].PeerID().String(), nodes[1:]...)
471 })
472
473 // `routing reprovide` is only available with the legacy provider.
474 // Sweep provider reprovides automatically on schedule.
475 if !sweep {
476 t.Run("Manual Reprovide trigger does not work when periodic reprovide is disabled", func(t *testing.T) {
477 t.Parallel()
478
479 nodes := initNodes(t, 2, func(n *harness.Node) {
480 n.SetIPFSConfig("Provide.Enabled", true)
481 n.SetIPFSConfig("Provide.DHT.Interval", "0")
482 })
483 defer nodes.StopDaemons()
484
485 res := nodes[0].RunIPFS("routing", "reprovide")
486 assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.DHT.Interval is set to '0'")
487 assert.Equal(t, 1, res.ExitCode())
488 })
489
490 t.Run("Manual Reprovide trigger does not work when Provide system is disabled", func(t *testing.T) {
491 t.Parallel()
492
493 nodes := initNodes(t, 2, func(n *harness.Node) {
494 n.SetIPFSConfig("Provide.Enabled", false)
495 })
496 defer nodes.StopDaemons()
497
498 cid := nodes[0].IPFSAddStr(time.Now().String())
499
500 expectNoProviders(t, cid, nodes[1:]...)
501
502 res := nodes[0].RunIPFS("routing", "reprovide")
503 assert.Contains(t, res.Stderr.Trimmed(), "invalid configuration: Provide.Enabled is set to 'false'")
504 assert.Equal(t, 1, res.ExitCode())
505
506 expectNoProviders(t, cid, nodes[1:]...)
507 })
508 }
509
510 t.Run("Provide with 'all' strategy", func(t *testing.T) {
511 t.Parallel()
512
513 nodes := initNodes(t, 2, func(n *harness.Node) {
514 n.SetIPFSConfig("Provide.Strategy", "all")
515 })
516 defer nodes.StopDaemons()
517 publisher := nodes[0]
518
519 cid := publisher.IPFSAddStr(uniq("all strategy"))
520 expectProviders(t, cid, publisher.PeerID().String(), nodes[1:]...)
521 })
522
523 t.Run("Provide with 'pinned' strategy", func(t *testing.T) {
524 t.Parallel()
525
526 nodes := initNodes(t, 2, func(n *harness.Node) {
527 n.SetIPFSConfig("Provide.Strategy", "pinned")
528 })
529 defer nodes.StopDaemons()
530 publisher := nodes[0]
531
532 // Add a non-pinned CID (should not be provided)
533 cid := publisher.IPFSAddStr(uniq("pinned strategy"), "--pin=false")
534 expectNoProviders(t, cid, nodes[1:]...)
535
536 // Pin the CID (should now be provided)
537 publisher.IPFS("pin", "add", cid)
538 expectProviders(t, cid, publisher.PeerID().String(), nodes[1:]...)
539 })
540
541 t.Run("Provide with 'pinned+mfs' strategy", func(t *testing.T) {
542 t.Parallel()
543
544 nodes := initNodes(t, 2, func(n *harness.Node) {
545 n.SetIPFSConfig("Provide.Strategy", "pinned+mfs")
546 })
547 defer nodes.StopDaemons()
548 publisher := nodes[0]
549
550 cidPinned := publisher.IPFSAddStr(uniq("pinned content"))
551 cidUnpinned := publisher.IPFSAddStr(uniq("unpinned content"), "--pin=false")
552 cidMFS := publisher.IPFSAddStr(uniq("mfs content"), "--pin=false")
553 publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
554
555 expectProviders(t, cidPinned, publisher.PeerID().String(), nodes[1:]...)
556 expectNoProviders(t, cidUnpinned, nodes[1:]...)
557 expectProviders(t, cidMFS, publisher.PeerID().String(), nodes[1:]...)
558 })
559
560 // addLargeFileInSubdir adds a 2 MiB file inside /subdir/ in MFS and
561 // returns the MFS root CID, the file root CID, and a chunk CID.
562 // The file is large enough to be split into multiple blocks.
563 // The resulting DAG: root-dir/subdir/largefile (2+ chunks).
564 addLargeFileInSubdir := func(t *testing.T, publisher *harness.Node) (cidRoot, cidSubdir, cidFile, cidChunk string) {
565 t.Helper()
566 largeData := random.Bytes(2 * 1024 * 1024) // 2 MiB = 2 chunks at 1 MiB
567
568 // Add file without pinning, then build directory structure in MFS
569 cidFile = publisher.IPFSAdd(bytes.NewReader(largeData), "-Q", "--pin=false")
570 publisher.IPFS("files", "mkdir", "-p", "/subdir")
571 publisher.IPFS("files", "cp", "/ipfs/"+cidFile, "/subdir/largefile")
572
573 // Get CIDs for the directory structure
574 cidRoot = publisher.IPFS("files", "stat", "--hash", "/").Stdout.Trimmed()
575 cidSubdir = publisher.IPFS("files", "stat", "--hash", "/subdir").Stdout.Trimmed()
576
577 // Get a chunk CID from the file's DAG links
578 dagOut := publisher.IPFS("dag", "get", cidFile)
579 var dagNode struct {
580 Links []struct {
581 Hash map[string]string `json:"Hash"`
582 } `json:"Links"`
583 }
584 require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
585 require.Greater(t, len(dagNode.Links), 1, "file should have multiple chunks")
586 cidChunk = dagNode.Links[0].Hash["/"]
587 require.NotEmpty(t, cidChunk)
588
589 return cidRoot, cidSubdir, cidFile, cidChunk
590 }
591
592 // +unique and +entities tests verify which CIDs end up in the DHT
593 // (strategy scope). Bloom filter deduplication correctness and
594 // entity type detection are tested in boxo/dag/walker/*_test.go.
595
596 t.Run("Provide with 'pinned+mfs+unique' strategy", func(t *testing.T) {
597 t.Parallel()
598
599 nodes := initNodes(t, 2, func(n *harness.Node) {
600 n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+unique")
601 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
602 })
603 defer nodes.StopDaemons()
604 publisher, peers := nodes[0], nodes[1:]
605
606 // +unique provides all blocks in pinned DAGs (same scope as
607 // pinned+mfs but with bloom filter dedup across pins).
608 // Use --fast-provide-dag and --fast-provide-wait on pin add
609 // so we can verify which blocks the strategy includes.
610 cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
611 publisher.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidRoot)
612 cidUnpinned := publisher.IPFSAddStr(uniq("unpinned content"), "--pin=false")
613
614 pid := publisher.PeerID().String()
615 // All blocks in the pinned DAG should be provided (including chunks)
616 expectProviders(t, cidRoot, pid, peers...)
617 expectProviders(t, cidSubdir, pid, peers...)
618 expectProviders(t, cidFile, pid, peers...)
619 expectProviders(t, cidChunk, pid, peers...)
620 expectNoProviders(t, cidUnpinned, peers...)
621 })
622
623 t.Run("Provide with 'pinned+mfs+entities' strategy", func(t *testing.T) {
624 t.Parallel()
625
626 nodes := initNodes(t, 2, func(n *harness.Node) {
627 n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+entities")
628 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
629 })
630 defer nodes.StopDaemons()
631 publisher, peers := nodes[0], nodes[1:]
632
633 // +entities provides only entity roots (files, directories,
634 // HAMT shards) and skips internal file chunks.
635 // Use --fast-provide-dag and --fast-provide-wait on pin add
636 // so we can verify which blocks the strategy skips.
637 cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
638 publisher.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidRoot)
639
640 pid := publisher.PeerID().String()
641 // Entity roots: directories and file root
642 expectProviders(t, cidRoot, pid, peers...)
643 expectProviders(t, cidSubdir, pid, peers...)
644 expectProviders(t, cidFile, pid, peers...)
645 // Internal chunk should NOT be provided (+entities skips chunks)
646 expectNoProviders(t, cidChunk, peers...)
647 })
648
649 t.Run("ipfs add --fast-provide-dag honors +entities (no chunk providing)", func(t *testing.T) {
650 t.Parallel()
651
652 nodes := initNodes(t, 2, func(n *harness.Node) {
653 n.SetIPFSConfig("Provide.Strategy", "pinned+entities")
654 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
655 })
656 defer nodes.StopDaemons()
657 publisher, peers := nodes[0], nodes[1:]
658
659 // Regression test for the providingDagService double-providing
660 // path. Before the fix, ipfs add --pin --fast-provide-dag wrapped
661 // the DAGService with providingDagService, which announced every
662 // block as it was written -- including chunks -- regardless of
663 // the +entities modifier. The post-add ExecuteFastProvideDAG
664 // walk then ran in parallel, so chunks ended up in the DHT
665 // despite +entities saying they should be skipped.
666 //
667 // After the fix, ExecuteFastProvideDAG is the single mechanism
668 // for --fast-provide-dag and respects the active strategy.
669 largeData := random.Bytes(2 * 1024 * 1024) // 2 MiB = 2 chunks
670 cidFile := publisher.IPFSAdd(bytes.NewReader(largeData),
671 "--fast-provide-dag", "--fast-provide-wait")
672
673 // Get a chunk CID from the file's DAG links
674 dagOut := publisher.IPFS("dag", "get", cidFile)
675 var dagNode struct {
676 Links []struct {
677 Hash map[string]string `json:"Hash"`
678 } `json:"Links"`
679 }
680 require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
681 require.Greater(t, len(dagNode.Links), 1, "file should have multiple chunks")
682 cidChunk := dagNode.Links[0].Hash["/"]
683 require.NotEmpty(t, cidChunk)
684
685 pid := publisher.PeerID().String()
686 // File root (entity) should be provided
687 expectProviders(t, cidFile, pid, peers...)
688 // Chunk should NOT be provided (+entities skips chunks)
689 expectNoProviders(t, cidChunk, peers...)
690 })
691
692 // addLargeFilestoreFile writes a 2 MiB file to the publisher's
693 // node directory and adds it via --nocopy, returning the root CID
694 // and a chunk CID from the file's DAG links. With the configured
695 // 1 MiB chunker the file produces multiple leaf blocks so we can
696 // distinguish root-level from chunk-level provide behavior.
697 addLargeFilestoreFile := func(t *testing.T, publisher *harness.Node, addArgs ...string) (cidRoot, cidChunk string) {
698 t.Helper()
699 filePath := filepath.Join(publisher.Dir, "filestore-"+strconv.FormatInt(time.Now().UnixNano(), 10)+".bin")
700 require.NoError(t, os.WriteFile(filePath, random.Bytes(2*1024*1024), 0o644))
701
702 args := append([]string{"add", "-q", "--nocopy"}, addArgs...)
703 args = append(args, filePath)
704 cidRoot = strings.TrimSpace(publisher.IPFS(args...).Stdout.String())
705
706 dagOut := publisher.IPFS("dag", "get", cidRoot)
707 var dagNode struct {
708 Links []struct {
709 Hash map[string]string `json:"Hash"`
710 } `json:"Links"`
711 }
712 require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
713 require.Greater(t, len(dagNode.Links), 1, "filestore file should have multiple chunks")
714 cidChunk = dagNode.Links[0].Hash["/"]
715 require.NotEmpty(t, cidChunk)
716 return cidRoot, cidChunk
717 }
718
719 t.Run("Filestore --nocopy with 'all' strategy provides every block", func(t *testing.T) {
720 t.Parallel()
721
722 nodes := initNodes(t, 2, func(n *harness.Node) {
723 n.SetIPFSConfig("Experimental.FilestoreEnabled", true)
724 n.SetIPFSConfig("Provide.Strategy", "all")
725 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
726 })
727 defer nodes.StopDaemons()
728 publisher, peers := nodes[0], nodes[1:]
729
730 // Positive control: with the default 'all' strategy the
731 // filestore Put path provides every block as it is written,
732 // including non-root chunks.
733 cidRoot, cidChunk := addLargeFilestoreFile(t, publisher)
734
735 pid := publisher.PeerID().String()
736 expectProviders(t, cidRoot, pid, peers...)
737 expectProviders(t, cidChunk, pid, peers...)
738 })
739
740 t.Run("Filestore --nocopy with selective strategy skips write-time provide", func(t *testing.T) {
741 t.Parallel()
742
743 nodes := initNodes(t, 2, func(n *harness.Node) {
744 n.SetIPFSConfig("Experimental.FilestoreEnabled", true)
745 n.SetIPFSConfig("Provide.Strategy", "pinned")
746 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
747 })
748 defer nodes.StopDaemons()
749 publisher, peers := nodes[0], nodes[1:]
750
751 // With a selective strategy the filestore must not eagerly
752 // announce blocks at write time. --pin=false skips the pin
753 // (so fast-provide-root has nothing to do) and
754 // --fast-provide-root=false disables it explicitly, isolating
755 // the assertion to the filestore's internal provide path.
756 cidRoot, cidChunk := addLargeFilestoreFile(t, publisher,
757 "--pin=false", "--fast-provide-root=false")
758
759 expectNoProviders(t, cidRoot, peers...)
760 expectNoProviders(t, cidChunk, peers...)
761 })
762
763 t.Run("Filestore --nocopy + selective strategy + --fast-provide-dag walks DAG", func(t *testing.T) {
764 t.Parallel()
765
766 nodes := initNodes(t, 2, func(n *harness.Node) {
767 n.SetIPFSConfig("Experimental.FilestoreEnabled", true)
768 n.SetIPFSConfig("Provide.Strategy", "pinned")
769 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
770 })
771 defer nodes.StopDaemons()
772 publisher, peers := nodes[0], nodes[1:]
773
774 // The selective-strategy gate skips the filestore's write-time
775 // provide, but the post-add ExecuteFastProvideDAG walk reads
776 // blocks through the wrapping blockstore (which transparently
777 // serves filestore-backed content) and announces each block,
778 // honoring the active strategy. This is the integration test
779 // behind the changelog claim that filestore content now plays
780 // well with the fast-provide-dag flag.
781 cidRoot, cidChunk := addLargeFilestoreFile(t, publisher,
782 "--fast-provide-dag", "--fast-provide-wait")
783
784 pid := publisher.PeerID().String()
785 expectProviders(t, cidRoot, pid, peers...)
786 expectProviders(t, cidChunk, pid, peers...)
787 })
788
789 t.Run("Provide with 'roots' strategy", func(t *testing.T) {
790 t.Parallel()
791
792 nodes := initNodes(t, 2, func(n *harness.Node) {
793 n.SetIPFSConfig("Provide.Strategy", "roots")
794 })
795 defer nodes.StopDaemons()
796 publisher := nodes[0]
797
798 // Add with -w: the wrapper directory is the recursive pin root,
799 // the file inside is a child block of that pin (not a root).
800 // Use --only-hash first to learn the child CID without providing.
801 data := random.Bytes(1000)
802 cidChild := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "--only-hash")
803 cidRoot := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "-w")
804
805 // 'roots' strategy provides only pin roots, not child blocks.
806 expectProviders(t, cidRoot, publisher.PeerID().String(), nodes[1:]...)
807 expectNoProviders(t, cidChild, nodes[1:]...)
808 })
809
810 t.Run("Provide with 'mfs' strategy", func(t *testing.T) {
811 t.Parallel()
812
813 nodes := initNodes(t, 2, func(n *harness.Node) {
814 n.SetIPFSConfig("Provide.Strategy", "mfs")
815 })
816 defer nodes.StopDaemons()
817 publisher := nodes[0]
818
819 // 'mfs' only provides content in MFS. Pinned content outside
820 // MFS should NOT be provided (mfs excludes pinned by default).
821 cidPinned := publisher.IPFSAddStr(uniq("pinned but not mfs"))
822 expectNoProviders(t, cidPinned, nodes[1:]...)
823
824 // Add to MFS (should be provided)
825 data := random.Bytes(1000)
826 cidMFS := publisher.IPFSAdd(bytes.NewReader(data), "-Q", "--pin=false")
827 publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
828 expectProviders(t, cidMFS, publisher.PeerID().String(), nodes[1:]...)
829
830 // Pinned CID still not provided (mfs strategy ignores pins)
831 expectNoProviders(t, cidPinned, nodes[1:]...)
832 })
833
834 // Reprovide tests: add content offline, start daemon, wait for reprovide.
835 //
836 // Each test waits for TWO reprovide cycles to confirm the schedule
837 // works repeatedly, not just on the initial bootstrap. The second
838 // cycle also catches bugs where state isn't persisted across cycles.
839 //
840 // Legacy: `routing reprovide` blocks until the reprovide cycle finishes,
841 // so we call it and check results immediately after.
842 //
843 // Sweep: no manual trigger exists. Instead, we set
844 // Provide.DHT.Interval=30s on the importing node and poll
845 // `provide stat` until the cycle completes.
846
847 // verifyReprovide waits for two reprovide cycles and asserts which
848 // CIDs are/aren't findable after each. minCIDs is the expected
849 // number of provided CIDs per cycle.
850 verifyReprovide := func(
851 t *testing.T,
852 publisher *harness.Node,
853 queriers harness.Nodes,
854 minCIDs int64,
855 provided []string,
856 notProvided []string,
857 ) {
858 t.Helper()
859 pid := publisher.PeerID().String()
860 check := func() {
861 for _, c := range provided {
862 expectProviders(t, c, pid, queriers...)
863 }
864 for _, c := range notProvided {
865 expectNoProviders(t, c, queriers...)
866 }
867 }
868
869 after1 := awaitReprovide(t, publisher, minCIDs)
870 check()
871 // Second cycle: confirms the schedule runs repeatedly.
872 awaitReprovide(t, publisher, after1+minCIDs)
873 check()
874 }
875
876 {
877
878 t.Run("Reprovides with 'all' strategy when strategy is '' (empty)", func(t *testing.T) {
879 t.Parallel()
880
881 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
882 n.SetIPFSConfig("Provide.Strategy", "")
883 })
884 publisher := nodes[0]
885 if sweep {
886 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
887 }
888
889 cid := publisher.IPFSAddStr(time.Now().String())
890
891 nodes = nodes.StartDaemons().Connect()
892 defer nodes.StopDaemons()
893 peers := nodes[1:]
894
895 verifyReprovide(t, publisher, peers, 1, // 1 block added
896 []string{cid}, nil)
897 })
898
899 t.Run("Reprovides with 'all' strategy", func(t *testing.T) {
900 t.Parallel()
901
902 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
903 n.SetIPFSConfig("Provide.Strategy", "all")
904 })
905 publisher := nodes[0]
906 if sweep {
907 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
908 }
909
910 cid := publisher.IPFSAddStr(time.Now().String())
911
912 nodes = nodes.StartDaemons().Connect()
913 defer nodes.StopDaemons()
914 peers := nodes[1:]
915
916 verifyReprovide(t, publisher, peers, 1, // 1 block added
917 []string{cid}, nil)
918 })
919
920 t.Run("Reprovides with 'pinned' strategy", func(t *testing.T) {
921 t.Parallel()
922
923 foo := random.Bytes(1000)
924 bar := random.Bytes(1000)
925
926 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
927 n.SetIPFSConfig("Provide.Strategy", "pinned")
928 })
929 publisher := nodes[0]
930 if sweep {
931 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
932 }
933
934 // Add a pin while offline
935 cidBarDir := publisher.IPFSAdd(bytes.NewReader(bar), "-Q", "-w")
936
937 nodes = nodes.StartDaemons().Connect()
938 defer nodes.StopDaemons()
939 peers := nodes[1:]
940
941 // Add content without pinning while daemon is online
942 cidFoo := publisher.IPFSAdd(bytes.NewReader(foo), "--pin=false")
943 cidBar := publisher.IPFSAdd(bytes.NewReader(bar), "--pin=false")
944
945 verifyReprovide(t, publisher, peers, 2, // cidBar + cidBarDir (bar is child of the wrapped dir pin)
946 []string{cidBar, cidBarDir},
947 []string{cidFoo}) // cidFoo not pinned
948 })
949
950 t.Run("Reprovides with 'roots' strategy", func(t *testing.T) {
951 t.Parallel()
952
953 bar := random.Bytes(1000)
954
955 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
956 n.SetIPFSConfig("Provide.Strategy", "roots")
957 })
958 publisher := nodes[0]
959 if sweep {
960 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
961 }
962
963 // Compute the child CID without storing anything (safe
964 // offline, daemon not started yet).
965 cidChild := publisher.IPFSAdd(bytes.NewReader(bar), "-Q", "--only-hash")
966 // Add with -w: pins the wrapper directory as root. The file
967 // inside is a child block of that pin, not a root.
968 cidRoot := publisher.IPFSAdd(bytes.NewReader(bar), "-Q", "-w")
969
970 nodes = nodes.StartDaemons().Connect()
971 defer nodes.StopDaemons()
972 peers := nodes[1:]
973
974 verifyReprovide(t, publisher, peers, 1, // cidRoot (only pin root)
975 []string{cidRoot},
976 []string{cidChild}) // child of pin, not a root
977 })
978
979 t.Run("Reprovides with 'mfs' strategy", func(t *testing.T) {
980 t.Parallel()
981
982 bar := random.Bytes(1000)
983
984 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
985 n.SetIPFSConfig("Provide.Strategy", "mfs")
986 })
987 publisher := nodes[0]
988 if sweep {
989 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
990 }
991
992 // Add to MFS (should be provided)
993 cidMFS := publisher.IPFSAdd(bytes.NewReader(bar), "--pin=false", "-Q")
994 publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
995 // Pin something NOT in MFS (should NOT be provided)
996 cidPinned := publisher.IPFSAddStr(uniq("pinned but not mfs"))
997
998 nodes = nodes.StartDaemons().Connect()
999 defer nodes.StopDaemons()
1000 peers := nodes[1:]
1001
1002 verifyReprovide(t, publisher, peers, 1, // cidMFS only
1003 []string{cidMFS},
1004 []string{cidPinned}) // mfs strategy ignores pinned content outside MFS
1005 })
1006
1007 t.Run("Reprovides with 'pinned+mfs' strategy", func(t *testing.T) {
1008 t.Parallel()
1009
1010 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
1011 n.SetIPFSConfig("Provide.Strategy", "pinned+mfs")
1012 })
1013 publisher := nodes[0]
1014 if sweep {
1015 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
1016 }
1017
1018 // Add a pinned CID (should be provided)
1019 cidPinned := publisher.IPFSAddStr(uniq("pinned content"), "--pin=true")
1020 // Add a CID to MFS (should be provided)
1021 cidMFS := publisher.IPFSAddStr(uniq("mfs content"))
1022 publisher.IPFS("files", "cp", "/ipfs/"+cidMFS, "/myfile")
1023 // Add a CID that is neither pinned nor in MFS (should not be provided)
1024 cidNeither := publisher.IPFSAddStr(uniq("neither content"), "--pin=false")
1025
1026 nodes = nodes.StartDaemons().Connect()
1027 defer nodes.StopDaemons()
1028 peers := nodes[1:]
1029
1030 verifyReprovide(t, publisher, peers, 2, // cidPinned + cidMFS
1031 []string{cidPinned, cidMFS},
1032 []string{cidNeither}) // neither pinned nor in MFS
1033 })
1034
1035 t.Run("Reprovides with 'pinned+mfs+unique' strategy", func(t *testing.T) {
1036 t.Parallel()
1037
1038 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
1039 n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+unique")
1040 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
1041 })
1042 publisher := nodes[0]
1043 if sweep {
1044 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
1045 }
1046
1047 // Build a directory DAG with a multi-chunk file in MFS, then pin it.
1048 cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
1049 publisher.IPFS("pin", "add", cidRoot)
1050 cidUnpinned := publisher.IPFSAddStr(uniq("unpinned content"), "--pin=false")
1051
1052 nodes = nodes.StartDaemons().Connect()
1053 defer nodes.StopDaemons()
1054 peers := nodes[1:]
1055
1056 // +unique provides all blocks in pinned DAGs (same as pinned+mfs)
1057 verifyReprovide(t, publisher, peers, 4, // root + subdir + file + chunks
1058 []string{cidRoot, cidSubdir, cidFile, cidChunk},
1059 []string{cidUnpinned})
1060 })
1061
1062 t.Run("Reprovides with 'pinned+mfs+entities' strategy", func(t *testing.T) {
1063 t.Parallel()
1064
1065 nodes := initNodesWithoutStart(t, 2, func(n *harness.Node) {
1066 n.SetIPFSConfig("Provide.Strategy", "pinned+mfs+entities")
1067 n.SetIPFSConfig("Import.UnixFSChunker", "size-1048576") // 1 MiB chunks
1068 })
1069 publisher := nodes[0]
1070 if sweep {
1071 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
1072 }
1073
1074 // Build a directory DAG with a multi-chunk file in MFS, then pin it.
1075 cidRoot, cidSubdir, cidFile, cidChunk := addLargeFileInSubdir(t, publisher)
1076 publisher.IPFS("pin", "add", cidRoot)
1077
1078 nodes = nodes.StartDaemons().Connect()
1079 defer nodes.StopDaemons()
1080 peers := nodes[1:]
1081
1082 // Entity roots: directories and file root (not chunks)
1083 verifyReprovide(t, publisher, peers, 3, // root + subdir + file (not chunks)
1084 []string{cidRoot, cidSubdir, cidFile},
1085 []string{cidChunk}) // chunks skipped by +entities
1086 })
1087 }
1088
1089 t.Run("provide clear command removes items from provide queue", func(t *testing.T) {
1090 t.Parallel()
1091
1092 nodes := harness.NewT(t).NewNodes(1).Init()
1093 nodes.ForEachPar(func(n *harness.Node) {
1094 n.SetIPFSConfig("Provide.Enabled", true)
1095 n.SetIPFSConfig("Provide.DHT.Interval", "22h")
1096 n.SetIPFSConfig("Provide.Strategy", "all")
1097 })
1098 nodes.StartDaemons()
1099 defer nodes.StopDaemons()
1100
1101 // Clear the provide queue first time - works regardless of queue state
1102 res1 := nodes[0].IPFS("provide", "clear")
1103 require.NoError(t, res1.Err)
1104
1105 // Should report cleared items and proper message format
1106 assert.Contains(t, res1.Stdout.String(), "removed")
1107 assert.Contains(t, res1.Stdout.String(), "items from provide queue")
1108
1109 // Clear the provide queue second time - should definitely report 0 items
1110 res2 := nodes[0].IPFS("provide", "clear")
1111 require.NoError(t, res2.Err)
1112
1113 // Should report 0 items cleared since queue was already cleared
1114 assert.Contains(t, res2.Stdout.String(), "removed 0 items from provide queue")
1115 })
1116
1117 t.Run("provide clear command with quiet option", func(t *testing.T) {
1118 t.Parallel()
1119
1120 nodes := harness.NewT(t).NewNodes(1).Init()
1121 nodes.ForEachPar(func(n *harness.Node) {
1122 n.SetIPFSConfig("Provide.Enabled", true)
1123 n.SetIPFSConfig("Provide.DHT.Interval", "22h")
1124 n.SetIPFSConfig("Provide.Strategy", "all")
1125 })
1126 nodes.StartDaemons()
1127 defer nodes.StopDaemons()
1128
1129 // Clear the provide queue with quiet option
1130 res := nodes[0].IPFS("provide", "clear", "-q")
1131 require.NoError(t, res.Err)
1132
1133 // Should have no output when quiet
1134 assert.Empty(t, res.Stdout.String())
1135 })
1136
1137 t.Run("provide clear command works when provider is disabled", func(t *testing.T) {
1138 t.Parallel()
1139
1140 nodes := harness.NewT(t).NewNodes(1).Init()
1141 nodes.ForEachPar(func(n *harness.Node) {
1142 n.SetIPFSConfig("Provide.Enabled", false)
1143 n.SetIPFSConfig("Provide.DHT.Interval", "22h")
1144 n.SetIPFSConfig("Provide.Strategy", "all")
1145 })
1146 nodes.StartDaemons()
1147 defer nodes.StopDaemons()
1148
1149 // Clear should succeed even when provider is disabled
1150 res := nodes[0].IPFS("provide", "clear")
1151 require.NoError(t, res.Err)
1152 })
1153
1154 t.Run("provide clear command returns JSON with removed item count", func(t *testing.T) {
1155 t.Parallel()
1156
1157 nodes := harness.NewT(t).NewNodes(1).Init()
1158 nodes.ForEachPar(func(n *harness.Node) {
1159 n.SetIPFSConfig("Provide.Enabled", true)
1160 n.SetIPFSConfig("Provide.DHT.Interval", "22h")
1161 n.SetIPFSConfig("Provide.Strategy", "all")
1162 })
1163 nodes.StartDaemons()
1164 defer nodes.StopDaemons()
1165
1166 // Clear the provide queue with JSON encoding
1167 res := nodes[0].IPFS("provide", "clear", "--enc=json")
1168 require.NoError(t, res.Err)
1169
1170 // Should return valid JSON with the number of removed items
1171 output := res.Stdout.String()
1172 assert.NotEmpty(t, output)
1173
1174 // Parse JSON to verify structure
1175 var result int
1176 err := json.Unmarshal([]byte(output), &result)
1177 require.NoError(t, err, "Output should be valid JSON")
1178
1179 // Should be a non-negative integer (0 or positive)
1180 assert.GreaterOrEqual(t, result, 0)
1181 })
1182 }
1183
1184 // runResumeTests validates Provide.DHT.ResumeEnabled behavior for SweepingProvider.
1185 //
1186 // Background: The provider tracks current_time_offset = (now - cycleStart) % interval
1187 // where cycleStart is the timestamp marking the beginning of the reprovide cycle.
1188 // With ResumeEnabled=true, cycleStart persists in the datastore across restarts.
1189 // With ResumeEnabled=false, cycleStart resets to 'now' on each startup.
1190 func runResumeTests(t *testing.T, apply cfgApplier) {
1191 t.Helper()
1192
1193 const (
1194 reprovideInterval = 30 * time.Second
1195 initialRuntime = 10 * time.Second // Let cycle progress
1196 downtime = 5 * time.Second // Simulated offline period
1197 restartTime = 2 * time.Second // Daemon restart stabilization
1198
1199 // Thresholds account for timing jitter (~2-3s margin)
1200 minOffsetBeforeRestart = 8 * time.Second // Expect ~10s
1201 minOffsetAfterResume = 12 * time.Second // Expect ~17s (10s + 5s + 2s)
1202 maxOffsetAfterReset = 5 * time.Second // Expect ~2s (fresh start)
1203 )
1204
1205 setupNode := func(t *testing.T, resumeEnabled bool) *harness.Node {
1206 node := harness.NewT(t).NewNode().Init()
1207 apply(node) // Sets Provide.DHT.SweepEnabled=true
1208 node.SetIPFSConfig("Provide.DHT.ResumeEnabled", resumeEnabled)
1209 node.SetIPFSConfig("Provide.DHT.Interval", reprovideInterval.String())
1210 node.SetIPFSConfig("Bootstrap", []string{})
1211 node.StartDaemon()
1212 return node
1213 }
1214
1215 t.Run("preserves cycle state across restart", func(t *testing.T) {
1216 t.Parallel()
1217
1218 node := setupNode(t, true)
1219 defer node.StopDaemon()
1220
1221 for i := range 10 {
1222 node.IPFSAddStr(fmt.Sprintf("resume-test-%d-%d", i, time.Now().UnixNano()))
1223 }
1224
1225 time.Sleep(initialRuntime)
1226
1227 beforeRestart := node.IPFS("provide", "stat", "--enc=json")
1228 offsetBeforeRestart, _, err := parseProvideStatJSON(beforeRestart.Stdout.String())
1229 require.NoError(t, err)
1230 require.Greater(t, offsetBeforeRestart, minOffsetBeforeRestart,
1231 "cycle should have progressed")
1232
1233 node.StopDaemon()
1234 time.Sleep(downtime)
1235 node.StartDaemon()
1236 time.Sleep(restartTime)
1237
1238 afterRestart := node.IPFS("provide", "stat", "--enc=json")
1239 offsetAfterRestart, _, err := parseProvideStatJSON(afterRestart.Stdout.String())
1240 require.NoError(t, err)
1241
1242 assert.GreaterOrEqual(t, offsetAfterRestart, minOffsetAfterResume,
1243 "offset should account for downtime")
1244 })
1245
1246 t.Run("resets cycle when disabled", func(t *testing.T) {
1247 t.Parallel()
1248
1249 node := setupNode(t, false)
1250 defer node.StopDaemon()
1251
1252 for i := range 10 {
1253 node.IPFSAddStr(fmt.Sprintf("no-resume-%d-%d", i, time.Now().UnixNano()))
1254 }
1255
1256 time.Sleep(initialRuntime)
1257
1258 beforeRestart := node.IPFS("provide", "stat", "--enc=json")
1259 offsetBeforeRestart, _, err := parseProvideStatJSON(beforeRestart.Stdout.String())
1260 require.NoError(t, err)
1261 require.Greater(t, offsetBeforeRestart, minOffsetBeforeRestart,
1262 "cycle should have progressed")
1263
1264 node.StopDaemon()
1265 time.Sleep(downtime)
1266 node.StartDaemon()
1267 time.Sleep(restartTime)
1268
1269 afterRestart := node.IPFS("provide", "stat", "--enc=json")
1270 offsetAfterRestart, _, err := parseProvideStatJSON(afterRestart.Stdout.String())
1271 require.NoError(t, err)
1272
1273 assert.Less(t, offsetAfterRestart, maxOffsetAfterReset,
1274 "offset should reset to near zero")
1275 })
1276 }
1277
1278 type provideStatJSON struct {
1279 Sweep struct {
1280 Timing struct {
1281 CurrentTimeOffset int64 `json:"current_time_offset"` // nanoseconds
1282 } `json:"timing"`
1283 Schedule struct {
1284 NextReprovidePrefix string `json:"next_reprovide_prefix"`
1285 } `json:"schedule"`
1286 Operations struct {
1287 Ongoing struct {
1288 KeyReprovides int `json:"key_reprovides"`
1289 } `json:"ongoing"`
1290 Past struct {
1291 KeysProvided int64 `json:"keys_provided"`
1292 } `json:"past"`
1293 } `json:"operations"`
1294 Queues struct {
1295 PendingKeyProvides int64 `json:"pending_key_provides"`
1296 } `json:"queues"`
1297 } `json:"Sweep"`
1298 }
1299
1300 // parseProvideStatJSON extracts timing and schedule information from
1301 // the JSON output of 'ipfs provide stat --enc=json'.
1302 func parseProvideStatJSON(output string) (offset time.Duration, prefix string, err error) {
1303 var stat provideStatJSON
1304 if err := json.Unmarshal([]byte(output), &stat); err != nil {
1305 return 0, "", err
1306 }
1307 offset = time.Duration(stat.Sweep.Timing.CurrentTimeOffset)
1308 prefix = stat.Sweep.Schedule.NextReprovidePrefix
1309 return offset, prefix, nil
1310 }
1311
1312 // waitForSweepReprovide polls `provide stat --enc=json` until the
1313 // sweep provider has provided at least minCIDs and no work is pending.
1314 // Pass 0 for minCIDs to just wait for any provide activity to finish.
1315 // Returns the total CIDs provided so far (for use as minCIDs in a
1316 // subsequent call to wait for the next cycle).
1317 // The importing node must have a short Provide.DHT.Interval so the
1318 // reprovide cycle completes within the timeout.
1319 func waitForSweepReprovide(t *testing.T, n *harness.Node, timeout time.Duration, minCIDs int64) int64 {
1320 t.Helper()
1321 if minCIDs == 0 {
1322 minCIDs = 1
1323 }
1324 deadline := time.Now().Add(timeout)
1325 for time.Now().Before(deadline) {
1326 res := n.RunIPFS("provide", "stat", "--enc=json")
1327 if res.ExitCode() == 0 {
1328 var stat provideStatJSON
1329 if err := json.Unmarshal(res.Stdout.Bytes(), &stat); err == nil {
1330 s := stat.Sweep
1331 if s.Operations.Past.KeysProvided >= minCIDs &&
1332 s.Queues.PendingKeyProvides == 0 &&
1333 s.Operations.Ongoing.KeyReprovides == 0 {
1334 return s.Operations.Past.KeysProvided
1335 }
1336 }
1337 }
1338 time.Sleep(500 * time.Millisecond)
1339 }
1340 t.Fatalf("sweep reprovide: expected at least %d CIDs provided within %s", minCIDs, timeout)
1341 return 0
1342 }
1343
1344 func TestProvider(t *testing.T) {
1345 t.Parallel()
1346
1347 variants := []struct {
1348 name string
1349 sweep bool
1350 apply cfgApplier
1351 awaitReprovide awaitReprovideFunc
1352 }{
1353 {
1354 name: "LegacyProvider",
1355 sweep: false,
1356 apply: func(n *harness.Node) {
1357 n.SetIPFSConfig("Provide.DHT.SweepEnabled", false)
1358 },
1359 // `routing reprovide` blocks until the cycle finishes.
1360 // minCIDs is ignored (legacy has no stat counter).
1361 awaitReprovide: func(t *testing.T, n *harness.Node, minCIDs int64) int64 {
1362 n.IPFS("routing", "reprovide")
1363 return minCIDs
1364 },
1365 },
1366 {
1367 name: "SweepingProvider",
1368 sweep: true,
1369 apply: func(n *harness.Node) {
1370 n.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1371 },
1372 // No manual trigger exists for sweep. Poll `provide stat`
1373 // until the reprovide cycle completes.
1374 awaitReprovide: func(t *testing.T, n *harness.Node, minCIDs int64) int64 {
1375 // 90s accounts for provider bootstrap time (connecting
1376 // to ephemeral peers, measuring prefix length) before
1377 // the 30s reprovide cycle starts. On CI with parallel
1378 // tests, bootstrap can take 20-30s.
1379 return waitForSweepReprovide(t, n, 90*time.Second, minCIDs)
1380 },
1381 },
1382 }
1383
1384 for _, v := range variants {
1385 t.Run(v.name, func(t *testing.T) {
1386 // t.Parallel()
1387 runProviderSuite(t, v.sweep, v.apply, v.awaitReprovide)
1388
1389 // Resume tests only apply to SweepingProvider
1390 if v.sweep {
1391 runResumeTests(t, v.apply)
1392 }
1393 })
1394 }
1395 }
1396
1397 // TestProviderUniqueDedupLogging verifies that the +unique bloom filter
1398 // deduplication produces a "skippedBranches" log with a value > 0 when
1399 // two pins share content. Tests both the fast-provide-dag path (immediate
1400 // provide on pin add) and the reprovide cycle path.
1401 func TestProviderUniqueDedupLogging(t *testing.T) {
1402 t.Parallel()
1403
1404 // Shared data that both pins will reference. Two pins containing
1405 // the same file block give the bloom something to dedup.
1406 sharedData := random.Bytes(10 * 1024) // 10 KiB, single block
1407
1408 t.Run("fast-provide-dag dedup across pins in single call", func(t *testing.T) {
1409 t.Parallel()
1410
1411 h := harness.NewT(t)
1412 node := h.NewNode().Init()
1413 node.SetIPFSConfig("Provide.Strategy", "pinned+unique")
1414 node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1415 node.SetIPFSConfig("Import.UnixFSChunker", "size-5120") // 5 KiB chunks
1416 h.BootstrapWithStubDHT(harness.Nodes{node})
1417
1418 node.StartDaemonWithReq(harness.RunRequest{
1419 CmdOpts: []harness.CmdOpt{
1420 harness.RunWithEnv(map[string]string{
1421 // dagwalker: bloom creation log
1422 // core/commands/cmdenv: fast-provide-dag finished log
1423 "GOLOG_LOG_LEVEL": "error,dagwalker=info,core/commands/cmdenv=info",
1424 }),
1425 },
1426 }, "")
1427 defer node.StopDaemon()
1428
1429 // 10 KiB file with 5 KiB chunks = 1 file root + 2 chunks = 3 blocks.
1430 // Two dirs each containing the file under different names:
1431 // dirA/fileA → same 3 blocks
1432 // dirB/fileB → same 3 blocks
1433 // Pinning both in a single `pin add` shares one bloom tracker.
1434 // Walking dirA: dirA + file root + chunk1 + chunk2 = 4 provided.
1435 // Walking dirB: dirB + file root (bloom hit, skip subtree) = 1 provided, 1 skipped.
1436 // Total: 5 provided, 1 skipped branch (file root in dirB; its
1437 // 2 chunks are never visited because the parent was skipped).
1438 cidFile := node.IPFSAdd(bytes.NewReader(sharedData), "-Q", "--pin=false")
1439 node.IPFS("files", "mkdir", "-p", "/dirA")
1440 node.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirA/fileA")
1441 cidDirA := node.IPFS("files", "stat", "--hash", "/dirA").Stdout.Trimmed()
1442 node.IPFS("files", "mkdir", "-p", "/dirB")
1443 node.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirB/fileB")
1444 cidDirB := node.IPFS("files", "stat", "--hash", "/dirB").Stdout.Trimmed()
1445 require.NotEqual(t, cidDirA, cidDirB, "dirs must differ to test dedup")
1446 // Single pin add with both CIDs shares one bloom.
1447 node.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidDirA, cidDirB)
1448
1449 daemonLog := node.Daemon.Stderr.String()
1450 require.Contains(t, daemonLog, "bloom tracker created")
1451 require.NotContains(t, daemonLog, "bloom tracker autoscaled")
1452 require.Contains(t, daemonLog, `"providedCIDs": 5`)
1453 require.Contains(t, daemonLog, `"skippedBranches": 1`)
1454 })
1455
1456 t.Run("reprovide cycle dedup across pins", func(t *testing.T) {
1457 t.Parallel()
1458
1459 h := harness.NewT(t)
1460 nodes := h.NewNodes(2).Init()
1461 for _, n := range nodes {
1462 n.SetIPFSConfig("Provide.Strategy", "pinned+unique")
1463 n.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1464 n.SetIPFSConfig("Import.UnixFSChunker", "size-5120") // 5 KiB chunks
1465 }
1466 publisher := nodes[0]
1467 publisher.SetIPFSConfig("Provide.DHT.Interval", "30s")
1468 h.BootstrapWithStubDHT(nodes)
1469
1470 // Same file structure as fast-provide-dag test above.
1471 // The reprovide cycle walks all recursive pins:
1472 // pin dirA: dirA + file root + chunk1 + chunk2 = 4 provided
1473 // pin empty MFS root (always present): 1 provided
1474 // pin dirB: dirB + file root (bloom hit, skip subtree) = 1 provided, 1 skipped
1475 // Total: 6 provided, 1 skipped branch.
1476 cidFile := publisher.IPFSAdd(bytes.NewReader(sharedData), "-Q", "--pin=false")
1477 publisher.IPFS("files", "mkdir", "-p", "/dirA")
1478 publisher.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirA/fileA")
1479 cidDirA := publisher.IPFS("files", "stat", "--hash", "/dirA").Stdout.Trimmed()
1480 publisher.IPFS("pin", "add", cidDirA)
1481 publisher.IPFS("files", "mkdir", "-p", "/dirB")
1482 publisher.IPFS("files", "cp", "/ipfs/"+cidFile, "/dirB/fileB")
1483 cidDirB := publisher.IPFS("files", "stat", "--hash", "/dirB").Stdout.Trimmed()
1484 require.NotEqual(t, cidDirA, cidDirB, "dirs must differ to test dedup")
1485 publisher.IPFS("pin", "add", cidDirB)
1486
1487 nodes[0].StartDaemonWithReq(harness.RunRequest{
1488 CmdOpts: []harness.CmdOpt{
1489 harness.RunWithEnv(map[string]string{
1490 "GOLOG_LOG_LEVEL": "error,dagwalker=info,provider=info",
1491 }),
1492 },
1493 }, "")
1494 nodes[1].StartDaemon()
1495 defer nodes.StopDaemons()
1496 nodes.Connect()
1497
1498 waitForSweepReprovide(t, publisher, 90*time.Second, 6)
1499
1500 daemonLog := publisher.Daemon.Stderr.String()
1501 require.Contains(t, daemonLog, "bloom tracker created")
1502 require.NotContains(t, daemonLog, "bloom tracker autoscaled")
1503 require.Contains(t, daemonLog, `"providedCIDs": 6`)
1504 require.Contains(t, daemonLog, `"skippedBranches": 1`)
1505 })
1506 }
1507
1508 // TestProviderFastProvideDAGAsyncSurvives verifies that
1509 // --fast-provide-dag without --fast-provide-wait runs a background
1510 // DAG walk that outlives the command handler and publishes every
1511 // block of the newly added DAG to the routing system.
1512 //
1513 // The async walk runs in a goroutine parented on the IpfsNode
1514 // lifetime context (not req.Context), so it keeps running after
1515 // `ipfs add` returns and is only cancelled on daemon shutdown.
1516 //
1517 // Provide.DHT.Interval is set high so the scheduled reprovide
1518 // cycle cannot fire during the test window. That makes the async
1519 // walk the only path that can publish non-root block CIDs.
1520 func TestProviderFastProvideDAGAsyncSurvives(t *testing.T) {
1521 t.Parallel()
1522
1523 h := harness.NewT(t)
1524 nodes := h.NewNodes(2).Init()
1525 for _, n := range nodes {
1526 n.SetIPFSConfig("Provide.Strategy", "pinned")
1527 n.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1528 // Small chunks so a modest file produces many leaf blocks.
1529 n.SetIPFSConfig("Import.UnixFSChunker", "size-1024")
1530 }
1531 publisher, peers := nodes[0], nodes[1:]
1532 publisher.SetIPFSConfig("Provide.DHT.Interval", "1h")
1533 h.BootstrapWithStubDHT(nodes)
1534
1535 publisher.StartDaemonWithReq(harness.RunRequest{
1536 CmdOpts: []harness.CmdOpt{
1537 harness.RunWithEnv(map[string]string{
1538 "GOLOG_LOG_LEVEL": "error,core/commands/cmdenv=info",
1539 }),
1540 },
1541 }, "")
1542 nodes[1].StartDaemon()
1543 defer nodes.StopDaemons()
1544 nodes.Connect()
1545
1546 // 16 KiB + 1 KiB chunks yields a file root plus many leaf
1547 // blocks, so the providedCIDs count after the walk is
1548 // unambiguous.
1549 data := random.Bytes(16 * 1024)
1550 cidFile := publisher.IPFSAdd(bytes.NewReader(data), "-Q",
1551 "--pin=true",
1552 "--fast-provide-dag=true",
1553 // --fast-provide-wait deliberately omitted: the walk
1554 // runs in the background after `ipfs add` returns.
1555 )
1556
1557 // Pull a chunk CID out of the file DAG. Chunks are not pin
1558 // roots, so fast-provide-root does not touch them; only the
1559 // DAG walk can announce them.
1560 dagOut := publisher.IPFS("dag", "get", cidFile)
1561 var dagNode struct {
1562 Links []struct {
1563 Hash map[string]string `json:"Hash"`
1564 } `json:"Links"`
1565 }
1566 require.NoError(t, json.Unmarshal(dagOut.Stdout.Bytes(), &dagNode))
1567 require.Greater(t, len(dagNode.Links), 1, "file should have multiple chunks")
1568 cidChunk := dagNode.Links[0].Hash["/"]
1569 require.NotEmpty(t, cidChunk)
1570
1571 // The async walk logs "fast-provide-dag: finished" with a
1572 // providedCIDs count on completion. A full walk of this file
1573 // visits the root plus every leaf chunk, so the count is much
1574 // larger than 2.
1575 providedRe := regexp.MustCompile(`"providedCIDs": (\d+)`)
1576 var providedCount int
1577 require.Eventually(t, func() bool {
1578 m := providedRe.FindStringSubmatch(publisher.Daemon.Stderr.String())
1579 if len(m) != 2 {
1580 return false
1581 }
1582 n, err := strconv.Atoi(m[1])
1583 if err != nil {
1584 return false
1585 }
1586 providedCount = n
1587 return true
1588 }, 30*time.Second, 200*time.Millisecond, "async fast-provide-dag walk did not log 'finished'")
1589
1590 require.Greater(t, providedCount, 2,
1591 "providedCIDs=%d is too small for a full walk of the file DAG", providedCount)
1592
1593 // End-to-end: the peer can find the publisher as a provider
1594 // for a chunk CID, which only the async walk could have
1595 // announced within the test window.
1596 pid := publisher.PeerID().String()
1597 var found bool
1598 for _, peer := range peers {
1599 for i := time.Duration(0); i*timeStep < timeout; i++ {
1600 res := peer.IPFS("routing", "findprovs", "-n=1", cidChunk)
1601 if res.Stdout.Trimmed() == pid {
1602 found = true
1603 break
1604 }
1605 }
1606 }
1607 require.True(t, found, "chunk %s not announced by the async walk", cidChunk)
1608 }
1609
1610 // TestHTTPOnlyProviderWithSweepEnabled tests that provider records are correctly
1611 // sent to HTTP routers when Routing.Type="custom" with only HTTP routers configured,
1612 // even when Provide.DHT.SweepEnabled=true (the default since v0.39).
1613 //
1614 // This is a regression test for https://github.com/ipfs/kubo/issues/11089
1615 func TestHTTPOnlyProviderWithSweepEnabled(t *testing.T) {
1616 t.Parallel()
1617
1618 // Track provide requests received by the mock HTTP router
1619 var provideRequests atomic.Int32
1620 mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
1621 if (r.Method == http.MethodPut || r.Method == http.MethodPost) &&
1622 strings.HasPrefix(r.URL.Path, "/routing/v1/providers") {
1623 provideRequests.Add(1)
1624 w.WriteHeader(http.StatusOK)
1625 } else if strings.HasPrefix(r.URL.Path, "/routing/v1/providers") && r.Method == http.MethodGet {
1626 // Return empty providers for findprovs
1627 w.Header().Set("Content-Type", "application/x-ndjson")
1628 w.WriteHeader(http.StatusOK)
1629 } else {
1630 w.WriteHeader(http.StatusNotFound)
1631 }
1632 }))
1633 defer mockServer.Close()
1634
1635 h := harness.NewT(t)
1636 node := h.NewNode().Init()
1637
1638 // Explicitly set SweepEnabled=true (the default since v0.39, but be explicit for test clarity)
1639 node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1640 node.SetIPFSConfig("Provide.Enabled", true)
1641
1642 // Configure HTTP-only custom routing (no DHT) with explicit Routing.Type=custom
1643 routingConf := map[string]any{
1644 "Type": "custom", // Explicitly set Routing.Type=custom
1645 "Methods": map[string]any{
1646 "provide": map[string]any{"RouterName": "HTTPRouter"},
1647 "get-ipns": map[string]any{"RouterName": "HTTPRouter"},
1648 "put-ipns": map[string]any{"RouterName": "HTTPRouter"},
1649 "find-peers": map[string]any{"RouterName": "HTTPRouter"},
1650 "find-providers": map[string]any{"RouterName": "HTTPRouter"},
1651 },
1652 "Routers": map[string]any{
1653 "HTTPRouter": map[string]any{
1654 "Type": "http",
1655 "Parameters": map[string]any{
1656 "Endpoint": mockServer.URL,
1657 },
1658 },
1659 },
1660 }
1661 node.SetIPFSConfig("Routing", routingConf)
1662 node.StartDaemon()
1663 defer node.StopDaemon()
1664
1665 // Add content and manually provide it
1666 cid := node.IPFSAddStr(time.Now().String())
1667
1668 // Manual provide should succeed even without libp2p peers
1669 res := node.RunIPFS("routing", "provide", cid)
1670 // Check that the command succeeded (exit code 0) and no provide-related errors
1671 assert.Equal(t, 0, res.ExitCode(), "routing provide should succeed with HTTP-only routing and SweepEnabled=true")
1672 assert.NotContains(t, res.Stderr.String(), "cannot provide", "should not have provide errors")
1673
1674 // Verify HTTP router received at least one provide request
1675 assert.Greater(t, provideRequests.Load(), int32(0),
1676 "HTTP router should have received provide requests")
1677
1678 // Verify 'provide stat' works with HTTP-only routing (regression test for stats)
1679 statRes := node.RunIPFS("provide", "stat")
1680 assert.Equal(t, 0, statRes.ExitCode(), "provide stat should succeed with HTTP-only routing")
1681 assert.NotContains(t, statRes.Stderr.String(), "stats not available",
1682 "should not report stats unavailable")
1683 // LegacyProvider outputs "TotalReprovides:" in its stats
1684 assert.Contains(t, statRes.Stdout.String(), "TotalReprovides:",
1685 "should show legacy provider stats")
1686 }
1687
1688 // TestProviderKeystoreDatastoreCompaction verifies that the SweepingProvider's
1689 // keystore uses a datastore factory that creates separate physical datastores
1690 // and reclaims disk space by deleting old datastores after each reset cycle.
1691 //
1692 // The keystore uses two alternating namespaces ("0" and "1") plus a "meta"
1693 // namespace. The lifecycle is:
1694 // 1. First start: namespace "0" is created as the initial active datastore
1695 // 2. First reset (keystore sync at startup): "1" is created, data is written,
1696 // namespaces swap, "0" is destroyed from disk via os.RemoveAll
1697 // 3. Restart: "1" and "meta" survive on disk
1698 // 4. Second reset: "0" is recreated, namespaces swap, "1" is destroyed
1699 func TestProviderKeystoreDatastorePurge(t *testing.T) {
1700 t.Parallel()
1701
1702 h := harness.NewT(t)
1703 node := h.NewNode().Init()
1704 node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1705 node.SetIPFSConfig("Provide.Enabled", true)
1706 node.SetIPFSConfig("Bootstrap", []string{})
1707
1708 // Add content offline so the keystore has something to sync on startup.
1709 for i := range 5 {
1710 node.IPFSAddStr(fmt.Sprintf("keystore-compaction-test-%d", i))
1711 }
1712
1713 keystoreBase := filepath.Join(node.Dir, "provider-keystore")
1714 ns0 := filepath.Join(keystoreBase, "0")
1715 ns1 := filepath.Join(keystoreBase, "1")
1716
1717 // Directory should not exist before starting the daemon.
1718 _, err := os.Stat(keystoreBase)
1719 require.True(t, os.IsNotExist(err), "provider-keystore should not exist before daemon start")
1720
1721 // --- First start: triggers keystore sync (ResetCids) ---
1722 // Init creates "0", then reset swaps to "1" and destroys "0".
1723 node.StartDaemon()
1724
1725 require.Eventually(t, func() bool {
1726 return dirExists(ns1) && !dirExists(ns0)
1727 }, 30*time.Second, 200*time.Millisecond,
1728 "after first reset: ns1 should exist, ns0 should be destroyed")
1729
1730 // --- Restart: triggers a second keystore sync (ResetCids) ---
1731 // Reset swaps back to "0" and destroys "1".
1732 node.StopDaemon()
1733
1734 // Between restarts: ns1 survives on disk, ns0 does not.
1735 assert.True(t, dirExists(ns1), "ns1 should survive shutdown")
1736 assert.False(t, dirExists(ns0), "ns0 should not reappear between restarts")
1737
1738 node.StartDaemon()
1739
1740 require.Eventually(t, func() bool {
1741 return dirExists(ns0) && !dirExists(ns1)
1742 }, 30*time.Second, 200*time.Millisecond,
1743 "after second reset: ns0 should exist, ns1 should be destroyed")
1744
1745 node.StopDaemon()
1746 }
1747
1748 // TestProviderKeystoreMigrationPurge verifies that orphaned keystore data
1749 // left in the shared repo datastore by older Kubo versions is purged on
1750 // the first sweep-enabled daemon start. The migration is triggered by the
1751 // absence of the <repo>/provider-keystore/ directory.
1752 func TestProviderKeystoreMigrationPurge(t *testing.T) {
1753 t.Parallel()
1754
1755 h := harness.NewT(t)
1756 node := h.NewNode().Init()
1757 node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1758 node.SetIPFSConfig("Provide.Enabled", true)
1759 node.SetIPFSConfig("Bootstrap", []string{})
1760
1761 keystoreBase := filepath.Join(node.Dir, "provider-keystore")
1762
1763 // Pre-seed orphaned keystore data into the shared datastore, simulating
1764 // the layout produced by older Kubo that stored keystore entries inline.
1765 const numOrphans = 10
1766 for i := range numOrphans {
1767 node.DatastorePut(
1768 fmt.Sprintf("/provider/keystore/%d/fake-key-%d", i%2, i),
1769 fmt.Sprintf("orphan-%d", i),
1770 )
1771 }
1772
1773 // The orphaned keys should be visible via diag datastore.
1774 count := node.DatastoreCount("/provider/keystore/")
1775 require.Equal(t, int64(numOrphans), count, "orphaned keys should be present before migration")
1776
1777 // The provider-keystore directory must not exist yet (its absence
1778 // triggers the migration).
1779 require.False(t, dirExists(keystoreBase),
1780 "provider-keystore/ should not exist before first sweep-enabled start")
1781
1782 // Start the daemon: this triggers the one-time migration purge.
1783 node.StartDaemon()
1784 node.StopDaemon()
1785
1786 // After migration the seeded orphaned keys should be gone from the
1787 // shared datastore. The diag datastore count command mounts the
1788 // separate provider-keystore datastores, so we check for the specific
1789 // fake keys we seeded to confirm they were purged.
1790 for i := range numOrphans {
1791 key := fmt.Sprintf("/provider/keystore/%d/fake-key-%d", i%2, i)
1792 assert.False(t, node.DatastoreHasKey(key),
1793 "orphaned key %s should be purged after migration", key)
1794 }
1795
1796 // The provider-keystore directory should now exist.
1797 assert.True(t, dirExists(keystoreBase),
1798 "provider-keystore/ should exist after sweep-enabled daemon ran")
1799 }
1800
1801 func dirExists(path string) bool {
1802 info, err := os.Stat(path)
1803 return err == nil && info.IsDir()
1804 }
1805
1806 // TestProviderKeystoreSyncShutdownQuiet verifies two shutdown UX
1807 // guarantees for a daemon running the sweeping provider with a
1808 // pin-walking strategy (see ipfs/kubo#11292):
1809 //
1810 // 1. Shutdown-caused keystore-sync errors never appear at Error
1811 // level. The fix classifies keystore.ErrClosed and context
1812 // cancellation as shutdown-caused and logs at Debug as
1813 // "interrupted by shutdown" instead.
1814 // 2. `ipfs pin ls --stream` running against the daemon returns a
1815 // meaningful error (no panic, no hang) when the daemon is
1816 // shutting down mid-stream.
1817 //
1818 // Determinism: with Provide.DHT.Interval=10ms the periodic
1819 // reprovide goroutine runs syncKeystore back-to-back (ticks coalesce
1820 // under the select), so it is always mid-sync when StopDaemon
1821 // closes the keystore. The line-scan below fails on the exact
1822 // Error+err=keystore-closed/context-canceled combination the old
1823 // code emitted. Empirically this catches the regression on most
1824 // runs (~3 of 5 on a fast workstation); the first few bug-free
1825 // runs were verified by temporarily reverting core/node/provider.go.
1826 func TestProviderKeystoreSyncShutdownQuiet(t *testing.T) {
1827 t.Parallel()
1828
1829 h := harness.NewT(t)
1830 node := h.NewNode().Init()
1831 node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
1832 node.SetIPFSConfig("Provide.Enabled", true)
1833 node.SetIPFSConfig("Provide.Strategy", "pinned+mfs+entities")
1834 // Tight Interval: once the startup sync completes, the periodic
1835 // goroutine runs syncKeystore back-to-back (ticks coalesce under
1836 // the select), so it is always mid-sync when StopDaemon fires.
1837 // This makes the shutdown interrupt deterministic. Briefly
1838 // during startup the first periodic tick may overlap the startup
1839 // sync and emit "reset already in progress" at Error; the log
1840 // scan below explicitly ignores that unrelated class of error.
1841 node.SetIPFSConfig("Provide.DHT.Interval", "10ms")
1842 node.SetIPFSConfig("Bootstrap", []string{})
1843
1844 // Seed recursive pins so the keystore sync has meaningful work.
1845 // Offline bulk add + bulk pin is much faster than per-file
1846 // IPFSAddStr calls for this count.
1847 const nPins = 500
1848 dir := t.TempDir()
1849 for i := range nPins {
1850 require.NoError(t, os.WriteFile(
1851 filepath.Join(dir, fmt.Sprintf("f%04d", i)),
1852 fmt.Appendf(nil, "keystore-shutdown-content-%d", i),
1853 0o600,
1854 ))
1855 }
1856 // --pin=false so the wrapping dir is not auto-pinned; each file
1857 // is then pinned individually below to get nPins separate pin
1858 // index entries (one big recursive pin would not exercise the
1859 // pin-index streamIndex walk the same way).
1860 addRes := node.IPFS("add", "-r", "-q", "--pin=false", dir)
1861 addedCIDs := strings.Split(strings.TrimSpace(addRes.Stdout.String()), "\n")
1862 require.GreaterOrEqual(t, len(addedCIDs), nPins, "expected at least %d CIDs from bulk add", nPins)
1863 pinArgs := append([]string{"pin", "add"}, addedCIDs[:nPins]...)
1864 node.IPFS(pinArgs...)
1865
1866 node.StartDaemonWithReq(harness.RunRequest{
1867 CmdOpts: []harness.CmdOpt{
1868 harness.RunWithEnv(map[string]string{
1869 // Debug for the provider subsystem so the shutdown
1870 // Debug line is visible for post-hoc inspection.
1871 "GOLOG_LOG_LEVEL": "error,provider=debug",
1872 }),
1873 },
1874 }, "")
1875
1876 // Wait for the startup sync to complete so periodic has sole
1877 // access to the keystore when we shut down.
1878 require.Eventually(t, func() bool {
1879 return strings.Contains(node.Daemon.Stderr.String(), "provider keystore sync completed")
1880 }, 30*time.Second, 50*time.Millisecond, "startup keystore sync should complete")
1881
1882 // Let periodic reprovide fire several times.
1883 time.Sleep(1 * time.Second)
1884
1885 // Kick off `ipfs pin ls --stream` against the live RPC. The
1886 // server-side channel is held by the pinner's streamIndex
1887 // goroutine; when StopDaemon below tears down the keystore and
1888 // datastore, the HTTP stream closes under the CLI, which must
1889 // exit cleanly with a meaningful error (no panic, no hang).
1890 pinLsDone := make(chan *harness.RunResult, 1)
1891 go func() {
1892 pinLsDone <- node.RunIPFS("pin", "ls", "--stream")
1893 }()
1894 // Brief delay so the pin ls RPC has started streaming.
1895 time.Sleep(100 * time.Millisecond)
1896
1897 node.StopDaemon()
1898
1899 // --- Daemon-side assertions ---
1900
1901 daemonLog := node.Daemon.Stderr.String()
1902
1903 // Scan for the specific bug pattern: an Error-level line from
1904 // the provider subsystem about "keystore sync" whose err field
1905 // is the shutdown-caused "keystore is closed" or "context
1906 // canceled". The fix routes those to Debug; only unrelated
1907 // errors (e.g. "reset already in progress" from test-induced
1908 // overlap) remain at Error and are ignored by this check.
1909 for line := range strings.SplitSeq(daemonLog, "\n") {
1910 if !strings.Contains(line, "\tERROR\t") {
1911 continue
1912 }
1913 if !strings.Contains(line, "provider keystore sync") {
1914 continue
1915 }
1916 if strings.Contains(line, `"err": "keystore is closed"`) ||
1917 strings.Contains(line, `"err": "context canceled"`) {
1918 t.Errorf("shutdown-caused keystore sync error should be logged at Debug, got Error:\n%s", line)
1919 }
1920 }
1921
1922 // --- Client-side assertions (ipfs pin ls --stream) ---
1923
1924 var pinLs *harness.RunResult
1925 select {
1926 case pinLs = <-pinLsDone:
1927 case <-time.After(15 * time.Second):
1928 t.Fatal("ipfs pin ls --stream did not return within 15s of daemon shutdown")
1929 }
1930
1931 pinLsOut := pinLs.Stdout.String() + pinLs.Stderr.String()
1932 require.NotContains(t, pinLsOut, "panic:",
1933 "ipfs pin ls must not observe a daemon panic")
1934 // Either the stream drained before shutdown (exit 0) or the
1935 // server dropped it mid-stream (non-zero exit with a meaningful
1936 // error message). Silent non-zero exits are confusing and fail.
1937 if pinLs.ExitCode() != 0 {
1938 require.NotEmpty(t, strings.TrimSpace(pinLs.Stderr.String()),
1939 "pin ls exited non-zero but produced no error message")
1940 }
1941 }