| 1 | package harness |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "io/fs" |
| 10 | "net/http" |
| 11 | "os" |
| 12 | "os/exec" |
| 13 | "path/filepath" |
| 14 | "runtime" |
| 15 | "strconv" |
| 16 | "strings" |
| 17 | "syscall" |
| 18 | "time" |
| 19 | |
| 20 | logging "github.com/ipfs/go-log/v2" |
| 21 | "github.com/ipfs/kubo/config" |
| 22 | serial "github.com/ipfs/kubo/config/serialize" |
| 23 | "github.com/libp2p/go-libp2p/core/peer" |
| 24 | rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager" |
| 25 | "github.com/multiformats/go-multiaddr" |
| 26 | manet "github.com/multiformats/go-multiaddr/net" |
| 27 | ) |
| 28 | |
| 29 | var log = logging.Logger("testharness") |
| 30 | |
| 31 | // Node is a single Kubo node. |
| 32 | // Each node has its own config and can run its own Kubo daemon. |
| 33 | type Node struct { |
| 34 | ID int |
| 35 | Dir string |
| 36 | |
| 37 | APIListenAddr multiaddr.Multiaddr |
| 38 | GatewayListenAddr multiaddr.Multiaddr |
| 39 | SwarmAddr multiaddr.Multiaddr |
| 40 | EnableMDNS bool |
| 41 | |
| 42 | IPFSBin string |
| 43 | Runner *Runner |
| 44 | |
| 45 | Daemon *RunResult |
| 46 | } |
| 47 | |
| 48 | func BuildNode(ipfsBin, baseDir string, id int) *Node { |
| 49 | dir := filepath.Join(baseDir, strconv.Itoa(id)) |
| 50 | if err := os.MkdirAll(dir, 0o755); err != nil { |
| 51 | panic(err) |
| 52 | } |
| 53 | |
| 54 | env := environToMap(os.Environ()) |
| 55 | env["IPFS_PATH"] = dir |
| 56 | |
| 57 | // If using "ipfs" binary name, provide helpful binary information |
| 58 | if ipfsBin == "ipfs" { |
| 59 | // Check if cmd/ipfs/ipfs exists (simple relative path check) |
| 60 | localBinary := "cmd/ipfs/ipfs" |
| 61 | localExists := false |
| 62 | if _, err := os.Stat(localBinary); err == nil { |
| 63 | localExists = true |
| 64 | if abs, err := filepath.Abs(localBinary); err == nil { |
| 65 | localBinary = abs |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | // Check if ipfs is available in PATH |
| 70 | pathBinary, pathErr := exec.LookPath("ipfs") |
| 71 | |
| 72 | // Handle different scenarios |
| 73 | if pathErr != nil { |
| 74 | // No ipfs in PATH |
| 75 | if localExists { |
| 76 | fmt.Printf("WARNING: No 'ipfs' found in PATH, but local binary exists at %s\n", localBinary) |
| 77 | fmt.Printf("Consider adding it to PATH or run: export PATH=\"$(pwd)/cmd/ipfs:$PATH\"\n") |
| 78 | } else { |
| 79 | fmt.Printf("ERROR: No 'ipfs' binary found in PATH and no local build at cmd/ipfs/ipfs\n") |
| 80 | fmt.Printf("Run 'make build' first or install ipfs and add it to PATH\n") |
| 81 | panic("ipfs binary not available") |
| 82 | } |
| 83 | } else { |
| 84 | // ipfs found in PATH |
| 85 | if localExists && localBinary != pathBinary { |
| 86 | fmt.Printf("NOTE: Local binary at %s differs from PATH binary at %s\n", localBinary, pathBinary) |
| 87 | fmt.Printf("Consider adding the local binary to PATH if you want to use the version built by 'make build'\n") |
| 88 | } |
| 89 | // If they match or no local binary, no message needed |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | return &Node{ |
| 94 | ID: id, |
| 95 | Dir: dir, |
| 96 | IPFSBin: ipfsBin, |
| 97 | Runner: &Runner{ |
| 98 | Env: env, |
| 99 | Dir: dir, |
| 100 | }, |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | func (n *Node) WriteBytes(filename string, b []byte) { |
| 105 | f, err := os.Create(filepath.Join(n.Dir, filename)) |
| 106 | if err != nil { |
| 107 | panic(err) |
| 108 | } |
| 109 | defer f.Close() |
| 110 | _, err = io.Copy(f, bytes.NewReader(b)) |
| 111 | if err != nil { |
| 112 | panic(err) |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | // ReadFile reads the specific file. If it is relative, it is relative the node's root dir. |
| 117 | func (n *Node) ReadFile(filename string) string { |
| 118 | f := filename |
| 119 | if !filepath.IsAbs(filename) { |
| 120 | f = filepath.Join(n.Dir, filename) |
| 121 | } |
| 122 | b, err := os.ReadFile(f) |
| 123 | if err != nil { |
| 124 | panic(err) |
| 125 | } |
| 126 | return string(b) |
| 127 | } |
| 128 | |
| 129 | func (n *Node) ConfigFile() string { |
| 130 | return filepath.Join(n.Dir, "config") |
| 131 | } |
| 132 | |
| 133 | func (n *Node) ReadConfig() *config.Config { |
| 134 | cfg, err := serial.Load(filepath.Join(n.Dir, "config")) |
| 135 | if err != nil { |
| 136 | panic(err) |
| 137 | } |
| 138 | return cfg |
| 139 | } |
| 140 | |
| 141 | func (n *Node) WriteConfig(c *config.Config) { |
| 142 | err := serial.WriteConfigFile(filepath.Join(n.Dir, "config"), c) |
| 143 | if err != nil { |
| 144 | panic(err) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | func (n *Node) UpdateConfig(f func(cfg *config.Config)) { |
| 149 | cfg := n.ReadConfig() |
| 150 | f(cfg) |
| 151 | n.WriteConfig(cfg) |
| 152 | } |
| 153 | |
| 154 | func (n *Node) ReadUserResourceOverrides() *rcmgr.PartialLimitConfig { |
| 155 | var r rcmgr.PartialLimitConfig |
| 156 | err := serial.ReadConfigFile(filepath.Join(n.Dir, "libp2p-resource-limit-overrides.json"), &r) |
| 157 | switch err { |
| 158 | case nil, serial.ErrNotInitialized: |
| 159 | return &r |
| 160 | default: |
| 161 | panic(err) |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | func (n *Node) WriteUserSuppliedResourceOverrides(c *rcmgr.PartialLimitConfig) { |
| 166 | err := serial.WriteConfigFile(filepath.Join(n.Dir, "libp2p-resource-limit-overrides.json"), c) |
| 167 | if err != nil { |
| 168 | panic(err) |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | func (n *Node) UpdateUserSuppliedResourceManagerOverrides(f func(overrides *rcmgr.PartialLimitConfig)) { |
| 173 | overrides := n.ReadUserResourceOverrides() |
| 174 | f(overrides) |
| 175 | n.WriteUserSuppliedResourceOverrides(overrides) |
| 176 | } |
| 177 | |
| 178 | func (n *Node) IPFS(args ...string) *RunResult { |
| 179 | res := n.RunIPFS(args...) |
| 180 | n.Runner.AssertNoError(res) |
| 181 | return res |
| 182 | } |
| 183 | |
| 184 | func (n *Node) PipeStrToIPFS(s string, args ...string) *RunResult { |
| 185 | return n.PipeToIPFS(strings.NewReader(s), args...) |
| 186 | } |
| 187 | |
| 188 | func (n *Node) PipeToIPFS(reader io.Reader, args ...string) *RunResult { |
| 189 | res := n.RunPipeToIPFS(reader, args...) |
| 190 | n.Runner.AssertNoError(res) |
| 191 | return res |
| 192 | } |
| 193 | |
| 194 | func (n *Node) RunPipeToIPFS(reader io.Reader, args ...string) *RunResult { |
| 195 | return n.Runner.Run(RunRequest{ |
| 196 | Path: n.IPFSBin, |
| 197 | Args: args, |
| 198 | CmdOpts: []CmdOpt{RunWithStdin(reader)}, |
| 199 | }) |
| 200 | } |
| 201 | |
| 202 | func (n *Node) RunIPFS(args ...string) *RunResult { |
| 203 | return n.Runner.Run(RunRequest{ |
| 204 | Path: n.IPFSBin, |
| 205 | Args: args, |
| 206 | }) |
| 207 | } |
| 208 | |
| 209 | // Init initializes and configures the IPFS node, after which it is ready to run. |
| 210 | func (n *Node) Init(ipfsArgs ...string) *Node { |
| 211 | n.Runner.MustRun(RunRequest{ |
| 212 | Path: n.IPFSBin, |
| 213 | Args: append([]string{"init"}, ipfsArgs...), |
| 214 | }) |
| 215 | |
| 216 | if n.SwarmAddr == nil { |
| 217 | swarmAddr, err := multiaddr.NewMultiaddr("/ip4/127.0.0.1/tcp/0") |
| 218 | if err != nil { |
| 219 | panic(err) |
| 220 | } |
| 221 | n.SwarmAddr = swarmAddr |
| 222 | } |
| 223 | |
| 224 | if n.APIListenAddr == nil { |
| 225 | apiAddr, err := multiaddr.NewMultiaddr("/ip4/127.0.0.1/tcp/0") |
| 226 | if err != nil { |
| 227 | panic(err) |
| 228 | } |
| 229 | n.APIListenAddr = apiAddr |
| 230 | } |
| 231 | |
| 232 | if n.GatewayListenAddr == nil { |
| 233 | gatewayAddr, err := multiaddr.NewMultiaddr("/ip4/127.0.0.1/tcp/0") |
| 234 | if err != nil { |
| 235 | panic(err) |
| 236 | } |
| 237 | n.GatewayListenAddr = gatewayAddr |
| 238 | } |
| 239 | |
| 240 | n.UpdateConfig(func(cfg *config.Config) { |
| 241 | cfg.Bootstrap = []string{} |
| 242 | cfg.Addresses.Swarm = []string{n.SwarmAddr.String()} |
| 243 | cfg.Addresses.API = []string{n.APIListenAddr.String()} |
| 244 | cfg.Addresses.Gateway = []string{n.GatewayListenAddr.String()} |
| 245 | cfg.Swarm.DisableNatPortMap = true |
| 246 | cfg.Discovery.MDNS.Enabled = n.EnableMDNS |
| 247 | cfg.Routing.LoopbackAddressesOnLanDHT = config.True |
| 248 | // Telemetry disabled by default in tests. |
| 249 | cfg.Plugins = config.Plugins{ |
| 250 | Plugins: map[string]config.Plugin{ |
| 251 | "telemetry": { |
| 252 | Disabled: true, |
| 253 | }, |
| 254 | }, |
| 255 | } |
| 256 | }) |
| 257 | return n |
| 258 | } |
| 259 | |
| 260 | // StartDaemonWithReq runs a Kubo daemon with the given request. |
| 261 | // This overwrites the request Path with the Kubo bin path. |
| 262 | // |
| 263 | // For example, if you want to run the daemon and see stderr and stdout to debug: |
| 264 | // |
| 265 | // node.StartDaemonWithReq(harness.RunRequest{ |
| 266 | // CmdOpts: []harness.CmdOpt{ |
| 267 | // harness.RunWithStderr(os.Stdout), |
| 268 | // harness.RunWithStdout(os.Stdout), |
| 269 | // }, |
| 270 | // }) |
| 271 | func (n *Node) StartDaemonWithReq(req RunRequest, authorization string) *Node { |
| 272 | alive := n.IsAlive() |
| 273 | if alive { |
| 274 | log.Panicf("node %d is already running", n.ID) |
| 275 | } |
| 276 | newReq := req |
| 277 | newReq.Path = n.IPFSBin |
| 278 | newReq.Args = append([]string{"daemon"}, req.Args...) |
| 279 | newReq.RunFunc = (*exec.Cmd).Start |
| 280 | |
| 281 | log.Debugf("starting node %d", n.ID) |
| 282 | res := n.Runner.MustRun(newReq) |
| 283 | |
| 284 | n.Daemon = res |
| 285 | |
| 286 | log.Debugf("node %d started, checking API", n.ID) |
| 287 | n.WaitOnAPI(authorization) |
| 288 | return n |
| 289 | } |
| 290 | |
| 291 | func (n *Node) StartDaemon(ipfsArgs ...string) *Node { |
| 292 | return n.StartDaemonWithReq(RunRequest{ |
| 293 | Args: ipfsArgs, |
| 294 | }, "") |
| 295 | } |
| 296 | |
| 297 | func (n *Node) StartDaemonWithAuthorization(secret string, ipfsArgs ...string) *Node { |
| 298 | return n.StartDaemonWithReq(RunRequest{ |
| 299 | Args: ipfsArgs, |
| 300 | }, secret) |
| 301 | } |
| 302 | |
| 303 | func (n *Node) signalAndWait(watch <-chan struct{}, signal os.Signal, t time.Duration) bool { |
| 304 | err := n.Daemon.Cmd.Process.Signal(signal) |
| 305 | if err != nil { |
| 306 | // On Windows, Process.Wait() sets the handle state to "released" |
| 307 | // rather than "done", so a subsequent Signal() returns EINVAL |
| 308 | // instead of ErrProcessDone. Treat both as "already exited". |
| 309 | if errors.Is(err, os.ErrProcessDone) || errors.Is(err, syscall.EINVAL) { |
| 310 | log.Debugf("process for node %d has already finished", n.ID) |
| 311 | return true |
| 312 | } |
| 313 | log.Panicf("error killing daemon for node %d with peer ID %s: %s", n.ID, n.PeerID(), err.Error()) |
| 314 | } |
| 315 | timer := time.NewTimer(t) |
| 316 | defer timer.Stop() |
| 317 | select { |
| 318 | case <-watch: |
| 319 | return true |
| 320 | case <-timer.C: |
| 321 | return false |
| 322 | } |
| 323 | } |
| 324 | |
| 325 | func (n *Node) StopDaemon() *Node { |
| 326 | log.Debugf("stopping node %d", n.ID) |
| 327 | if n.Daemon == nil { |
| 328 | log.Debugf("didn't stop node %d since no daemon present", n.ID) |
| 329 | return n |
| 330 | } |
| 331 | watch := make(chan struct{}, 1) |
| 332 | go func() { |
| 333 | _, _ = n.Daemon.Cmd.Process.Wait() |
| 334 | watch <- struct{}{} |
| 335 | }() |
| 336 | |
| 337 | // os.Interrupt does not support interrupts on Windows https://github.com/golang/go/issues/46345 |
| 338 | if runtime.GOOS == "windows" { |
| 339 | if n.signalAndWait(watch, syscall.SIGKILL, 5*time.Second) { |
| 340 | return n |
| 341 | } |
| 342 | log.Panicf("timed out stopping node %d with peer ID %s", n.ID, n.PeerID()) |
| 343 | } |
| 344 | |
| 345 | log.Debugf("signaling node %d with SIGTERM", n.ID) |
| 346 | if n.signalAndWait(watch, syscall.SIGTERM, 1*time.Second) { |
| 347 | return n |
| 348 | } |
| 349 | log.Debugf("signaling node %d with SIGTERM", n.ID) |
| 350 | if n.signalAndWait(watch, syscall.SIGTERM, 2*time.Second) { |
| 351 | return n |
| 352 | } |
| 353 | log.Debugf("signaling node %d with SIGQUIT", n.ID) |
| 354 | if n.signalAndWait(watch, syscall.SIGQUIT, 5*time.Second) { |
| 355 | return n |
| 356 | } |
| 357 | log.Debugf("signaling node %d with SIGKILL", n.ID) |
| 358 | if n.signalAndWait(watch, syscall.SIGKILL, 5*time.Second) { |
| 359 | return n |
| 360 | } |
| 361 | log.Panicf("timed out stopping node %d with peer ID %s", n.ID, n.PeerID()) |
| 362 | return n |
| 363 | } |
| 364 | |
| 365 | func (n *Node) APIAddr() multiaddr.Multiaddr { |
| 366 | ma, err := n.TryAPIAddr() |
| 367 | if err != nil { |
| 368 | panic(err) |
| 369 | } |
| 370 | return ma |
| 371 | } |
| 372 | |
| 373 | func (n *Node) APIURL() string { |
| 374 | apiAddr := n.APIAddr() |
| 375 | netAddr, err := manet.ToNetAddr(apiAddr) |
| 376 | if err != nil { |
| 377 | panic(err) |
| 378 | } |
| 379 | return "http://" + netAddr.String() |
| 380 | } |
| 381 | |
| 382 | func (n *Node) TryAPIAddr() (multiaddr.Multiaddr, error) { |
| 383 | b, err := os.ReadFile(filepath.Join(n.Dir, "api")) |
| 384 | if err != nil { |
| 385 | return nil, err |
| 386 | } |
| 387 | ma, err := multiaddr.NewMultiaddr(string(b)) |
| 388 | if err != nil { |
| 389 | return nil, err |
| 390 | } |
| 391 | return ma, nil |
| 392 | } |
| 393 | |
| 394 | func (n *Node) checkAPI(authorization string) bool { |
| 395 | apiAddr, err := n.TryAPIAddr() |
| 396 | if err != nil { |
| 397 | log.Debugf("node %d API addr not available yet: %s", n.ID, err.Error()) |
| 398 | return false |
| 399 | } |
| 400 | |
| 401 | if unixAddr, err := apiAddr.ValueForProtocol(multiaddr.P_UNIX); err == nil { |
| 402 | parts := strings.SplitN(unixAddr, "/", 2) |
| 403 | if len(parts) < 1 { |
| 404 | panic("malformed unix socket address") |
| 405 | } |
| 406 | fileName := "/" + parts[1] |
| 407 | _, err := os.Stat(fileName) |
| 408 | return !errors.Is(err, fs.ErrNotExist) |
| 409 | } |
| 410 | |
| 411 | ip, err := apiAddr.ValueForProtocol(multiaddr.P_IP4) |
| 412 | if err != nil { |
| 413 | panic(err) |
| 414 | } |
| 415 | port, err := apiAddr.ValueForProtocol(multiaddr.P_TCP) |
| 416 | if err != nil { |
| 417 | panic(err) |
| 418 | } |
| 419 | url := fmt.Sprintf("http://%s:%s/api/v0/id", ip, port) |
| 420 | log.Debugf("checking API for node %d at %s", n.ID, url) |
| 421 | |
| 422 | req, err := http.NewRequest(http.MethodPost, url, nil) |
| 423 | if err != nil { |
| 424 | panic(err) |
| 425 | } |
| 426 | if authorization != "" { |
| 427 | req.Header.Set("Authorization", authorization) |
| 428 | } |
| 429 | |
| 430 | httpResp, err := http.DefaultClient.Do(req) |
| 431 | if err != nil { |
| 432 | log.Debugf("node %d API check error: %s", err.Error()) |
| 433 | return false |
| 434 | } |
| 435 | defer httpResp.Body.Close() |
| 436 | resp := struct { |
| 437 | ID string |
| 438 | }{} |
| 439 | |
| 440 | respBytes, err := io.ReadAll(httpResp.Body) |
| 441 | if err != nil { |
| 442 | log.Debugf("error reading API check response for node %d: %s", n.ID, err.Error()) |
| 443 | return false |
| 444 | } |
| 445 | log.Debugf("got API check response for node %d: %s", n.ID, string(respBytes)) |
| 446 | |
| 447 | err = json.Unmarshal(respBytes, &resp) |
| 448 | if err != nil { |
| 449 | log.Debugf("error decoding API check response for node %d: %s", n.ID, err.Error()) |
| 450 | return false |
| 451 | } |
| 452 | if resp.ID == "" { |
| 453 | log.Debugf("API check response for node %d did not contain a Peer ID", n.ID) |
| 454 | return false |
| 455 | } |
| 456 | respPeerID, err := peer.Decode(resp.ID) |
| 457 | if err != nil { |
| 458 | panic(err) |
| 459 | } |
| 460 | |
| 461 | peerID := n.PeerID() |
| 462 | if respPeerID != peerID { |
| 463 | log.Panicf("expected peer ID %s but got %s", peerID, resp.ID) |
| 464 | } |
| 465 | |
| 466 | log.Debugf("API check for node %d successful", n.ID) |
| 467 | return true |
| 468 | } |
| 469 | |
| 470 | func (n *Node) PeerID() peer.ID { |
| 471 | cfg := n.ReadConfig() |
| 472 | id, err := peer.Decode(cfg.Identity.PeerID) |
| 473 | if err != nil { |
| 474 | panic(err) |
| 475 | } |
| 476 | return id |
| 477 | } |
| 478 | |
| 479 | func (n *Node) WaitOnAPI(authorization string) *Node { |
| 480 | log.Debugf("waiting on API for node %d", n.ID) |
| 481 | for range 50 { |
| 482 | if n.checkAPI(authorization) { |
| 483 | log.Debugf("daemon API found, daemon stdout: %s", n.Daemon.Stdout.String()) |
| 484 | return n |
| 485 | } |
| 486 | time.Sleep(400 * time.Millisecond) |
| 487 | } |
| 488 | log.Panicf("node %d with peer ID %s failed to come online: \n%s\n\n%s", n.ID, n.PeerID(), n.Daemon.Stderr.String(), n.Daemon.Stdout.String()) |
| 489 | return n |
| 490 | } |
| 491 | |
| 492 | func (n *Node) IsAlive() bool { |
| 493 | if n.Daemon == nil || n.Daemon.Cmd == nil || n.Daemon.Cmd.Process == nil { |
| 494 | return false |
| 495 | } |
| 496 | log.Debugf("signaling node %d daemon process for liveness check", n.ID) |
| 497 | err := n.Daemon.Cmd.Process.Signal(syscall.Signal(0)) |
| 498 | if err == nil { |
| 499 | log.Debugf("node %d daemon is alive", n.ID) |
| 500 | return true |
| 501 | } |
| 502 | log.Debugf("node %d daemon not alive: %s", err.Error()) |
| 503 | return false |
| 504 | } |
| 505 | |
| 506 | func (n *Node) SwarmAddrs() []multiaddr.Multiaddr { |
| 507 | res := n.Runner.Run(RunRequest{ |
| 508 | Path: n.IPFSBin, |
| 509 | Args: []string{"swarm", "addrs", "local"}, |
| 510 | }) |
| 511 | if res.ExitCode() != 0 { |
| 512 | // If swarm command fails (e.g., daemon not online), return empty slice |
| 513 | log.Debugf("Node %d: swarm addrs local failed (exit %d): %s", n.ID, res.ExitCode(), res.Stderr.String()) |
| 514 | return []multiaddr.Multiaddr{} |
| 515 | } |
| 516 | out := strings.TrimSpace(res.Stdout.String()) |
| 517 | if out == "" { |
| 518 | log.Debugf("Node %d: swarm addrs local returned empty output", n.ID) |
| 519 | return []multiaddr.Multiaddr{} |
| 520 | } |
| 521 | log.Debugf("Node %d: swarm addrs local output: %s", n.ID, out) |
| 522 | outLines := strings.Split(out, "\n") |
| 523 | var addrs []multiaddr.Multiaddr |
| 524 | for _, addrStr := range outLines { |
| 525 | addrStr = strings.TrimSpace(addrStr) |
| 526 | if addrStr == "" { |
| 527 | continue |
| 528 | } |
| 529 | ma, err := multiaddr.NewMultiaddr(addrStr) |
| 530 | if err != nil { |
| 531 | panic(err) |
| 532 | } |
| 533 | addrs = append(addrs, ma) |
| 534 | } |
| 535 | log.Debugf("Node %d: parsed %d swarm addresses", n.ID, len(addrs)) |
| 536 | return addrs |
| 537 | } |
| 538 | |
| 539 | // SwarmAddrsWithTimeout waits for swarm addresses to be available |
| 540 | func (n *Node) SwarmAddrsWithTimeout(timeout time.Duration) []multiaddr.Multiaddr { |
| 541 | start := time.Now() |
| 542 | for time.Since(start) < timeout { |
| 543 | addrs := n.SwarmAddrs() |
| 544 | if len(addrs) > 0 { |
| 545 | return addrs |
| 546 | } |
| 547 | time.Sleep(100 * time.Millisecond) |
| 548 | } |
| 549 | return []multiaddr.Multiaddr{} |
| 550 | } |
| 551 | |
| 552 | func (n *Node) SwarmAddrsWithPeerIDs() []multiaddr.Multiaddr { |
| 553 | return n.SwarmAddrsWithPeerIDsTimeout(5 * time.Second) |
| 554 | } |
| 555 | |
| 556 | func (n *Node) SwarmAddrsWithPeerIDsTimeout(timeout time.Duration) []multiaddr.Multiaddr { |
| 557 | ipfsProtocol := multiaddr.ProtocolWithCode(multiaddr.P_IPFS).Name |
| 558 | peerID := n.PeerID() |
| 559 | var addrs []multiaddr.Multiaddr |
| 560 | for _, ma := range n.SwarmAddrsWithTimeout(timeout) { |
| 561 | // add the peer ID to the multiaddr if it doesn't have it |
| 562 | _, err := ma.ValueForProtocol(multiaddr.P_IPFS) |
| 563 | if errors.Is(err, multiaddr.ErrProtocolNotFound) { |
| 564 | comp, err := multiaddr.NewComponent(ipfsProtocol, peerID.String()) |
| 565 | if err != nil { |
| 566 | panic(err) |
| 567 | } |
| 568 | ma = ma.Encapsulate(comp) |
| 569 | } |
| 570 | addrs = append(addrs, ma) |
| 571 | } |
| 572 | return addrs |
| 573 | } |
| 574 | |
| 575 | func (n *Node) SwarmAddrsWithoutPeerIDs() []multiaddr.Multiaddr { |
| 576 | var addrs []multiaddr.Multiaddr |
| 577 | for _, ma := range n.SwarmAddrs() { |
| 578 | i := 0 |
| 579 | for _, c := range ma { |
| 580 | if c.Protocol().Code == multiaddr.P_IPFS { |
| 581 | continue |
| 582 | } |
| 583 | ma[i] = c |
| 584 | i++ |
| 585 | } |
| 586 | ma = ma[:i] |
| 587 | if len(ma) > 0 { |
| 588 | addrs = append(addrs, ma) |
| 589 | } |
| 590 | } |
| 591 | return addrs |
| 592 | } |
| 593 | |
| 594 | func (n *Node) Connect(other *Node) *Node { |
| 595 | // Get the peer addresses to connect to |
| 596 | addrs := other.SwarmAddrsWithPeerIDs() |
| 597 | if len(addrs) == 0 { |
| 598 | // If no addresses available, skip connection |
| 599 | log.Debugf("No swarm addresses available for connection") |
| 600 | return n |
| 601 | } |
| 602 | // Use Run instead of MustRun to avoid panics on connection failures |
| 603 | res := n.Runner.Run(RunRequest{ |
| 604 | Path: n.IPFSBin, |
| 605 | Args: []string{"swarm", "connect", addrs[0].String()}, |
| 606 | }) |
| 607 | if res.ExitCode() != 0 { |
| 608 | log.Debugf("swarm connect failed: %s", res.Stderr.String()) |
| 609 | } |
| 610 | return n |
| 611 | } |
| 612 | |
| 613 | // ConnectAndWait connects to another node and waits for the connection to be established |
| 614 | func (n *Node) ConnectAndWait(other *Node, timeout time.Duration) error { |
| 615 | // Get the peer addresses to connect to - wait up to half the timeout for addresses |
| 616 | addrs := other.SwarmAddrsWithPeerIDsTimeout(timeout / 2) |
| 617 | if len(addrs) == 0 { |
| 618 | return fmt.Errorf("no swarm addresses available for node %d after waiting %v", other.ID, timeout/2) |
| 619 | } |
| 620 | |
| 621 | otherPeerID := other.PeerID() |
| 622 | |
| 623 | // Try to connect |
| 624 | res := n.Runner.Run(RunRequest{ |
| 625 | Path: n.IPFSBin, |
| 626 | Args: []string{"swarm", "connect", addrs[0].String()}, |
| 627 | }) |
| 628 | if res.ExitCode() != 0 { |
| 629 | return fmt.Errorf("swarm connect failed: %s", res.Stderr.String()) |
| 630 | } |
| 631 | |
| 632 | // Wait for connection to be established |
| 633 | start := time.Now() |
| 634 | for time.Since(start) < timeout { |
| 635 | peers := n.Peers() |
| 636 | for _, peerAddr := range peers { |
| 637 | if peerID, err := peerAddr.ValueForProtocol(multiaddr.P_P2P); err == nil { |
| 638 | if peerID == otherPeerID.String() { |
| 639 | return nil // Connection established |
| 640 | } |
| 641 | } |
| 642 | } |
| 643 | time.Sleep(100 * time.Millisecond) |
| 644 | } |
| 645 | |
| 646 | return fmt.Errorf("timeout waiting for connection to node %d (peer %s)", other.ID, otherPeerID) |
| 647 | } |
| 648 | |
| 649 | func (n *Node) Peers() []multiaddr.Multiaddr { |
| 650 | // Wait for daemon to be ready if it's supposed to be running |
| 651 | if n.Daemon != nil && n.Daemon.Cmd != nil && n.Daemon.Cmd.Process != nil { |
| 652 | // Give daemon a short time to become ready |
| 653 | for range 10 { |
| 654 | if n.IsAlive() { |
| 655 | break |
| 656 | } |
| 657 | time.Sleep(100 * time.Millisecond) |
| 658 | } |
| 659 | } |
| 660 | res := n.Runner.Run(RunRequest{ |
| 661 | Path: n.IPFSBin, |
| 662 | Args: []string{"swarm", "peers"}, |
| 663 | }) |
| 664 | if res.ExitCode() != 0 { |
| 665 | // If swarm peers fails (e.g., daemon not online), return empty slice |
| 666 | log.Debugf("swarm peers failed: %s", res.Stderr.String()) |
| 667 | return []multiaddr.Multiaddr{} |
| 668 | } |
| 669 | var addrs []multiaddr.Multiaddr |
| 670 | for _, line := range res.Stdout.Lines() { |
| 671 | ma, err := multiaddr.NewMultiaddr(line) |
| 672 | if err != nil { |
| 673 | panic(err) |
| 674 | } |
| 675 | addrs = append(addrs, ma) |
| 676 | } |
| 677 | return addrs |
| 678 | } |
| 679 | |
| 680 | func (n *Node) PeerWith(other *Node) { |
| 681 | n.UpdateConfig(func(cfg *config.Config) { |
| 682 | var addrs []multiaddr.Multiaddr |
| 683 | for _, addrStr := range other.ReadConfig().Addresses.Swarm { |
| 684 | ma, err := multiaddr.NewMultiaddr(addrStr) |
| 685 | if err != nil { |
| 686 | panic(err) |
| 687 | } |
| 688 | addrs = append(addrs, ma) |
| 689 | } |
| 690 | |
| 691 | cfg.Peering.Peers = append(cfg.Peering.Peers, peer.AddrInfo{ |
| 692 | ID: other.PeerID(), |
| 693 | Addrs: addrs, |
| 694 | }) |
| 695 | }) |
| 696 | } |
| 697 | |
| 698 | func (n *Node) Disconnect(other *Node) { |
| 699 | n.IPFS("swarm", "disconnect", "/p2p/"+other.PeerID().String()) |
| 700 | } |
| 701 | |
| 702 | // GatewayURL waits for the gateway file and then returns its contents or times out. |
| 703 | func (n *Node) GatewayURL() string { |
| 704 | timer := time.NewTimer(1 * time.Second) |
| 705 | defer timer.Stop() |
| 706 | for { |
| 707 | select { |
| 708 | case <-timer.C: |
| 709 | panic("timeout waiting for gateway file") |
| 710 | default: |
| 711 | b, err := os.ReadFile(filepath.Join(n.Dir, "gateway")) |
| 712 | if err == nil { |
| 713 | return strings.TrimSpace(string(b)) |
| 714 | } |
| 715 | if !errors.Is(err, fs.ErrNotExist) { |
| 716 | panic(err) |
| 717 | } |
| 718 | time.Sleep(1 * time.Millisecond) |
| 719 | } |
| 720 | } |
| 721 | } |
| 722 | |
| 723 | func (n *Node) GatewayClient() *HTTPClient { |
| 724 | return &HTTPClient{ |
| 725 | Client: http.DefaultClient, |
| 726 | BaseURL: n.GatewayURL(), |
| 727 | } |
| 728 | } |
| 729 | |
| 730 | func (n *Node) APIClient() *HTTPClient { |
| 731 | return &HTTPClient{ |
| 732 | Client: http.DefaultClient, |
| 733 | BaseURL: n.APIURL(), |
| 734 | } |
| 735 | } |
| 736 | |
| 737 | // DatastoreCount returns the count of entries matching the given prefix. |
| 738 | // Requires the daemon to be stopped. |
| 739 | func (n *Node) DatastoreCount(prefix string) int64 { |
| 740 | res := n.IPFS("diag", "datastore", "count", prefix) |
| 741 | count, _ := strconv.ParseInt(strings.TrimSpace(res.Stdout.String()), 10, 64) |
| 742 | return count |
| 743 | } |
| 744 | |
| 745 | // DatastorePut writes a key-value pair to the datastore. |
| 746 | // Requires the daemon to be stopped. |
| 747 | func (n *Node) DatastorePut(key, value string) { |
| 748 | n.IPFS("diag", "datastore", "put", key, value) |
| 749 | } |
| 750 | |
| 751 | // DatastoreGet retrieves the value at the given key. |
| 752 | // Requires the daemon to be stopped. Returns nil if key not found. |
| 753 | func (n *Node) DatastoreGet(key string) []byte { |
| 754 | res := n.RunIPFS("diag", "datastore", "get", key) |
| 755 | if res.Err != nil { |
| 756 | return nil |
| 757 | } |
| 758 | return res.Stdout.Bytes() |
| 759 | } |
| 760 | |
| 761 | // DatastoreHasKey checks if a key exists in the datastore. |
| 762 | // Requires the daemon to be stopped. |
| 763 | func (n *Node) DatastoreHasKey(key string) bool { |
| 764 | res := n.RunIPFS("diag", "datastore", "get", key) |
| 765 | return res.Err == nil |
| 766 | } |