refactor: remove goprocess (#10872)
* refactor: remove goprocess The `goprocess` package is no longer needed. It can be replaces by modern `context` and `context.AfterFunc`. * mod tidy * log unmount errors on shutdown * Do not log non-mounted errors on shutdown * Use WaitGroup associated with IPFS node to wait for services to whutdown * Prefer explicit Close to context.ArterFunc * Do not use node-level WaitGroup * Unmount for non-supported platforms * fix return values * test: daemon shuts down gracefully make sure ongoing operations dont block shutdown * test(cli): add TestFUSE * test: smarter RequiresFUSE opportunistically run FUSE tests if env has fusermount and TEST_FUSE was not explicitly set * docs: changelog --------- Co-authored-by: gammazero <gammazero@users.noreply.github.com> Co-authored-by: Marcin Rataj <lidel@lidel.org>
Andrew Gillis committed
Aug 5, 2025 at 15:33 UTC
90b73d2ad2d74bf85983b8bb9d1178c92c88b55c
23 files changed
+450
-95
cmd/ipfs/kubo/daemon.go
+33
-16
@@ -35,7 +35,6 @@ import (
35
fsrepo "github.com/ipfs/kubo/repo/fsrepo"
36
"github.com/ipfs/kubo/repo/fsrepo/migrations"
37
"github.com/ipfs/kubo/repo/fsrepo/migrations/ipfsfetcher"
38
- goprocess "github.com/jbenet/goprocess"
38
p2pcrypto "github.com/libp2p/go-libp2p/core/crypto"
39
pnet "github.com/libp2p/go-libp2p/core/pnet"
40
"github.com/libp2p/go-libp2p/core/protocol"
@@ -537,10 +536,19 @@ take effect.
536
if err != nil {
537
return err
538
}
539
+
540
+ pluginErrc := make(chan error, 1)
541
select {
541
- case <-node.Process.Closing():
542
+ case <-node.Context().Done():
543
+ close(pluginErrc)
544
default:
543
- node.Process.AddChild(goprocess.WithTeardown(cctx.Plugins.Close))
545
+ context.AfterFunc(node.Context(), func() {
546
+ err := cctx.Plugins.Close()
547
+ if err != nil {
548
+ pluginErrc <- fmt.Errorf("closing plugins: %w", err)
549
+ }
550
+ close(pluginErrc)
551
+ })
552
}
553
554
// construct api endpoint - every time
@@ -558,6 +566,11 @@ take effect.
566
if err := mountFuse(req, cctx); err != nil {
567
return err
568
}
569
+ defer func() {
570
+ if _err != nil {
571
+ nodeMount.Unmount(node)
572
+ }
573
+ }()
574
}
575
576
// repo blockstore GC - if --enable-gc flag is present
@@ -703,10 +716,17 @@ take effect.
716
log.Fatal("Support for IPFS_REUSEPORT was removed. Use LIBP2P_TCP_REUSEPORT instead.")
717
}
718
719
+ unmountErrc := make(chan error)
720
+ context.AfterFunc(node.Context(), func() {
721
+ <-node.Context().Done()
722
+ nodeMount.Unmount(node)
723
+ close(unmountErrc)
724
+ })
725
+
726
// collect long-running errors and block for shutdown
727
// TODO(cryptix): our fuse currently doesn't follow this pattern for graceful shutdown
728
var errs error
709
- for err := range merge(apiErrc, gwErrc, gcErrc, p2pGwErrc) {
729
+ for err := range merge(apiErrc, gwErrc, gcErrc, p2pGwErrc, pluginErrc, unmountErrc) {
730
if err != nil {
731
errs = multierr.Append(errs, err)
732
}
@@ -1053,14 +1073,13 @@ func serveTrustlessGatewayOverLibp2p(cctx *oldcmds.Context) (<-chan error, error
1073
1074
errc := make(chan error, 1)
1075
go func() {
1056
- defer close(errc)
1076
errc <- h.Serve()
1077
+ close(errc)
1078
}()
1079
1060
- go func() {
1061
- <-node.Process.Closing()
1080
+ context.AfterFunc(node.Context(), func() {
1081
h.Close()
1063
- }()
1082
+ })
1083
1084
return errc, nil
1085
}
@@ -1145,14 +1164,14 @@ func maybeRunGC(req *cmds.Request, node *core.IpfsNode) (<-chan error, error) {
1164
return errc, nil
1165
}
1166
1148
-// merge does fan-in of multiple read-only error channels
1149
-// taken from http://blog.golang.org/pipelines
1167
+// merge does fan-in of multiple read-only error channels.
1168
func merge(cs ...<-chan error) <-chan error {
1169
var wg sync.WaitGroup
1170
out := make(chan error)
1171
1154
- // Start an output goroutine for each input channel in cs. output
1155
- // copies values from c to out until c is closed, then calls wg.Done.
1172
+ // Start a goroutine for each input channel in cs, that copies values from
1173
+ // the input channel to the output channel until the input channel is
1174
+ // closed.
1175
output := func(c <-chan error) {
1176
for n := range c {
1177
out <- n
@@ -1166,8 +1185,8 @@ func merge(cs ...<-chan error) <-chan error {
1185
}
1186
}
1187
1169
- // Start a goroutine to close out once all the output goroutines are
1170
- // done. This must start after the wg.Add call.
1188
+ // Start a goroutine to close out once all the output goroutines, and other
1189
+ // things to wait on, are done.
1190
go func() {
1191
wg.Wait()
1192
close(out)
@@ -1238,8 +1257,6 @@ Visit https://github.com/ipfs/kubo/releases or https://dist.ipfs.tech/#kubo and
1257
select {
1258
case <-ctx.Done():
1259
return
1241
- case <-nd.Process.Closing():
1242
- return
1260
case <-ticker.C:
1261
continue
1262
}
cmd/ipfswatch/main.go
+4
-6
@@ -21,7 +21,6 @@ import (
21
22
fsnotify "github.com/fsnotify/fsnotify"
23
"github.com/ipfs/boxo/files"
24
- process "github.com/jbenet/goprocess"
24
)
25
26
var (
@@ -54,7 +53,6 @@ func main() {
53
}
54
55
func run(ipfsPath, watchPath string) error {
57
- proc := process.WithParent(process.Background())
56
log.Printf("running IPFSWatch on '%s' using repo at '%s'...", watchPath, ipfsPath)
57
58
ipfsPath, err := fsutil.ExpandHome(ipfsPath)
@@ -99,11 +97,11 @@ func run(ipfsPath, watchPath string) error {
97
corehttp.WebUIOption,
98
corehttp.CommandsOption(cmdCtx(node, ipfsPath)),
99
}
102
- proc.Go(func(p process.Process) {
100
+ go func() {
101
if err := corehttp.ListenAndServe(node, addr, opts...); err != nil {
102
return
103
}
106
- })
104
+ }()
105
}
106
107
interrupts := make(chan os.Signal, 1)
@@ -137,7 +135,7 @@ func run(ipfsPath, watchPath string) error {
135
}
136
}
137
}
140
- proc.Go(func(p process.Process) {
138
+ go func() {
139
file, err := os.Open(e.Name)
140
if err != nil {
141
log.Println(err)
@@ -162,7 +160,7 @@ func run(ipfsPath, watchPath string) error {
160
log.Println(err)
161
}
162
log.Printf("added %s... key: %s", e.Name, k)
165
- })
163
+ }()
164
}
165
case err := <-watcher.Errors:
166
log.Println(err)
core/core.go
+1
-3
@@ -29,7 +29,6 @@ import (
29
provider "github.com/ipfs/boxo/provider"
30
ipld "github.com/ipfs/go-ipld-format"
31
logging "github.com/ipfs/go-log/v2"
32
- goprocess "github.com/jbenet/goprocess"
32
ddht "github.com/libp2p/go-libp2p-kad-dht/dual"
33
pubsub "github.com/libp2p/go-libp2p-pubsub"
34
psrouter "github.com/libp2p/go-libp2p-pubsub-router"
@@ -119,8 +118,7 @@ type IpfsNode struct {
118
119
P2P *p2p.P2P `optional:"true"`
120
122
- Process goprocess.Process
123
- ctx context.Context
121
+ ctx context.Context
122
123
stop func() error
124
core/corehttp/corehttp.go
+20
-13
@@ -13,8 +13,6 @@ import (
13
14
logging "github.com/ipfs/go-log/v2"
15
core "github.com/ipfs/kubo/core"
16
- "github.com/jbenet/goprocess"
17
- periodicproc "github.com/jbenet/goprocess/periodic"
16
ma "github.com/multiformats/go-multiaddr"
17
manet "github.com/multiformats/go-multiaddr/net"
18
)
@@ -97,7 +95,7 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
95
}
96
97
select {
100
- case <-node.Process.Closing():
98
+ case <-node.Context().Done():
99
return fmt.Errorf("failed to start server, process closing")
100
default:
101
}
@@ -107,20 +105,31 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
105
}
106
107
var serverError error
110
- serverProc := node.Process.Go(func(p goprocess.Process) {
108
+ serverClosed := make(chan struct{})
109
+ go func() {
110
serverError = server.Serve(lis)
112
- })
111
+ close(serverClosed)
112
+ }()
113
114
// wait for server to exit.
115
select {
116
- case <-serverProc.Closed():
116
+ case <-serverClosed:
117
// if node being closed before server exits, close server
118
- case <-node.Process.Closing():
118
+ case <-node.Context().Done():
119
log.Infof("server at %s terminating...", addr)
120
121
- warnProc := periodicproc.Tick(5*time.Second, func(_ goprocess.Process) {
122
- log.Infof("waiting for server at %s to terminate...", addr)
123
- })
121
+ go func() {
122
+ ticker := time.NewTicker(5 * time.Second)
123
+ defer ticker.Stop()
124
+ for {
125
+ select {
126
+ case <-ticker.C:
127
+ log.Infof("waiting for server at %s to terminate...", addr)
128
+ case <-serverClosed:
129
+ return
130
+ }
131
+ }
132
+ }()
133
134
// This timeout shouldn't be necessary if all of our commands
135
// are obeying their contexts but we should have *some* timeout.
@@ -130,10 +139,8 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
139
140
// Should have already closed but we still need to wait for it
141
// to set the error.
133
- <-serverProc.Closed()
142
+ <-serverClosed
143
serverError = err
135
-
136
- warnProc.Close()
144
}
145
146
log.Infof("server at %s terminated", addr)
core/node/groups.go
-2
@@ -445,8 +445,6 @@ func IPFS(ctx context.Context, bcfg *BuildCfg) fx.Option {
445
return fx.Options(
446
bcfgOpts,
447
448
- fx.Provide(baseProcess),
449
-
448
Storage(bcfg, cfg),
449
Identity(cfg),
450
IPNS,
core/node/helpers.go
-12
@@ -4,7 +4,6 @@ import (
4
"context"
5
"errors"
6
7
- "github.com/jbenet/goprocess"
7
"go.uber.org/fx"
8
)
9
@@ -55,14 +54,3 @@ func maybeInvoke(opt interface{}, enable bool) fx.Option {
54
}
55
return fx.Options()
56
}
58
-
59
-// baseProcess creates a goprocess which is closed when the lifecycle signals it to stop
60
-func baseProcess(lc fx.Lifecycle) goprocess.Process {
61
- p := goprocess.WithParent(goprocess.Background())
62
- lc.Append(fx.Hook{
63
- OnStop: func(_ context.Context) error {
64
- return p.Close()
65
- },
66
- })
67
- return p
68
-}
docs/changelogs/v0.37.md
+7
-4
@@ -11,7 +11,7 @@ This release was brought to you by the [Interplanetary Shipyard](https://ipship
11
- [Overview](#overview)
12
- [🔦 Highlights](#-highlights)
13
- [Clear provide queue when reprovide strategy changes](#clear-provide-queue-when-reprovide-strategy-changes)
14
- - [Remove unnecessary packages from thirdparty](#remove-unnecessary-packages-from-thirdparty)
14
+ - [Removed unnecessary dependencies](#removed-unnecessary-dependencies)
15
- [📦️ Important dependency updates](#-important-dependency-updates)
16
- [📝 Changelog](#-changelog)
17
- [👨👩👧👦 Contributors](#-contributors)
@@ -31,13 +31,16 @@ A new `ipfs provide clear` command also allows manual queue clearing for debuggi
31
> [!NOTE]
32
> Upgrading to Kubo 0.37 will automatically clear any preexisting provide queue. The next time `Reprovider.Interval` hits, `Reprovider.Strategy` will be executed on a clean slate, ensuring consistent behavior with your current configuration.
33
34
-#### Remove unnecessary packages from thirdparty
34
+#### Removed unnecessary dependencies
35
36
-Removed unnecessary packages from the `thirdparty` area of kubo repositroy.
36
+Kubo has been cleaned up by removing unnecessary dependencies and packages:
37
38
- Removed `thirdparty/assert` (replaced by `github.com/stretchr/testify/require`)
39
-- Removed `thirdparty/dir` (replaced by `misc/fsutil)`
39
+- Removed `thirdparty/dir` (replaced by `misc/fsutil`)
40
- Removed `thirdparty/notifier` (unused)
41
+- Removed `goprocess` dependency (replaced with native Go `context` patterns)
42
+
43
+These changes reduce the dependency footprint while improving code maintainability and following Go best practices.
44
45
#### 📦️ Important dependency updates
46
docs/examples/kubo-as-a-library/go.mod
-1
@@ -102,7 +102,6 @@ require (
102
github.com/ipshipyard/p2p-forge v0.6.1 // indirect
103
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
104
github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
105
- github.com/jbenet/goprocess v0.1.4 // indirect
105
github.com/klauspost/compress v1.18.0 // indirect
106
github.com/klauspost/cpuid/v2 v2.2.10 // indirect
107
github.com/koron/go-ssdp v0.0.6 // indirect
docs/examples/kubo-as-a-library/go.sum
-2
@@ -377,8 +377,6 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo
377
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
378
github.com/jbenet/goprocess v0.0.0-20160826012719-b497e2f366b8/go.mod h1:Ly/wlsjFq/qrU3Rar62tu1gASgGw6chQbSh/XgIIXCY=
379
github.com/jbenet/goprocess v0.1.3/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
380
-github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0o=
381
-github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
380
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
381
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
382
github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
fuse/ipns/mount_unix.go
+1
-1
@@ -29,5 +29,5 @@ func Mount(ipfs *core.IpfsNode, ipnsmp, ipfsmp string) (mount.Mount, error) {
29
return nil, err
30
}
31
32
- return mount.NewMount(ipfs.Process, fsys, ipnsmp, allowOther)
32
+ return mount.NewMount(fsys, ipnsmp, allowOther)
33
}
fuse/mfs/mount_unix.go
+1
-1
@@ -17,5 +17,5 @@ func Mount(ipfs *core.IpfsNode, mountpoint string) (mount.Mount, error) {
17
}
18
allowOther := cfg.Mounts.FuseAllowOther
19
fsys := NewFileSystem(ipfs)
20
- return mount.NewMount(ipfs.Process, fsys, mountpoint, allowOther)
20
+ return mount.NewMount(fsys, mountpoint, allowOther)
21
}
fuse/mount/fuse.go
+10
-13
@@ -11,7 +11,6 @@ import (
11
12
"bazil.org/fuse"
13
"bazil.org/fuse/fs"
14
- "github.com/jbenet/goprocess"
14
)
15
16
var ErrNotMounted = errors.New("not mounted")
@@ -25,12 +24,12 @@ type mount struct {
24
active bool
25
activeLock *sync.RWMutex
26
28
- proc goprocess.Process
27
+ unmountOnce sync.Once
28
}
29
30
// Mount mounts a fuse fs.FS at a given location, and returns a Mount instance.
32
-// parent is a ContextGroup to bind the mount's ContextGroup to.
33
-func NewMount(p goprocess.Process, fsys fs.FS, mountpoint string, allowOther bool) (Mount, error) {
31
+// ctx is parent is a ContextGroup to bind the mount's ContextGroup to.
32
+func NewMount(fsys fs.FS, mountpoint string, allowOther bool) (Mount, error) {
33
var conn *fuse.Conn
34
var err error
35
@@ -54,12 +53,10 @@ func NewMount(p goprocess.Process, fsys fs.FS, mountpoint string, allowOther boo
53
filesys: fsys,
54
active: false,
55
activeLock: &sync.RWMutex{},
57
- proc: goprocess.WithParent(p), // link it to parent.
56
}
59
- m.proc.SetTeardown(m.unmount)
57
58
// launch the mounting process.
62
- if err := m.mount(); err != nil {
59
+ if err = m.mount(); err != nil {
60
_ = m.Unmount() // just in case.
61
return nil, err
62
}
@@ -135,10 +132,6 @@ func (m *mount) unmount() error {
132
return nil
133
}
134
138
-func (m *mount) Process() goprocess.Process {
139
- return m.proc
140
-}
141
-
135
func (m *mount) MountPoint() string {
136
return m.mpoint
137
}
@@ -148,8 +141,12 @@ func (m *mount) Unmount() error {
141
return ErrNotMounted
142
}
143
151
- // call Process Close(), which calls unmount() exactly once.
152
- return m.proc.Close()
144
+ var err error
145
+ m.unmountOnce.Do(func() {
146
+ err = m.unmount()
147
+ })
148
+
149
+ return err
150
}
151
152
func (m *mount) IsActive() bool {
fuse/mount/mount.go
-5
@@ -9,7 +9,6 @@ import (
9
"time"
10
11
logging "github.com/ipfs/go-log/v2"
12
- goprocess "github.com/jbenet/goprocess"
12
)
13
14
var log = logging.Logger("mount")
@@ -26,10 +25,6 @@ type Mount interface {
25
26
// Checks if the mount is still active.
27
IsActive() bool
29
-
30
- // Process returns the mount's Process to be able to link it
31
- // to other processes. Unmount upon closing.
32
- Process() goprocess.Process
28
}
29
30
// ForceUnmount attempts to forcibly unmount a given mount.
fuse/node/mount_nofuse.go
+4
@@ -12,3 +12,7 @@ import (
12
func Mount(node *core.IpfsNode, fsdir, nsdir, mfsdir string) error {
13
return errors.New("not compiled in")
14
}
15
+
16
+func Unmount(node *core.IpfsNode) {
17
+ return
18
+}
fuse/node/mount_notsupp.go
+4
@@ -12,3 +12,7 @@ import (
12
func Mount(node *core.IpfsNode, fsdir, nsdir, mfsdir string) error {
13
return errors.New("FUSE not supported on OpenBSD or NetBSD. See #5334 (https://github.com/ipfs/kubo/issues/5334).")
14
}
15
+
16
+func Unmount(node *core.IpfsNode) {
17
+ return
18
+}
fuse/node/mount_unix.go
+19
-9
@@ -36,24 +36,34 @@ func Mount(node *core.IpfsNode, fsdir, nsdir, mfsdir string) error {
36
// check if we already have live mounts.
37
// if the user said "Mount", then there must be something wrong.
38
// so, close them and try again.
39
+ Unmount(node)
40
+
41
+ if err := platformFuseChecks(node); err != nil {
42
+ return err
43
+ }
44
+
45
+ return doMount(node, fsdir, nsdir, mfsdir)
46
+}
47
+
48
+func Unmount(node *core.IpfsNode) {
49
if node.Mounts.Ipfs != nil && node.Mounts.Ipfs.IsActive() {
50
// best effort
41
- _ = node.Mounts.Ipfs.Unmount()
51
+ if err := node.Mounts.Ipfs.Unmount(); err != nil {
52
+ log.Errorf("error unmounting IPFS: %s", err)
53
+ }
54
}
55
if node.Mounts.Ipns != nil && node.Mounts.Ipns.IsActive() {
56
// best effort
45
- _ = node.Mounts.Ipns.Unmount()
57
+ if err := node.Mounts.Ipns.Unmount(); err != nil {
58
+ log.Errorf("error unmounting IPNS: %s", err)
59
+ }
60
}
61
if node.Mounts.Mfs != nil && node.Mounts.Mfs.IsActive() {
62
// best effort
49
- _ = node.Mounts.Mfs.Unmount()
50
- }
51
-
52
- if err := platformFuseChecks(node); err != nil {
53
- return err
63
+ if err := node.Mounts.Mfs.Unmount(); err != nil {
64
+ log.Errorf("error unmounting MFS: %s", err)
65
+ }
66
}
55
-
56
- return doMount(node, fsdir, nsdir, mfsdir)
67
}
68
69
func doMount(node *core.IpfsNode, fsdir, nsdir, mfsdir string) error {
fuse/node/mount_windows.go
+6
@@ -9,3 +9,9 @@ func Mount(node *core.IpfsNode, fsdir, nsdir, mfsdir string) error {
9
// currently a no-op, but we don't want to return an error
10
return nil
11
}
12
+
13
+func Unmount(node *core.IpfsNode) {
14
+ // TODO
15
+ // currently a no-op
16
+ return
17
+}
fuse/readonly/mount_unix.go
+1
-1
@@ -17,5 +17,5 @@ func Mount(ipfs *core.IpfsNode, mountpoint string) (mount.Mount, error) {
17
}
18
allowOther := cfg.Mounts.FuseAllowOther
19
fsys := NewFileSystem(ipfs)
20
- return mount.NewMount(ipfs.Process, fsys, mountpoint, allowOther)
20
+ return mount.NewMount(fsys, mountpoint, allowOther)
21
}
go.mod
-1
@@ -49,7 +49,6 @@ require (
49
github.com/ipld/go-ipld-prime v0.21.0
50
github.com/ipshipyard/p2p-forge v0.6.1
51
github.com/jbenet/go-temp-err-catcher v0.1.0
52
- github.com/jbenet/goprocess v0.1.4
52
github.com/julienschmidt/httprouter v1.3.0
53
github.com/libp2p/go-doh-resolver v0.5.0
54
github.com/libp2p/go-libp2p v0.42.1
go.sum
-3
@@ -443,14 +443,11 @@ github.com/ipshipyard/p2p-forge v0.6.1 h1:987/hUC1YxI56CcMX6iTB+9BLjFV0d2SJnig9Z
443
github.com/ipshipyard/p2p-forge v0.6.1/go.mod h1:pj8Zcs+ex5OMq5a1bFLHqW0oL3qYO0v5eGLZmit0l7U=
444
github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
445
github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
446
-github.com/jbenet/go-cienv v0.1.0 h1:Vc/s0QbQtoxX8MwwSLWWh+xNNZvM3Lw7NsTcHrvvhMc=
446
github.com/jbenet/go-cienv v0.1.0/go.mod h1:TqNnHUmJgXau0nCzC7kXWeotg3J9W34CUv5Djy1+FlA=
447
github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABoLk/+KKHggpk=
448
github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk=
449
github.com/jbenet/goprocess v0.0.0-20160826012719-b497e2f366b8/go.mod h1:Ly/wlsjFq/qrU3Rar62tu1gASgGw6chQbSh/XgIIXCY=
450
github.com/jbenet/goprocess v0.1.3/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
452
-github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0o=
453
-github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4=
451
github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
452
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
453
github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
test/cli/daemon_test.go
+131
@@ -1,10 +1,20 @@
1
package cli
2
3
import (
4
+ "bytes"
5
+ "crypto/rand"
6
+ "fmt"
7
+ "io"
8
+ "net/http"
9
"os/exec"
10
"testing"
11
+ "time"
12
13
+ "github.com/ipfs/kubo/config"
14
"github.com/ipfs/kubo/test/cli/harness"
15
+ "github.com/multiformats/go-multiaddr"
16
+ manet "github.com/multiformats/go-multiaddr/net"
17
+ "github.com/stretchr/testify/require"
18
)
19
20
func TestDaemon(t *testing.T) {
@@ -22,4 +32,125 @@ func TestDaemon(t *testing.T) {
32
33
node.StopDaemon()
34
})
35
+
36
+ t.Run("daemon shuts down gracefully with active operations", func(t *testing.T) {
37
+ t.Parallel()
38
+
39
+ // Start daemon with multiple components active via config
40
+ node := harness.NewT(t).NewNode().Init()
41
+
42
+ // Enable experimental features and pubsub via config
43
+ node.UpdateConfig(func(cfg *config.Config) {
44
+ cfg.Pubsub.Enabled = config.True // Instead of --enable-pubsub-experiment
45
+ cfg.Experimental.P2pHttpProxy = true // Enable P2P HTTP proxy
46
+ cfg.Experimental.GatewayOverLibp2p = true // Enable gateway over libp2p
47
+ })
48
+
49
+ node.StartDaemon("--enable-gc")
50
+
51
+ // Start background operations to simulate real daemon workload:
52
+ // 1. "ipfs add" simulates content onboarding/ingestion work
53
+ // 2. Gateway request simulates content retrieval and gateway processing work
54
+
55
+ // Background operation 1: Continuous add of random data to simulate onboarding
56
+ addDone := make(chan struct{})
57
+ go func() {
58
+ defer close(addDone)
59
+
60
+ // Start the add command asynchronously
61
+ res := node.Runner.Run(harness.RunRequest{
62
+ Path: node.IPFSBin,
63
+ Args: []string{"add", "--progress=false", "-"},
64
+ RunFunc: (*exec.Cmd).Start,
65
+ CmdOpts: []harness.CmdOpt{
66
+ harness.RunWithStdin(&infiniteReader{}),
67
+ },
68
+ })
69
+
70
+ // Wait for command to finish (when daemon stops)
71
+ if res.Cmd != nil {
72
+ _ = res.Cmd.Wait() // Ignore error, expect command to be killed during shutdown
73
+ }
74
+ }()
75
+
76
+ // Background operation 2: Gateway CAR request to simulate retrieval work
77
+ gatewayDone := make(chan struct{})
78
+ go func() {
79
+ defer close(gatewayDone)
80
+
81
+ // First add a file sized to ensure gateway request takes ~1 minute
82
+ largeData := make([]byte, 512*1024) // 512KB of data
83
+ _, _ = rand.Read(largeData) // Always succeeds for crypto/rand
84
+ testCID := node.IPFSAdd(bytes.NewReader(largeData))
85
+
86
+ // Get gateway address from config
87
+ cfg := node.ReadConfig()
88
+ gatewayMaddr, err := multiaddr.NewMultiaddr(cfg.Addresses.Gateway[0])
89
+ if err != nil {
90
+ return
91
+ }
92
+ gatewayAddr, err := manet.ToNetAddr(gatewayMaddr)
93
+ if err != nil {
94
+ return
95
+ }
96
+
97
+ // Request CAR but slow reading to simulate heavy gateway load
98
+ gatewayURL := fmt.Sprintf("http://%s/ipfs/%s?format=car", gatewayAddr, testCID)
99
+
100
+ client := &http.Client{Timeout: 90 * time.Second}
101
+ resp, err := client.Get(gatewayURL)
102
+ if err == nil {
103
+ defer resp.Body.Close()
104
+ // Read response slowly: 512KB ÷ 1KB × 125ms = ~64 seconds (1+ minute) total
105
+ // This ensures operation is still active when we shutdown at 2 seconds
106
+ buf := make([]byte, 1024) // 1KB buffer
107
+ for {
108
+ if _, err := io.ReadFull(resp.Body, buf); err != nil {
109
+ return
110
+ }
111
+ time.Sleep(125 * time.Millisecond) // 125ms delay = ~64s total for 512KB
112
+ }
113
+ }
114
+ }()
115
+
116
+ // Let operations run for 2 seconds to ensure they're active
117
+ time.Sleep(2 * time.Second)
118
+
119
+ // Trigger graceful shutdown
120
+ shutdownStart := time.Now()
121
+ node.StopDaemon()
122
+ shutdownDuration := time.Since(shutdownStart)
123
+
124
+ // Verify clean shutdown:
125
+ // - Daemon should stop within reasonable time (not hang)
126
+ require.Less(t, shutdownDuration, 10*time.Second, "daemon should shut down within 10 seconds")
127
+
128
+ // Wait for background operations to complete (with timeout)
129
+ select {
130
+ case <-addDone:
131
+ // Good, add operation terminated
132
+ case <-time.After(5 * time.Second):
133
+ t.Error("add operation did not terminate within 5 seconds after daemon shutdown")
134
+ }
135
+
136
+ select {
137
+ case <-gatewayDone:
138
+ // Good, gateway operation terminated
139
+ case <-time.After(5 * time.Second):
140
+ t.Error("gateway operation did not terminate within 5 seconds after daemon shutdown")
141
+ }
142
+
143
+ // Verify we can restart with same repo (no lock issues)
144
+ node.StartDaemon()
145
+ node.StopDaemon()
146
+ })
147
+}
148
+
149
+// infiniteReader provides an infinite stream of random data
150
+type infiniteReader struct{}
151
+
152
+func (r *infiniteReader) Read(p []byte) (n int, err error) {
153
+ _, _ = rand.Read(p) // Always succeeds for crypto/rand
154
+ time.Sleep(50 * time.Millisecond) // Rate limit to simulate steady stream
155
+ return len(p), nil
156
}
test/cli/fuse_test.go
new
+166
@@ -0,0 +1,166 @@
1
+package cli
2
+
3
+import (
4
+ "os"
5
+ "os/exec"
6
+ "path/filepath"
7
+ "runtime"
8
+ "strings"
9
+ "testing"
10
+
11
+ "github.com/ipfs/kubo/test/cli/harness"
12
+ "github.com/ipfs/kubo/test/cli/testutils"
13
+ "github.com/stretchr/testify/require"
14
+)
15
+
16
+func TestFUSE(t *testing.T) {
17
+ testutils.RequiresFUSE(t)
18
+ t.Parallel()
19
+
20
+ t.Run("mount and unmount work correctly", func(t *testing.T) {
21
+ t.Parallel()
22
+
23
+ // Create a node and start daemon
24
+ node := harness.NewT(t).NewNode().Init()
25
+ node.StartDaemon()
26
+
27
+ // Create mount directories in the node's working directory
28
+ nodeDir := node.Dir
29
+ ipfsMount := filepath.Join(nodeDir, "ipfs")
30
+ ipnsMount := filepath.Join(nodeDir, "ipns")
31
+ mfsMount := filepath.Join(nodeDir, "mfs")
32
+
33
+ err := os.MkdirAll(ipfsMount, 0755)
34
+ require.NoError(t, err)
35
+ err = os.MkdirAll(ipnsMount, 0755)
36
+ require.NoError(t, err)
37
+ err = os.MkdirAll(mfsMount, 0755)
38
+ require.NoError(t, err)
39
+
40
+ // Ensure any existing mounts are cleaned up first
41
+ failOnError := false // mount points might not exist from previous runs
42
+ doUnmount(t, ipfsMount, failOnError)
43
+ doUnmount(t, ipnsMount, failOnError)
44
+ doUnmount(t, mfsMount, failOnError)
45
+
46
+ // Test mount operation
47
+ result := node.IPFS("mount", "-f", ipfsMount, "-n", ipnsMount, "-m", mfsMount)
48
+
49
+ // Verify mount output
50
+ expectedOutput := "IPFS mounted at: " + ipfsMount + "\n" +
51
+ "IPNS mounted at: " + ipnsMount + "\n" +
52
+ "MFS mounted at: " + mfsMount + "\n"
53
+ require.Equal(t, expectedOutput, result.Stdout.String())
54
+
55
+ // Test basic MFS functionality via FUSE mount
56
+ testFile := filepath.Join(mfsMount, "testfile")
57
+ testContent := "hello fuse world"
58
+
59
+ // Create file via FUSE mount
60
+ err = os.WriteFile(testFile, []byte(testContent), 0644)
61
+ require.NoError(t, err)
62
+
63
+ // Verify file appears in MFS via IPFS commands
64
+ result = node.IPFS("files", "ls", "/")
65
+ require.Contains(t, result.Stdout.String(), "testfile")
66
+
67
+ // Read content back via MFS FUSE mount
68
+ readContent, err := os.ReadFile(testFile)
69
+ require.NoError(t, err)
70
+ require.Equal(t, testContent, string(readContent))
71
+
72
+ // Get the CID of the MFS file
73
+ result = node.IPFS("files", "stat", "/testfile", "--format=<hash>")
74
+ fileCID := strings.TrimSpace(result.Stdout.String())
75
+ require.NotEmpty(t, fileCID, "should have a CID for the MFS file")
76
+
77
+ // Read the same content via IPFS FUSE mount using the CID
78
+ ipfsFile := filepath.Join(ipfsMount, fileCID)
79
+ ipfsContent, err := os.ReadFile(ipfsFile)
80
+ require.NoError(t, err)
81
+ require.Equal(t, testContent, string(ipfsContent), "content should match between MFS and IPFS mounts")
82
+
83
+ // Verify both FUSE mounts return identical data
84
+ require.Equal(t, readContent, ipfsContent, "MFS and IPFS FUSE mounts should return identical data")
85
+
86
+ // Test that mount directories cannot be removed while mounted
87
+ err = os.Remove(ipfsMount)
88
+ require.Error(t, err, "should not be able to remove mounted directory")
89
+
90
+ // Stop daemon - this should trigger automatic unmount via context cancellation
91
+ node.StopDaemon()
92
+
93
+ // Daemon shutdown should handle unmount synchronously via context.AfterFunc
94
+
95
+ // Verify directories can now be removed (indicating successful unmount)
96
+ err = os.Remove(ipfsMount)
97
+ require.NoError(t, err, "should be able to remove directory after unmount")
98
+ err = os.Remove(ipnsMount)
99
+ require.NoError(t, err, "should be able to remove directory after unmount")
100
+ err = os.Remove(mfsMount)
101
+ require.NoError(t, err, "should be able to remove directory after unmount")
102
+ })
103
+
104
+ t.Run("explicit unmount works", func(t *testing.T) {
105
+ t.Parallel()
106
+
107
+ node := harness.NewT(t).NewNode().Init()
108
+ node.StartDaemon()
109
+
110
+ // Create mount directories
111
+ nodeDir := node.Dir
112
+ ipfsMount := filepath.Join(nodeDir, "ipfs")
113
+ ipnsMount := filepath.Join(nodeDir, "ipns")
114
+ mfsMount := filepath.Join(nodeDir, "mfs")
115
+
116
+ err := os.MkdirAll(ipfsMount, 0755)
117
+ require.NoError(t, err)
118
+ err = os.MkdirAll(ipnsMount, 0755)
119
+ require.NoError(t, err)
120
+ err = os.MkdirAll(mfsMount, 0755)
121
+ require.NoError(t, err)
122
+
123
+ // Clean up any existing mounts
124
+ failOnError := false // mount points might not exist from previous runs
125
+ doUnmount(t, ipfsMount, failOnError)
126
+ doUnmount(t, ipnsMount, failOnError)
127
+ doUnmount(t, mfsMount, failOnError)
128
+
129
+ // Mount
130
+ node.IPFS("mount", "-f", ipfsMount, "-n", ipnsMount, "-m", mfsMount)
131
+
132
+ // Explicit unmount via platform-specific command
133
+ failOnError = true // test that explicit unmount works correctly
134
+ doUnmount(t, ipfsMount, failOnError)
135
+ doUnmount(t, ipnsMount, failOnError)
136
+ doUnmount(t, mfsMount, failOnError)
137
+
138
+ // Verify directories can be removed after explicit unmount
139
+ err = os.Remove(ipfsMount)
140
+ require.NoError(t, err)
141
+ err = os.Remove(ipnsMount)
142
+ require.NoError(t, err)
143
+ err = os.Remove(mfsMount)
144
+ require.NoError(t, err)
145
+
146
+ node.StopDaemon()
147
+ })
148
+}
149
+
150
+// doUnmount performs platform-specific unmount, similar to sharness do_umount
151
+// failOnError: if true, unmount errors cause test failure; if false, errors are ignored (useful for cleanup)
152
+func doUnmount(t *testing.T, mountPoint string, failOnError bool) {
153
+ t.Helper()
154
+ var cmd *exec.Cmd
155
+ if runtime.GOOS == "linux" {
156
+ // fusermount -u: unmount filesystem (strict - fails if busy)
157
+ cmd = exec.Command("fusermount", "-u", mountPoint)
158
+ } else {
159
+ cmd = exec.Command("umount", mountPoint)
160
+ }
161
+
162
+ err := cmd.Run()
163
+ if err != nil && failOnError {
164
+ t.Fatalf("failed to unmount %s: %v", mountPoint, err)
165
+ }
166
+}
test/cli/testutils/requires.go
+42
-2
@@ -2,6 +2,7 @@ package testutils
2
3
import (
4
"os"
5
+ "os/exec"
6
"runtime"
7
"testing"
8
)
@@ -13,9 +14,48 @@ func RequiresDocker(t *testing.T) {
14
}
15
16
func RequiresFUSE(t *testing.T) {
16
- if os.Getenv("TEST_FUSE") != "1" {
17
- t.SkipNow()
17
+ // Skip if FUSE tests are explicitly disabled
18
+ if os.Getenv("TEST_FUSE") == "0" {
19
+ t.Skip("FUSE tests disabled via TEST_FUSE=0")
20
+ }
21
+
22
+ // If TEST_FUSE=1 is set, always run (for backwards compatibility)
23
+ if os.Getenv("TEST_FUSE") == "1" {
24
+ return
25
+ }
26
+
27
+ // Auto-detect FUSE availability based on platform and tools
28
+ if !isFUSEAvailable(t) {
29
+ t.Skip("FUSE not available (no fusermount/umount found or unsupported platform)")
30
+ }
31
+}
32
+
33
+// isFUSEAvailable checks if FUSE is available on the current system
34
+func isFUSEAvailable(t *testing.T) bool {
35
+ t.Helper()
36
+
37
+ // Check platform support
38
+ switch runtime.GOOS {
39
+ case "linux", "darwin", "freebsd", "openbsd", "netbsd":
40
+ // These platforms potentially support FUSE
41
+ case "windows":
42
+ // Windows has limited FUSE support via WinFsp, but skip for now
43
+ return false
44
+ default:
45
+ // Unknown platform, assume no FUSE support
46
+ return false
47
}
48
+
49
+ // Check for required unmount tools
50
+ var unmountCmd string
51
+ if runtime.GOOS == "linux" {
52
+ unmountCmd = "fusermount"
53
+ } else {
54
+ unmountCmd = "umount"
55
+ }
56
+
57
+ _, err := exec.LookPath(unmountCmd)
58
+ return err == nil
59
}
60
61
func RequiresExpensive(t *testing.T) {