fix(provider): purge keystore datastore after reset (#11198)
* fix(provider): purge keystore datastore after reset * changelog * use MapDatastore if no datastore is configured * bump kad-dht to latest commit * purge orphaned keystore migration * bump kad-dht * use main datastore for keystore "meta" store * add provider/keystore/0 and /1 to ipfs diag command mount keystore datastores to /provider/keystore/0 and /1 so that they are included in the ipfs diag datastore command * fix(provider): reject unexpected keystore suffix to prevent stray deletions destroyDs calls os.RemoveAll with a suffix from the upstream library. If suffix were ever ".." or empty, this could delete wrong directories. Validate that suffix is "0" or "1" in both createDs and destroyDs. * fix(provider): close opened datastores when mounting partially fails If opening datastore "0" succeeds but "1" fails, MountKeystoreDatastores returned an error without closing "0". * fix(provider): defer batch creation in orphan purge until keys are found Avoids allocating a datastore batch when no orphaned keys exist. * fix(provider): warn on unrecognized datastore wrapper types findRootDatastoreSpec silently returns wrapper specs it doesn't know about. If a plugin adds a wrapper with a "child" field, openDatastoreAt gets the wrapper instead of the leaf backend and fails confusingly. Log a warning so operators can spot the issue. * docs: document keystore migration behavior on upgrade and downgrade - explain why context.Background() is used in the migration code - add changelog note about the provide cycle restarting on upgrade - add downgrade caveat about orphaned provider-keystore directory * chore(deps): bump go-libp2p-kad-dht to latest keystore factory commit * fix(provider): harden keystore migration and spec handling - chunk orphan purge into 4096-key batches to bound memory and match existing batching patterns in the same file - cancel the purge context via fx.Lifecycle OnStop so SIGINT during startup does not block indefinitely - deep-copy slices in copySpec (not just maps) so the function matches its documented "deep-copy" contract - return nil from findRootDatastoreSpec when no "/" mount exists, so callers fall back to in-memory instead of passing a mount-type spec to openDatastoreAt - rename local variable to avoid shadowing the mount package import * test(provider): add migration purge test and diag datastore put command - add `ipfs diag datastore put` subcommand for writing arbitrary key-value pairs to the datastore (offline, experimental) - add DatastorePut harness helper for CLI tests - add TestProviderKeystoreMigrationPurge: seeds orphaned keystore keys via `put`, starts the daemon to trigger migration, verifies the orphaned keys are purged and provider-keystore/ dir is created - add put/get roundtrip test for diag datastore * chore(deps): bump go-libp2p-kad-dht to 1bede74b8246 * fix(provider): log keystore datastore create and destroy operations * docs: rewrite provider keystore changelog to focus on user impact * bump kad-dht@master --------- Co-authored-by: Marcin Rataj <lidel@lidel.org>
Guillaume Michel committed
Mar 19, 2026 at 11:27 UTC
99c092b1df7c52a2e514cf678918b522ff9d578b
13 files changed
+633
-29
core/commands/commands_test.go
+1
@@ -79,6 +79,7 @@ func TestCommands(t *testing.T) {
79
"/diag/datastore",
80
"/diag/datastore/count",
81
"/diag/datastore/get",
82
+ "/diag/datastore/put",
83
"/diag/profile",
84
"/diag/sys",
85
"/files",
core/commands/diag.go
+89
-11
@@ -7,9 +7,11 @@ import (
7
"io"
8
9
"github.com/ipfs/go-datastore"
10
+ "github.com/ipfs/go-datastore/mount"
11
"github.com/ipfs/go-datastore/query"
12
cmds "github.com/ipfs/go-ipfs-cmds"
13
oldcmds "github.com/ipfs/kubo/commands"
14
+ node "github.com/ipfs/kubo/core/node"
15
fsrepo "github.com/ipfs/kubo/repo/fsrepo"
16
)
17
@@ -41,7 +43,11 @@ in production workflows. The datastore format may change between versions.
43
44
The daemon must not be running when calling these commands.
45
44
-EXAMPLE
46
+When the provider keystore datastores exist on disk (nodes with
47
+Provide.DHT.SweepEnabled=true), they are automatically mounted into the
48
+datastore view under /provider/keystore/0/ and /provider/keystore/1/.
49
+
50
+EXAMPLES
51
52
Inspecting pubsub seqno validator state:
53
@@ -51,10 +57,20 @@ Inspecting pubsub seqno validator state:
57
Key: /pubsub/seqno/12D3KooW...
58
Hex Dump:
59
00000000 18 81 81 c8 91 c0 ea f6 |........|
60
+
61
+Writing a test key (debugging only):
62
+
63
+ $ ipfs diag datastore put /test/mykey "hello"
64
+
65
+Inspecting provider keystore (requires SweepEnabled):
66
+
67
+ $ ipfs diag datastore count /provider/keystore/0/
68
+ $ ipfs diag datastore count /provider/keystore/1/
69
`,
70
},
71
Subcommands: map[string]*cmds.Command{
72
"get": diagDatastoreGetCmd,
73
+ "put": diagDatastorePutCmd,
74
"count": diagDatastoreCountCmd,
75
},
76
}
@@ -67,6 +83,36 @@ type diagDatastoreGetResult struct {
83
HexDump string `json:"hex_dump,omitempty"`
84
}
85
86
+// openDiagDatastore opens the repo datastore and conditionally mounts any
87
+// provider keystore datastores that exist on disk. It returns the composite
88
+// datastore and a cleanup function that must be called when done.
89
+func openDiagDatastore(env cmds.Environment) (datastore.Datastore, func(), error) {
90
+ cctx := env.(*oldcmds.Context)
91
+ repo, err := fsrepo.Open(cctx.ConfigRoot)
92
+ if err != nil {
93
+ return nil, nil, fmt.Errorf("failed to open repo: %w", err)
94
+ }
95
+
96
+ extraMounts, extraCloser, err := node.MountKeystoreDatastores(repo)
97
+ if err != nil {
98
+ repo.Close()
99
+ return nil, nil, err
100
+ }
101
+
102
+ closer := func() {
103
+ extraCloser()
104
+ repo.Close()
105
+ }
106
+
107
+ if len(extraMounts) == 0 {
108
+ return repo.Datastore(), closer, nil
109
+ }
110
+
111
+ mounts := []mount.Mount{{Prefix: datastore.NewKey("/"), Datastore: repo.Datastore()}}
112
+ mounts = append(mounts, extraMounts...)
113
+ return mount.New(mounts), closer, nil
114
+}
115
+
116
var diagDatastoreGetCmd = &cmds.Command{
117
Status: cmds.Experimental,
118
Helptext: cmds.HelpText{
@@ -89,16 +135,14 @@ WARNING: FOR DEBUGGING/TESTING ONLY
135
NoRemote: true,
136
PreRun: DaemonNotRunning,
137
Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
92
- cctx := env.(*oldcmds.Context)
93
- repo, err := fsrepo.Open(cctx.ConfigRoot)
138
+ ds, closer, err := openDiagDatastore(env)
139
if err != nil {
95
- return fmt.Errorf("failed to open repo: %w", err)
140
+ return err
141
}
97
- defer repo.Close()
142
+ defer closer()
143
144
keyStr := req.Arguments[0]
145
key := datastore.NewKey(keyStr)
101
- ds := repo.Datastore()
146
147
val, err := ds.Get(req.Context, key)
148
if err != nil {
@@ -133,6 +177,42 @@ WARNING: FOR DEBUGGING/TESTING ONLY
177
},
178
}
179
180
+var diagDatastorePutCmd = &cmds.Command{
181
+ Status: cmds.Experimental,
182
+ Helptext: cmds.HelpText{
183
+ Tagline: "Write a raw key-value pair to the datastore.",
184
+ ShortDescription: `
185
+Stores the given value at the specified datastore key.
186
+
187
+The daemon must not be running when using this command.
188
+
189
+WARNING: FOR DEBUGGING/TESTING ONLY
190
+`,
191
+ },
192
+ Arguments: []cmds.Argument{
193
+ cmds.StringArg("key", true, false, "Datastore key (e.g., /test/mykey)"),
194
+ cmds.StringArg("value", true, false, "Value to store (as a string)"),
195
+ },
196
+ NoRemote: true,
197
+ PreRun: DaemonNotRunning,
198
+ Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
199
+ ds, closer, err := openDiagDatastore(env)
200
+ if err != nil {
201
+ return err
202
+ }
203
+ defer closer()
204
+
205
+ key := datastore.NewKey(req.Arguments[0])
206
+ if err := ds.Put(req.Context, key, []byte(req.Arguments[1])); err != nil {
207
+ return fmt.Errorf("failed to put key: %w", err)
208
+ }
209
+ if err := ds.Sync(req.Context, key); err != nil {
210
+ return fmt.Errorf("failed to sync: %w", err)
211
+ }
212
+ return nil
213
+ },
214
+}
215
+
216
type diagDatastoreCountResult struct {
217
Prefix string `json:"prefix"`
218
Count int64 `json:"count"`
@@ -156,15 +236,13 @@ WARNING: FOR DEBUGGING/TESTING ONLY
236
NoRemote: true,
237
PreRun: DaemonNotRunning,
238
Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
159
- cctx := env.(*oldcmds.Context)
160
- repo, err := fsrepo.Open(cctx.ConfigRoot)
239
+ ds, closer, err := openDiagDatastore(env)
240
if err != nil {
162
- return fmt.Errorf("failed to open repo: %w", err)
241
+ return err
242
}
164
- defer repo.Close()
243
+ defer closer()
244
245
prefix := req.Arguments[0]
167
- ds := repo.Datastore()
246
247
q := query.Query{
248
Prefix: prefix,
core/node/provider.go
+307
-9
@@ -4,6 +4,8 @@ import (
4
"context"
5
"errors"
6
"fmt"
7
+ "os"
8
+ "path/filepath"
9
"time"
10
11
"github.com/ipfs/boxo/blockstore"
@@ -14,11 +16,13 @@ import (
16
"github.com/ipfs/boxo/provider"
17
"github.com/ipfs/go-cid"
18
"github.com/ipfs/go-datastore"
19
+ "github.com/ipfs/go-datastore/mount"
20
"github.com/ipfs/go-datastore/namespace"
21
"github.com/ipfs/go-datastore/query"
22
log "github.com/ipfs/go-log/v2"
23
"github.com/ipfs/kubo/config"
24
"github.com/ipfs/kubo/repo"
25
+ "github.com/ipfs/kubo/repo/fsrepo"
26
irouting "github.com/ipfs/kubo/routing"
27
dht "github.com/libp2p/go-libp2p-kad-dht"
28
"github.com/libp2p/go-libp2p-kad-dht/amino"
@@ -48,14 +52,30 @@ const (
52
// Datastore key used to store previous reprovide strategy.
53
reprovideStrategyKey = "/reprovideStrategy"
54
51
- // Datastore namespace prefix for provider data.
52
- providerDatastorePrefix = "provider"
53
- // Datastore path for the provider keystore.
54
- keystoreDatastorePath = "keystore"
55
+ // KeystoreDatastorePath is the base directory for the provider keystore datastores.
56
+ KeystoreDatastorePath = "provider-keystore"
57
+)
58
+
59
+var (
60
+ // Datastore namespace key for provider data.
61
+ providerDatastoreKey = datastore.NewKey("provider")
62
+ // Datastore namespace key for provider keystore data.
63
+ keystoreDatastoreKey = datastore.NewKey("keystore")
64
)
65
66
var errAcceleratedDHTNotReady = errors.New("AcceleratedDHTClient: routing table not ready")
67
68
+// validateKeystoreSuffix rejects any suffix other than "0" or "1".
69
+// The upstream library uses these two values as alternating namespace
70
+// identifiers. Validating here prevents accidental deletion of unrelated
71
+// directories via os.RemoveAll if the upstream ever changes its scheme.
72
+func validateKeystoreSuffix(suffix string) error {
73
+ if suffix != "0" && suffix != "1" {
74
+ return fmt.Errorf("unexpected keystore suffix %q, expected \"0\" or \"1\"", suffix)
75
+ }
76
+ return nil
77
+}
78
+
79
// Interval between reprovide queue monitoring checks for slow reprovide alerts.
80
// Used when Provide.DHT.SweepEnabled=true
81
const reprovideAlertPollInterval = 15 * time.Minute
@@ -369,19 +389,297 @@ type addrsFilter interface {
389
FilteredAddrs() []ma.Multiaddr
390
}
391
392
+// findRootDatastoreSpec extracts the leaf datastore spec for the root ("/")
393
+// mount from the repo's Datastore.Spec config. It unwraps mount (picks the "/"
394
+// mountpoint), measure, and log wrappers to find the actual backend spec
395
+// (e.g., levelds, pebbleds).
396
+func findRootDatastoreSpec(spec map[string]any) map[string]any {
397
+ if spec == nil {
398
+ return nil
399
+ }
400
+ switch spec["type"] {
401
+ case "mount":
402
+ mounts, ok := spec["mounts"].([]any)
403
+ if !ok {
404
+ return spec
405
+ }
406
+ for _, m := range mounts {
407
+ mnt, ok := m.(map[string]any)
408
+ if !ok {
409
+ continue
410
+ }
411
+ if mnt["mountpoint"] == "/" {
412
+ return findRootDatastoreSpec(mnt)
413
+ }
414
+ }
415
+ // No root mount found; return nil so callers fall back gracefully
416
+ // (in-memory datastore or skip mounting) rather than passing a
417
+ // mount-type spec to openDatastoreAt which expects a leaf backend.
418
+ return nil
419
+ case "measure", "log":
420
+ if child, ok := spec["child"].(map[string]any); ok {
421
+ return findRootDatastoreSpec(child)
422
+ }
423
+ return spec
424
+ default:
425
+ if _, hasChild := spec["child"]; hasChild {
426
+ logger.Warnw("unrecognized datastore wrapper type, using as-is",
427
+ "type", spec["type"])
428
+ }
429
+ return spec
430
+ }
431
+}
432
+
433
+// MountKeystoreDatastores opens any provider keystore datastores that exist on
434
+// disk and returns them as mount.Mount entries ready to be combined with the
435
+// main repo datastore. The caller must call the returned cleanup function when
436
+// done. Returns nil mounts and a no-op closer if no keystores exist.
437
+func MountKeystoreDatastores(repo repo.Repo) ([]mount.Mount, func(), error) {
438
+ cfg, err := repo.Config()
439
+ if err != nil {
440
+ return nil, nil, fmt.Errorf("reading repo config: %w", err)
441
+ }
442
+
443
+ rootSpec := findRootDatastoreSpec(cfg.Datastore.Spec)
444
+ if rootSpec == nil {
445
+ return nil, func() {}, nil
446
+ }
447
+
448
+ keystoreBasePath := filepath.Join(repo.Path(), KeystoreDatastorePath)
449
+ var mounts []mount.Mount
450
+ var closers []func()
451
+
452
+ for _, suffix := range []string{"0", "1"} {
453
+ dir := filepath.Join(keystoreBasePath, suffix)
454
+ if _, err := os.Stat(dir); err != nil {
455
+ continue
456
+ }
457
+ ds, err := openDatastoreAt(rootSpec, dir)
458
+ if err != nil {
459
+ for _, c := range closers {
460
+ c()
461
+ }
462
+ return nil, nil, err
463
+ }
464
+ prefix := providerDatastoreKey.Child(keystoreDatastoreKey).ChildString(suffix)
465
+ mounts = append(mounts, mount.Mount{Prefix: prefix, Datastore: ds})
466
+ closers = append(closers, func() { ds.Close() })
467
+ }
468
+
469
+ closer := func() {
470
+ for _, c := range closers {
471
+ c()
472
+ }
473
+ }
474
+ return mounts, closer, nil
475
+}
476
+
477
+// openDatastoreAt opens a datastore using the given spec at the specified path.
478
+// It deep-copies the spec to avoid mutating the original.
479
+func openDatastoreAt(rootSpec map[string]any, path string) (datastore.Batching, error) {
480
+ spec := copySpec(rootSpec)
481
+ spec["path"] = path
482
+ dsc, err := fsrepo.AnyDatastoreConfig(spec)
483
+ if err != nil {
484
+ return nil, fmt.Errorf("creating datastore config for %s: %w", path, err)
485
+ }
486
+ return dsc.Create("")
487
+}
488
+
489
+// copySpec deep-copies a datastore spec map so modifications (e.g., changing
490
+// the path) don't affect the original.
491
+func copySpec(spec map[string]any) map[string]any {
492
+ if spec == nil {
493
+ return nil
494
+ }
495
+ cp := make(map[string]any, len(spec))
496
+ for k, v := range spec {
497
+ switch val := v.(type) {
498
+ case map[string]any:
499
+ cp[k] = copySpec(val)
500
+ case []any:
501
+ s := make([]any, len(val))
502
+ for i, elem := range val {
503
+ if m, ok := elem.(map[string]any); ok {
504
+ s[i] = copySpec(m)
505
+ } else {
506
+ s[i] = elem
507
+ }
508
+ }
509
+ cp[k] = s
510
+ default:
511
+ cp[k] = v
512
+ }
513
+ }
514
+ return cp
515
+}
516
+
517
+// purgeBatchSize is the number of keys deleted per batch commit during
518
+// orphaned keystore cleanup. Each commit is a cancellation checkpoint.
519
+const purgeBatchSize = 1 << 12 // 4096
520
+
521
+// purgeOrphanedKeystoreData deletes all keys under /provider/keystore/ from the
522
+// shared repo datastore. These were written by older Kubo versions that stored
523
+// provider keystore data inline in the shared datastore. The new code uses
524
+// separate filesystem datastores under <repo>/{KeystoreDatastorePath}/ instead.
525
+//
526
+// The operation is idempotent and safe to interrupt: partial completion is
527
+// fine because already-deleted keys are no-ops on re-run.
528
+func purgeOrphanedKeystoreData(ctx context.Context, ds datastore.Batching) error {
529
+ orphanedPrefix := providerDatastoreKey.Child(keystoreDatastoreKey).String()
530
+ syncKey := datastore.NewKey(orphanedPrefix)
531
+
532
+ results, err := ds.Query(ctx, query.Query{
533
+ Prefix: orphanedPrefix,
534
+ KeysOnly: true,
535
+ })
536
+ if err != nil {
537
+ return fmt.Errorf("querying orphaned keystore data: %w", err)
538
+ }
539
+ defer results.Close()
540
+
541
+ var batch datastore.Batch
542
+ var count, pending int
543
+ for result := range results.Next() {
544
+ if ctx.Err() != nil {
545
+ return ctx.Err()
546
+ }
547
+ if result.Error != nil {
548
+ return fmt.Errorf("iterating orphaned keystore data: %w", result.Error)
549
+ }
550
+ if batch == nil {
551
+ batch, err = ds.Batch(ctx)
552
+ if err != nil {
553
+ return fmt.Errorf("creating batch for orphaned keystore cleanup: %w", err)
554
+ }
555
+ }
556
+ if err := batch.Delete(ctx, datastore.NewKey(result.Key)); err != nil {
557
+ return fmt.Errorf("batch deleting orphaned key %s: %w", result.Key, err)
558
+ }
559
+ count++
560
+ pending++
561
+ if pending >= purgeBatchSize {
562
+ if err := batch.Commit(ctx); err != nil {
563
+ return fmt.Errorf("committing orphaned keystore cleanup batch: %w", err)
564
+ }
565
+ if err := ds.Sync(ctx, syncKey); err != nil {
566
+ return fmt.Errorf("syncing orphaned keystore cleanup: %w", err)
567
+ }
568
+ batch = nil
569
+ pending = 0
570
+ }
571
+ }
572
+ if pending > 0 {
573
+ if err := batch.Commit(ctx); err != nil {
574
+ return fmt.Errorf("committing orphaned keystore cleanup batch: %w", err)
575
+ }
576
+ if err := ds.Sync(ctx, syncKey); err != nil {
577
+ return fmt.Errorf("syncing orphaned keystore cleanup: %w", err)
578
+ }
579
+ }
580
+ if count > 0 {
581
+ logger.Infow("purged orphaned provider keystore data from shared datastore", "keys", count)
582
+ }
583
+ return nil
584
+}
585
+
586
func SweepingProviderOpt(cfg *config.Config) fx.Option {
587
reprovideInterval := cfg.Provide.DHT.Interval.WithDefault(config.DefaultProvideDHTInterval)
588
type providerInput struct {
589
fx.In
590
DHT routing.Routing `name:"dhtc"`
591
Repo repo.Repo
592
+ Lc fx.Lifecycle
593
}
594
sweepingReprovider := fx.Provide(func(in providerInput) (DHTProvider, *keystore.ResettableKeystore, error) {
380
- ds := namespace.Wrap(in.Repo.Datastore(), datastore.NewKey(providerDatastorePrefix))
381
- ks, err := keystore.NewResettableKeystore(ds,
382
- keystore.WithPrefixBits(16),
383
- keystore.WithDatastorePath(keystoreDatastorePath),
384
- keystore.WithBatchSize(int(cfg.Provide.DHT.KeystoreBatchSize.WithDefault(config.DefaultProvideDHTKeystoreBatchSize))),
595
+ ds := namespace.Wrap(in.Repo.Datastore(), providerDatastoreKey)
596
+
597
+ // Get repo path and config to determine datastore type
598
+ repoPath := in.Repo.Path()
599
+ repoCfg, err := in.Repo.Config()
600
+ if err != nil {
601
+ return nil, nil, fmt.Errorf("getting repo config: %w", err)
602
+ }
603
+
604
+ // Find the root datastore type (levelds, pebbleds, etc.)
605
+ rootSpec := findRootDatastoreSpec(repoCfg.Datastore.Spec)
606
+
607
+ // Keystore datastores live at <repo>/{KeystoreDatastorePath}/<suffix>
608
+ keystoreBasePath := filepath.Join(repoPath, KeystoreDatastorePath)
609
+
610
+ createDs := func(suffix string) (datastore.Batching, error) {
611
+ if err := validateKeystoreSuffix(suffix); err != nil {
612
+ return nil, err
613
+ }
614
+ // When no datastore spec is configured (e.g., test/mock repos),
615
+ // fall back to an in-memory datastore.
616
+ if rootSpec == nil {
617
+ return datastore.NewMapDatastore(), nil
618
+ }
619
+ if err := os.MkdirAll(keystoreBasePath, 0o755); err != nil {
620
+ return nil, fmt.Errorf("creating keystore base directory: %w", err)
621
+ }
622
+ ds, err := openDatastoreAt(rootSpec, filepath.Join(keystoreBasePath, suffix))
623
+ if err != nil {
624
+ return nil, err
625
+ }
626
+ logger.Infow("provider keystore: opened datastore", "suffix", suffix, "path", filepath.Join(keystoreBasePath, suffix))
627
+ return ds, nil
628
+ }
629
+
630
+ destroyDs := func(suffix string) error {
631
+ if err := validateKeystoreSuffix(suffix); err != nil {
632
+ return err
633
+ }
634
+ logger.Infow("provider keystore: removing datastore from disk", "suffix", suffix, "path", filepath.Join(keystoreBasePath, suffix))
635
+ return os.RemoveAll(filepath.Join(keystoreBasePath, suffix))
636
+ }
637
+
638
+ // One-time cleanup of stale keystore data left by older Kubo in the
639
+ // shared repo datastore under /provider/keystore/. New code stores
640
+ // bulk key data in separate filesystem datastores under
641
+ // <repo>/{KeystoreDatastorePath}/ while still using the same
642
+ // /provider/keystore/ namespace in the shared datastore for metadata.
643
+ //
644
+ // The absence of the keystoreBasePath directory signals a first run
645
+ // after upgrade: the directory is created later by createDs on first
646
+ // use, so it doubles as a "cleanup done" flag. If the process dies
647
+ // mid-purge the directory still won't exist and the cleanup re-runs
648
+ // on next start (it is idempotent). Must run synchronously before
649
+ // NewResettableKeystore to avoid racing with reads on the same
650
+ // namespace.
651
+ if _, statErr := os.Stat(keystoreBasePath); os.IsNotExist(statErr) {
652
+ logger.Infow("migrating provider keystore data from shared datastore to separate filesystem datastores", "path", keystoreBasePath)
653
+ // Create a cancellable context for the purge. The OnStop hook
654
+ // below calls purgeCancel when the node receives a shutdown
655
+ // signal (e.g., SIGINT), which interrupts the purge loop
656
+ // instead of blocking indefinitely.
657
+ purgeCtx, purgeCancel := context.WithCancel(context.Background())
658
+ in.Lc.Append(fx.Hook{
659
+ OnStop: func(_ context.Context) error {
660
+ purgeCancel()
661
+ return nil
662
+ },
663
+ })
664
+ if purgeErr := purgeOrphanedKeystoreData(purgeCtx, in.Repo.Datastore()); purgeErr != nil {
665
+ if purgeCtx.Err() != nil {
666
+ logger.Infow("provider keystore migration interrupted by shutdown, will resume on next start")
667
+ } else {
668
+ logger.Warnw("provider keystore migration failed, will retry on next start", "error", purgeErr)
669
+ }
670
+ } else {
671
+ logger.Infow("provider keystore migration completed")
672
+ }
673
+ purgeCancel()
674
+ }
675
+
676
+ keystoreDs := namespace.Wrap(ds, keystoreDatastoreKey)
677
+ ks, err := keystore.NewResettableKeystore(keystoreDs,
678
+ keystore.WithDatastoreFactory(createDs, destroyDs),
679
+ keystore.KeystoreOption(
680
+ keystore.WithPrefixBits(16),
681
+ keystore.WithBatchSize(int(cfg.Provide.DHT.KeystoreBatchSize.WithDefault(config.DefaultProvideDHTKeystoreBatchSize))),
682
+ ),
683
)
684
if err != nil {
685
return nil, nil, err
docs/changelogs/v0.41.md
+22
@@ -10,6 +10,7 @@ This release was brought to you by the [Shipyard](https://ipshipyard.com/) team.
10
11
- [Overview](#overview)
12
- [🔦 Highlights](#-highlights)
13
+ - [🗑️ Faster Provide Queue Disk Reclamation](#-faster-provide-queue-disk-reclamation)
14
- [🖥️ WebUI Improvements](#-webui-improvements)
15
- [🔧 Correct provider addresses for custom HTTP routing](#-correct-provider-addresses-for-custom-http-routing)
16
- [📦️ Dependency updates](#-dependency-updates)
@@ -20,6 +21,27 @@ This release was brought to you by the [Shipyard](https://ipshipyard.com/) team.
21
22
### 🔦 Highlights
23
24
+#### 🗑️ Faster Provide Queue Disk Reclamation
25
+
26
+Nodes with significant amount of data and DHT provide sweep enabled
27
+(`Provide.DHT.SweepEnabled`, the default since Kubo 0.39) could see their
28
+`datastore/` directory grow continuously.
29
+Each reprovide cycle rewrote the provider keystore inside the shared repo
30
+datastore, generating tombstones faster than the storage engine could compact
31
+them, and in default configuration Kubo was slow to reclaim this space.
32
+
33
+The provider keystore now lives in a dedicated datastore under
34
+`$IPFS_PATH/provider-keystore/`. After each reprovide cycle the old datastore
35
+is removed from disk entirely, so space is reclaimed immediately regardless
36
+of storage backend.
37
+
38
+On first start after upgrading, stale keystore data is cleaned up from the
39
+shared datastore automatically.
40
+
41
+To learn more, see [kubo#11096](https://github.com/ipfs/kubo/issues/11096),
42
+[kubo#11198](https://github.com/ipfs/kubo/pull/11198), and
43
+[go-libp2p-kad-dht#1233](https://github.com/libp2p/go-libp2p-kad-dht/pull/1233).
44
+
45
#### 🖥️ WebUI Improvements
46
47
IPFS Web UI has been updated to [v4.12.0](https://github.com/ipfs/ipfs-webui/releases/tag/v4.12.0).
docs/examples/kubo-as-a-library/go.mod
+1
-1
@@ -116,7 +116,7 @@ require (
116
github.com/libp2p/go-doh-resolver v0.5.0 // indirect
117
github.com/libp2p/go-flow-metrics v0.3.0 // indirect
118
github.com/libp2p/go-libp2p-asn-util v0.4.1 // indirect
119
- github.com/libp2p/go-libp2p-kad-dht v0.38.0 // indirect
119
+ github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29 // indirect
120
github.com/libp2p/go-libp2p-kbucket v0.8.0 // indirect
121
github.com/libp2p/go-libp2p-pubsub v0.15.0 // indirect
122
github.com/libp2p/go-libp2p-pubsub-router v0.6.0 // indirect
docs/examples/kubo-as-a-library/go.sum
+2
-2
@@ -486,8 +486,8 @@ github.com/libp2p/go-libp2p-asn-util v0.4.1 h1:xqL7++IKD9TBFMgnLPZR6/6iYhawHKHl9
486
github.com/libp2p/go-libp2p-asn-util v0.4.1/go.mod h1:d/NI6XZ9qxw67b4e+NgpQexCIiFYJjErASrYW4PFDN8=
487
github.com/libp2p/go-libp2p-core v0.2.4/go.mod h1:STh4fdfa5vDYr0/SzYYeqnt+E6KfEV5VxfIrm0bcI0g=
488
github.com/libp2p/go-libp2p-core v0.3.0/go.mod h1:ACp3DmS3/N64c2jDzcV429ukDpicbL6+TrrxANBjPGw=
489
-github.com/libp2p/go-libp2p-kad-dht v0.38.0 h1:NToFzwvICo6ghDfSwuTmROCtl9LDXSZT1VawEbm4NUs=
490
-github.com/libp2p/go-libp2p-kad-dht v0.38.0/go.mod h1:g/CefQilAnCMyUH52A6tUGbe17NgQ8q26MaZCA968iI=
489
+github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29 h1:ZSi0kAWeDUeDPnoiVZ75Hyun7+wksJpQxFiz6aWNrys=
490
+github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29/go.mod h1:g/CefQilAnCMyUH52A6tUGbe17NgQ8q26MaZCA968iI=
491
github.com/libp2p/go-libp2p-kbucket v0.3.1/go.mod h1:oyjT5O7tS9CQurok++ERgc46YLwEpuGoFq9ubvoUOio=
492
github.com/libp2p/go-libp2p-kbucket v0.8.0 h1:QAK7RzKJpYe+EuSEATAaaHYMYLkPDGC18m9jxPLnU8s=
493
github.com/libp2p/go-libp2p-kbucket v0.8.0/go.mod h1:JMlxqcEyKwO6ox716eyC0hmiduSWZZl6JY93mGaaqc4=
go.mod
+1
-1
@@ -52,7 +52,7 @@ require (
52
github.com/libp2p/go-doh-resolver v0.5.0
53
github.com/libp2p/go-libp2p v0.47.0
54
github.com/libp2p/go-libp2p-http v0.5.0
55
- github.com/libp2p/go-libp2p-kad-dht v0.38.0
55
+ github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29
56
github.com/libp2p/go-libp2p-kbucket v0.8.0
57
github.com/libp2p/go-libp2p-pubsub v0.15.0
58
github.com/libp2p/go-libp2p-pubsub-router v0.6.0
go.sum
+2
-2
@@ -543,8 +543,8 @@ github.com/libp2p/go-libp2p-gostream v0.6.0 h1:QfAiWeQRce6pqnYfmIVWJFXNdDyfiR/qk
543
github.com/libp2p/go-libp2p-gostream v0.6.0/go.mod h1:Nywu0gYZwfj7Jc91PQvbGU8dIpqbQQkjWgDuOrFaRdA=
544
github.com/libp2p/go-libp2p-http v0.5.0 h1:+x0AbLaUuLBArHubbbNRTsgWz0RjNTy6DJLOxQ3/QBc=
545
github.com/libp2p/go-libp2p-http v0.5.0/go.mod h1:glh87nZ35XCQyFsdzZps6+F4HYI6DctVFY5u1fehwSg=
546
-github.com/libp2p/go-libp2p-kad-dht v0.38.0 h1:NToFzwvICo6ghDfSwuTmROCtl9LDXSZT1VawEbm4NUs=
547
-github.com/libp2p/go-libp2p-kad-dht v0.38.0/go.mod h1:g/CefQilAnCMyUH52A6tUGbe17NgQ8q26MaZCA968iI=
546
+github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29 h1:ZSi0kAWeDUeDPnoiVZ75Hyun7+wksJpQxFiz6aWNrys=
547
+github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29/go.mod h1:g/CefQilAnCMyUH52A6tUGbe17NgQ8q26MaZCA968iI=
548
github.com/libp2p/go-libp2p-kbucket v0.3.1/go.mod h1:oyjT5O7tS9CQurok++ERgc46YLwEpuGoFq9ubvoUOio=
549
github.com/libp2p/go-libp2p-kbucket v0.8.0 h1:QAK7RzKJpYe+EuSEATAaaHYMYLkPDGC18m9jxPLnU8s=
550
github.com/libp2p/go-libp2p-kbucket v0.8.0/go.mod h1:JMlxqcEyKwO6ox716eyC0hmiduSWZZl6JY93mGaaqc4=
test/cli/diag_datastore_test.go
+79
@@ -2,6 +2,8 @@ package cli
2
3
import (
4
"encoding/json"
5
+ "os"
6
+ "path/filepath"
7
"testing"
8
9
"github.com/ipfs/kubo/test/cli/harness"
@@ -130,6 +132,18 @@ func TestDiagDatastore(t *testing.T) {
132
assert.Contains(t, res.Stderr.String(), "key not found")
133
})
134
135
+ t.Run("diag datastore put and get roundtrip", func(t *testing.T) {
136
+ t.Parallel()
137
+ node := harness.NewT(t).NewNode().Init()
138
+
139
+ node.DatastorePut("/test/roundtrip", "hello world")
140
+ assert.True(t, node.DatastoreHasKey("/test/roundtrip"))
141
+ assert.Equal(t, []byte("hello world"), node.DatastoreGet("/test/roundtrip"))
142
+
143
+ count := node.DatastoreCount("/test/")
144
+ assert.Equal(t, int64(1), count)
145
+ })
146
+
147
t.Run("diag datastore commands require daemon to be stopped", func(t *testing.T) {
148
t.Parallel()
149
node := harness.NewT(t).NewNode().Init().StartDaemon()
@@ -144,4 +158,69 @@ func TestDiagDatastore(t *testing.T) {
158
assert.Error(t, res.Err, "count should fail when daemon is running")
159
assert.Contains(t, res.Stderr.String(), "ipfs daemon is running")
160
})
161
+
162
+ t.Run("provider keystore datastores are visible in unified view", func(t *testing.T) {
163
+ t.Parallel()
164
+ node := harness.NewT(t).NewNode().Init()
165
+ node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
166
+ node.SetIPFSConfig("Provide.Enabled", true)
167
+
168
+ // Start daemon to create the provider-keystore datastores, then add data
169
+ node.StartDaemon()
170
+ cid := node.IPFSAddStr("data for provider keystore test")
171
+ node.IPFS("pin", "add", cid)
172
+ node.StopDaemon()
173
+
174
+ // Verify the provider-keystore directory was created
175
+ keystorePath := filepath.Join(node.Dir, "provider-keystore")
176
+ _, err := os.Stat(keystorePath)
177
+ require.NoError(t, err, "provider-keystore directory should exist after sweep-enabled daemon ran")
178
+
179
+ // Count entries in each keystore namespace via the unified view
180
+ for _, prefix := range []string{"/provider/keystore/0/", "/provider/keystore/1/"} {
181
+ res := node.IPFS("diag", "datastore", "count", prefix)
182
+ assert.NoError(t, res.Err)
183
+ t.Logf("count %s: %s", prefix, res.Stdout.String())
184
+ }
185
+
186
+ // The total count under /provider/keystore/ should include entries
187
+ // from both keystore instances (0 and 1)
188
+ count := node.DatastoreCount("/provider/keystore/")
189
+ t.Logf("total /provider/keystore/ entries: %d", count)
190
+ assert.Greater(t, count, int64(0), "should have provider keystore entries")
191
+ })
192
+
193
+ t.Run("provider keystore count JSON output", func(t *testing.T) {
194
+ t.Parallel()
195
+ node := harness.NewT(t).NewNode().Init()
196
+ node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
197
+ node.SetIPFSConfig("Provide.Enabled", true)
198
+
199
+ node.StartDaemon()
200
+ node.StopDaemon()
201
+
202
+ res := node.IPFS("diag", "datastore", "count", "/provider/keystore/0/", "--enc=json")
203
+ assert.NoError(t, res.Err)
204
+
205
+ var result struct {
206
+ Prefix string `json:"prefix"`
207
+ Count int64 `json:"count"`
208
+ }
209
+ err := json.Unmarshal(res.Stdout.Bytes(), &result)
210
+ require.NoError(t, err)
211
+ assert.Equal(t, "/provider/keystore/0/", result.Prefix)
212
+ assert.GreaterOrEqual(t, result.Count, int64(0), "count should be non-negative")
213
+ })
214
+
215
+ t.Run("works without provider keystore", func(t *testing.T) {
216
+ t.Parallel()
217
+ node := harness.NewT(t).NewNode().Init()
218
+
219
+ // No sweep enabled, no provider-keystore dirs — should still work fine
220
+ count := node.DatastoreCount("/provider/keystore/0/")
221
+ assert.Zero(t, count)
222
+
223
+ count = node.DatastoreCount("/")
224
+ assert.Greater(t, count, int64(0))
225
+ })
226
}
test/cli/harness/node.go
+6
@@ -739,6 +739,12 @@ func (n *Node) DatastoreCount(prefix string) int64 {
739
return count
740
}
741
742
+// DatastorePut writes a key-value pair to the datastore.
743
+// Requires the daemon to be stopped.
744
+func (n *Node) DatastorePut(key, value string) {
745
+ n.IPFS("diag", "datastore", "put", key, value)
746
+}
747
+
748
// DatastoreGet retrieves the value at the given key.
749
// Requires the daemon to be stopped. Returns nil if key not found.
750
func (n *Node) DatastoreGet(key string) []byte {
test/cli/provider_test.go
+120
@@ -6,6 +6,8 @@ import (
6
"fmt"
7
"net/http"
8
"net/http/httptest"
9
+ "os"
10
+ "path/filepath"
11
"strings"
12
"sync/atomic"
13
"testing"
@@ -842,3 +844,121 @@ func TestHTTPOnlyProviderWithSweepEnabled(t *testing.T) {
844
assert.Contains(t, statRes.Stdout.String(), "TotalReprovides:",
845
"should show legacy provider stats")
846
}
847
+
848
+// TestProviderKeystoreDatastoreCompaction verifies that the SweepingProvider's
849
+// keystore uses a datastore factory that creates separate physical datastores
850
+// and reclaims disk space by deleting old datastores after each reset cycle.
851
+//
852
+// The keystore uses two alternating namespaces ("0" and "1") plus a "meta"
853
+// namespace. The lifecycle is:
854
+// 1. First start: namespace "0" is created as the initial active datastore
855
+// 2. First reset (keystore sync at startup): "1" is created, data is written,
856
+// namespaces swap, "0" is destroyed from disk via os.RemoveAll
857
+// 3. Restart: "1" and "meta" survive on disk
858
+// 4. Second reset: "0" is recreated, namespaces swap, "1" is destroyed
859
+func TestProviderKeystoreDatastorePurge(t *testing.T) {
860
+ t.Parallel()
861
+
862
+ h := harness.NewT(t)
863
+ node := h.NewNode().Init()
864
+ node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
865
+ node.SetIPFSConfig("Provide.Enabled", true)
866
+ node.SetIPFSConfig("Bootstrap", []string{})
867
+
868
+ // Add content offline so the keystore has something to sync on startup.
869
+ for i := range 5 {
870
+ node.IPFSAddStr(fmt.Sprintf("keystore-compaction-test-%d", i))
871
+ }
872
+
873
+ keystoreBase := filepath.Join(node.Dir, "provider-keystore")
874
+ ns0 := filepath.Join(keystoreBase, "0")
875
+ ns1 := filepath.Join(keystoreBase, "1")
876
+
877
+ // Directory should not exist before starting the daemon.
878
+ _, err := os.Stat(keystoreBase)
879
+ require.True(t, os.IsNotExist(err), "provider-keystore should not exist before daemon start")
880
+
881
+ // --- First start: triggers keystore sync (ResetCids) ---
882
+ // Init creates "0", then reset swaps to "1" and destroys "0".
883
+ node.StartDaemon()
884
+
885
+ require.Eventually(t, func() bool {
886
+ return dirExists(ns1) && !dirExists(ns0)
887
+ }, 30*time.Second, 200*time.Millisecond,
888
+ "after first reset: ns1 should exist, ns0 should be destroyed")
889
+
890
+ // --- Restart: triggers a second keystore sync (ResetCids) ---
891
+ // Reset swaps back to "0" and destroys "1".
892
+ node.StopDaemon()
893
+
894
+ // Between restarts: ns1 survives on disk, ns0 does not.
895
+ assert.True(t, dirExists(ns1), "ns1 should survive shutdown")
896
+ assert.False(t, dirExists(ns0), "ns0 should not reappear between restarts")
897
+
898
+ node.StartDaemon()
899
+
900
+ require.Eventually(t, func() bool {
901
+ return dirExists(ns0) && !dirExists(ns1)
902
+ }, 30*time.Second, 200*time.Millisecond,
903
+ "after second reset: ns0 should exist, ns1 should be destroyed")
904
+
905
+ node.StopDaemon()
906
+}
907
+
908
+// TestProviderKeystoreMigrationPurge verifies that orphaned keystore data
909
+// left in the shared repo datastore by older Kubo versions is purged on
910
+// the first sweep-enabled daemon start. The migration is triggered by the
911
+// absence of the <repo>/provider-keystore/ directory.
912
+func TestProviderKeystoreMigrationPurge(t *testing.T) {
913
+ t.Parallel()
914
+
915
+ h := harness.NewT(t)
916
+ node := h.NewNode().Init()
917
+ node.SetIPFSConfig("Provide.DHT.SweepEnabled", true)
918
+ node.SetIPFSConfig("Provide.Enabled", true)
919
+ node.SetIPFSConfig("Bootstrap", []string{})
920
+
921
+ keystoreBase := filepath.Join(node.Dir, "provider-keystore")
922
+
923
+ // Pre-seed orphaned keystore data into the shared datastore, simulating
924
+ // the layout produced by older Kubo that stored keystore entries inline.
925
+ const numOrphans = 10
926
+ for i := range numOrphans {
927
+ node.DatastorePut(
928
+ fmt.Sprintf("/provider/keystore/%d/fake-key-%d", i%2, i),
929
+ fmt.Sprintf("orphan-%d", i),
930
+ )
931
+ }
932
+
933
+ // The orphaned keys should be visible via diag datastore.
934
+ count := node.DatastoreCount("/provider/keystore/")
935
+ require.Equal(t, int64(numOrphans), count, "orphaned keys should be present before migration")
936
+
937
+ // The provider-keystore directory must not exist yet (its absence
938
+ // triggers the migration).
939
+ require.False(t, dirExists(keystoreBase),
940
+ "provider-keystore/ should not exist before first sweep-enabled start")
941
+
942
+ // Start the daemon: this triggers the one-time migration purge.
943
+ node.StartDaemon()
944
+ node.StopDaemon()
945
+
946
+ // After migration the seeded orphaned keys should be gone from the
947
+ // shared datastore. The diag datastore count command mounts the
948
+ // separate provider-keystore datastores, so we check for the specific
949
+ // fake keys we seeded to confirm they were purged.
950
+ for i := range numOrphans {
951
+ key := fmt.Sprintf("/provider/keystore/%d/fake-key-%d", i%2, i)
952
+ assert.False(t, node.DatastoreHasKey(key),
953
+ "orphaned key %s should be purged after migration", key)
954
+ }
955
+
956
+ // The provider-keystore directory should now exist.
957
+ assert.True(t, dirExists(keystoreBase),
958
+ "provider-keystore/ should exist after sweep-enabled daemon ran")
959
+}
960
+
961
+func dirExists(path string) bool {
962
+ info, err := os.Stat(path)
963
+ return err == nil && info.IsDir()
964
+}
test/dependencies/go.mod
+1
-1
@@ -183,7 +183,7 @@ require (
183
github.com/libp2p/go-flow-metrics v0.3.0 // indirect
184
github.com/libp2p/go-libp2p v0.47.0 // indirect
185
github.com/libp2p/go-libp2p-asn-util v0.4.1 // indirect
186
- github.com/libp2p/go-libp2p-kad-dht v0.38.0 // indirect
186
+ github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29 // indirect
187
github.com/libp2p/go-libp2p-kbucket v0.8.0 // indirect
188
github.com/libp2p/go-libp2p-record v0.3.1 // indirect
189
github.com/libp2p/go-libp2p-routing-helpers v0.7.5 // indirect
test/dependencies/go.sum
+2
-2
@@ -578,8 +578,8 @@ github.com/libp2p/go-libp2p v0.47.0 h1:qQpBjSCWNQFF0hjBbKirMXE9RHLtSuzTDkTfr1rw0
578
github.com/libp2p/go-libp2p v0.47.0/go.mod h1:s8HPh7mMV933OtXzONaGFseCg/BE//m1V34p3x4EUOY=
579
github.com/libp2p/go-libp2p-asn-util v0.4.1 h1:xqL7++IKD9TBFMgnLPZR6/6iYhawHKHl950SO9L6n94=
580
github.com/libp2p/go-libp2p-asn-util v0.4.1/go.mod h1:d/NI6XZ9qxw67b4e+NgpQexCIiFYJjErASrYW4PFDN8=
581
-github.com/libp2p/go-libp2p-kad-dht v0.38.0 h1:NToFzwvICo6ghDfSwuTmROCtl9LDXSZT1VawEbm4NUs=
582
-github.com/libp2p/go-libp2p-kad-dht v0.38.0/go.mod h1:g/CefQilAnCMyUH52A6tUGbe17NgQ8q26MaZCA968iI=
581
+github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29 h1:ZSi0kAWeDUeDPnoiVZ75Hyun7+wksJpQxFiz6aWNrys=
582
+github.com/libp2p/go-libp2p-kad-dht v0.38.1-0.20260319095041-7ba6b28e4b29/go.mod h1:g/CefQilAnCMyUH52A6tUGbe17NgQ8q26MaZCA968iI=
583
github.com/libp2p/go-libp2p-kbucket v0.8.0 h1:QAK7RzKJpYe+EuSEATAaaHYMYLkPDGC18m9jxPLnU8s=
584
github.com/libp2p/go-libp2p-kbucket v0.8.0/go.mod h1:JMlxqcEyKwO6ox716eyC0hmiduSWZZl6JY93mGaaqc4=
585
github.com/libp2p/go-libp2p-record v0.3.1 h1:cly48Xi5GjNw5Wq+7gmjfBiG9HCzQVkiZOUZ8kUl+Fg=