diff --git a/.github/workflows/interop.yml b/.github/workflows/interop.yml index 966d3dc3429..0d9cbd2778a 100644 --- a/.github/workflows/interop.yml +++ b/.github/workflows/interop.yml @@ -63,71 +63,93 @@ jobs: run: shell: bash steps: + # Pin a concrete major instead of an lts/ alias. setup-node resolves an + # alias by fetching the actions/node-versions manifest from GitHub with no + # retry and no fallback, so a GitHub blip fails the job outright. A + # concrete version falls back to downloading from nodejs.org. - uses: actions/setup-node@v7 with: - node-version: lts/* + node-version: '24' - uses: actions/download-artifact@v8 with: name: kubo path: cmd/ipfs - run: chmod +x cmd/ipfs/ipfs - - run: sudo apt update - - run: sudo apt install -y libxkbcommon0 libxdamage1 libgbm1 libpango-1.0-0 libcairo2 # dependencies for playwright # Cache node_modules based on latest @helia/interop version from npm registry. # This ensures we always test against the latest release while still benefiting # from caching when the version hasn't changed. - name: Get latest @helia/interop version id: helia-version - run: echo "version=$(npm view @helia/interop version)" >> $GITHUB_OUTPUT - - name: Cache helia-interop node_modules - uses: actions/cache@v6 + run: | + set -euo pipefail + for attempt in 1 2 3; do + if version=$(npm view @helia/interop version); then break; fi + echo "npm view failed (attempt $attempt of 3), retrying" >&2 + sleep $((attempt * 5)) + done + if [ -z "${version:-}" ]; then + echo "could not resolve the latest @helia/interop version from the npm registry" >&2 + exit 1 + fi + echo "version=$version" >> "$GITHUB_OUTPUT" + # Restore and save are separate so a failed install cannot be cached under + # a version key and then reused by every later run on that version. + - name: Restore helia-interop node_modules + uses: actions/cache/restore@v6 id: helia-cache with: path: node_modules key: ${{ runner.os }}-helia-interop-${{ steps.helia-version.outputs.version }} - - name: Install @helia/interop + # Install the exact version the cache key names. Plain `npm install + # @helia/interop` is a second `latest` lookup, and helia has published + # several interop versions within an hour, so the key could name a version + # that is not the one on disk. + - name: Install @helia/interop@${{ steps.helia-version.outputs.version }} if: steps.helia-cache.outputs.cache-hit != 'true' - run: npm install @helia/interop - # Recurring @helia/interop regression: the published .aegir.js is unusable - # from inside node_modules, in one of two ways. - # - # 1. Missing: the package ships without .aegir.js in its "files", so - # `aegir test` has no config and discovers no spec files. Happened around - # ipfs/helia#1001 (fixed by ipfs/helia#1003), and again in ipfs/helia#1049 - # ("feat!: make libp2p optional"), which dropped it from 11.0.0-11.0.2. - # 2. Present but broken: the shipped config globs the TypeScript sources - # (./src/*.spec.ts), written for running tests inside the helia monorepo. - # Node refuses to type-strip .ts files under node_modules - # (ERR_UNSUPPORTED_NODE_MODULES_TYPE_STRIPPING, no tsconfig is shipped), - # so mocha dies on the first spec. This is what ipfs/helia#1066 restored - # in 11.0.3. - # - # In both cases we write a minimal config pointing at the prebuilt - # ./dist/src, which is all a `-t node` run needs; the rest of helia's own - # config is browser-only setup skipped under node. A shipped config that - # does not glob .ts files is left untouched, so this step self-heals once - # helia publishes a node_modules-safe config. It writes into node_modules - # and is fragile if helia's dist layout shifts. - - name: Work around unusable @helia/interop/.aegir.js (ipfs/helia#1049, ipfs/helia#1066) + run: npm install --fetch-retries=5 --fetch-retry-maxtimeout=120000 "@helia/interop@${{ steps.helia-version.outputs.version }}" + # Search for the compiled specs rather than hardcoding where they live. + # Upstream has already moved this stuff around more than once, and every + # time it did, this job failed with mocha's unhelpful "no test files + # found". Searching survives a move; if the specs stop being called + # *.spec.js the job still fails, but it says so in one line. + - name: Find the compiled interop specs + working-directory: node_modules/@helia/interop run: | - config=node_modules/@helia/interop/.aegir.js - if [ ! -f "$config" ]; then - echo "export default { test: { files: './dist/src/*.spec.js' } }" > "$config" - echo "injected minimal $config: file missing from published package" - elif grep -qE "files:.*\.spec\.ts" "$config"; then - echo "export default { test: { files: './dist/src/*.spec.js' } }" > "$config" - echo "replaced $config: shipped config globs .ts sources, which node cannot run from node_modules" - else - echo "$config is usable as shipped, no workaround needed" + set -euo pipefail + find . -name '*.spec.js' -not -path './node_modules/*' | sort > "$RUNNER_TEMP/helia-specs.txt" + count=$(wc -l < "$RUNNER_TEMP/helia-specs.txt") + echo "@helia/interop@${{ steps.helia-version.outputs.version }} ships $count compiled spec files:" + cat "$RUNNER_TEMP/helia-specs.txt" + if [ "$count" -eq 0 ]; then + echo "found no compiled *.spec.js in the published package, so its layout changed" >&2 + exit 1 fi + - name: Save helia-interop node_modules + if: steps.helia-cache.outputs.cache-hit != 'true' + uses: actions/cache/save@v6 + with: + path: node_modules + key: ${{ runner.os }}-helia-interop-${{ steps.helia-version.outputs.version }} # We run aegir directly (instead of the helia-interop binary) because the # bin spawns a bare `aegir test` with no way to pass the flags below. + # + # --files: pass the specs we found above rather than trusting the .aegir.js + # that @helia/interop publishes, which is written for running inside the + # helia monorepo and has broken this job twice. It globs the TypeScript + # sources (./src/*.spec.ts), which node refuses to type-strip under + # node_modules (ERR_UNSUPPORTED_NODE_MODULES_TYPE_STRIPPING, no tsconfig is + # shipped), and versions 11.0.0-11.0.2 shipped no config at all + # (ipfs/helia#1049), leaving aegir with nothing to discover. + # # --exit: the node interop specs spawn kubo daemons (via ipfsd-ctl + # KUBO_BINARY) whose libp2p/RPC handles keep Node's event loop alive after # the run, so mocha prints its summary and then hangs until the job timeout. # Force exit once the run completes. - name: Run helia-interop tests - run: npx aegir test -t node --bail -- --exit + run: | + set -euo pipefail + mapfile -t specs < "$RUNNER_TEMP/helia-specs.txt" + npx aegir test -t node --bail --files "${specs[@]}" -- --exit env: KUBO_BINARY: ${{ github.workspace }}/cmd/ipfs/ipfs working-directory: node_modules/@helia/interop diff --git a/core/corehttp/routing.go b/core/corehttp/routing.go index 239f8737bec..a61fc145a7f 100644 --- a/core/corehttp/routing.go +++ b/core/corehttp/routing.go @@ -131,7 +131,11 @@ func (r *contentRouter) GetClosestPeers(ctx context.Context, key cid.Cid) (iter. return nil, fmt.Errorf("GetClosestPeers not supported for DHT type %T", r.n.DHTClient) } - if err != nil { + // A lookup cut short by the deadline returns the closest peers found so far + // along with the context error. The HTTP routing server caps every request + // (server.DefaultRoutingTimeout), so on a slow query this is the difference + // between a useful answer and a 500. + if err != nil && len(peers) == 0 { return nil, err } diff --git a/core/coreunix/add_test.go b/core/coreunix/add_test.go index 9f5b1daec19..9592128ebb4 100644 --- a/core/coreunix/add_test.go +++ b/core/coreunix/add_test.go @@ -7,6 +7,7 @@ import ( "math/rand" "os" "path/filepath" + "sync" "testing" "time" @@ -29,6 +30,31 @@ import ( const testPeerID = "QmTFauExutTsy4XP6JbMFcw2Wa9645HJt2bTqL6qYDCKfe" +// signalFirstRead closes signal the first time the wrapped reader is read from. +// The gc tests use it to tell when the adder has moved on to a given file. +type signalFirstRead struct { + r io.Reader + once sync.Once + signal chan struct{} +} + +func (s *signalFirstRead) Read(p []byte) (int, error) { + s.once.Do(func() { close(s.signal) }) + return s.r.Read(p) +} + +// waitForGCRequest blocks until a gc is waiting for the pin lock. +func waitForGCRequest(ctx context.Context, t *testing.T, locker blockstore.GCLocker) { + t.Helper() + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); { + if locker.GCRequested(ctx) { + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("timed out waiting for gc to request the pin lock") +} + func TestAddMultipleGCLive(t *testing.T) { ctx := t.Context() r := &repo.Mock{ @@ -84,8 +110,10 @@ func TestAddMultipleGCLive(t *testing.T) { gc1out = gc.GC(ctx, node.Blockstore, node.Repo.Datastore(), node.Pinning, nil) }() - // Give GC goroutine time to reach GCLock (will block there waiting for adder) - time.Sleep(time.Millisecond * 100) + // Wait for the GC goroutine to reach GCLock, where it blocks behind the pin + // lock the adder holds. Until it gets there the adder has no reason to pause, + // and it would run to the end of the next file before yielding. + waitForGCRequest(ctx, t, node.Blockstore) // GC shouldn't get the lock until after the file is completely added select { @@ -126,8 +154,8 @@ func TestAddMultipleGCLive(t *testing.T) { gc2out = gc.GC(ctx, node.Blockstore, node.Repo.Datastore(), node.Pinning, nil) }() - // Give GC goroutine time to reach GCLock - time.Sleep(time.Millisecond * 100) + // Wait for the GC goroutine to reach GCLock, as above. + waitForGCRequest(ctx, t, node.Blockstore) select { case <-gc2started: @@ -187,7 +215,8 @@ func TestAddGCLive(t *testing.T) { // make two files with pipes so we can 'pause' the add for timing of the test piper, pipew := io.Pipe() - hangfile := files.NewReaderFile(piper) + addingHangfile := make(chan struct{}) + hangfile := files.NewReaderFile(&signalFirstRead{r: piper, signal: addingHangfile}) rfd := files.NewBytesFile([]byte("testfileD")) @@ -215,6 +244,11 @@ func TestAddGCLive(t *testing.T) { t.Fatal("add shouldn't complete yet") } + // Wait until the add is inside the hanging file. Between two files the adder + // hands the pin lock over to a waiting gc, so asking for gc before this point + // lets gc start immediately and the assertions below become meaningless. + <-addingHangfile + var gcout <-chan gc.Result gcstarted := make(chan struct{}) go func() { @@ -222,6 +256,10 @@ func TestAddGCLive(t *testing.T) { gcout = gc.GC(ctx, node.Blockstore, node.Repo.Datastore(), node.Pinning, nil) }() + // Wait for gc to actually queue up behind the pin lock the add holds, so the + // add has something to yield to once it finishes the current file. + waitForGCRequest(ctx, t, node.Blockstore) + // gc shouldn't start until we let the add finish its current file. if _, err := pipew.Write([]byte("some data for file b")); err != nil { t.Fatal(err) @@ -233,8 +271,6 @@ func TestAddGCLive(t *testing.T) { default: } - time.Sleep(time.Millisecond * 100) // make sure gc gets to requesting lock - // finish write and unblock gc pipew.Close() diff --git a/docs/examples/kubo-as-a-library/main.go b/docs/examples/kubo-as-a-library/main.go index aa8e983b1ea..76e14c45907 100644 --- a/docs/examples/kubo-as-a-library/main.go +++ b/docs/examples/kubo-as-a-library/main.go @@ -2,10 +2,10 @@ package main import ( "context" + "errors" "flag" "fmt" "io" - "log" "os" "path/filepath" "strings" @@ -81,6 +81,12 @@ func createTempRepo() (string, error) { // No automatic bootstrap: we connect only the peers we need. cfg.Bootstrap = []string{} + // No local peer discovery either. This example dials its second node by + // address, and mDNS would race that: a connection opened while the node is + // still being built is invisible to its own bitswap, and the fetch at the + // end then waits forever. It would also reach unrelated nodes on your LAN. + cfg.Discovery.MDNS.Enabled = false + // Optional: enable experimental features by modifying cfg before Init, e.g.: if *flagExp { // https://github.com/ipfs/kubo/blob/master/docs/experimental-features.md#ipfs-filestore @@ -170,18 +176,24 @@ func connectToPeers(ctx context.Context, ipfs icore.CoreAPI, peers []string) err pi.Addrs = append(pi.Addrs, pii.Addrs...) } + var mu sync.Mutex + var errs []error wg.Add(len(peerInfos)) for _, peerInfo := range peerInfos { go func(peerInfo *peer.AddrInfo) { defer wg.Done() err := ipfs.Swarm().Connect(ctx, *peerInfo) if err != nil { - log.Printf("failed to connect to %s: %s", peerInfo.ID, err) + mu.Lock() + errs = append(errs, fmt.Errorf("connecting to %s: %w", peerInfo.ID, err)) + mu.Unlock() } }(peerInfo) } wg.Wait() - return nil + // Report failures instead of logging them. Everything after this assumes the + // peers are reachable, so carrying on would stall somewhere less obvious. + return errors.Join(errs...) } func getUnixfsNode(path string) (files.Node, error) { @@ -209,7 +221,10 @@ func main() { fmt.Println("-- Getting an IPFS node running -- ") - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + // Well inside the 2 minute budget `go test` gives this example, so a stall + // ends in a panic naming the step that hung rather than the test harness + // killing us with no clue about where we were. + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) defer cancel() // Spawn a local peer using a temporary path, for testing purposes. diff --git a/docs/examples/kubo-as-a-library/main_test.go b/docs/examples/kubo-as-a-library/main_test.go index ecc2a592a7c..e840158730c 100644 --- a/docs/examples/kubo-as-a-library/main_test.go +++ b/docs/examples/kubo-as-a-library/main_test.go @@ -14,7 +14,9 @@ func TestExample(t *testing.T) { t.Log("Starting go run main.go...") start := time.Now() - cmd := exec.Command("go", "run", "main.go") + // CommandContext so the child is killed with the test instead of being left + // behind as an orphan when the example hangs. + cmd := exec.CommandContext(t.Context(), "go", "run", "main.go") cmd.Env = append(os.Environ(), "GOLOG_LOG_LEVEL=error") // reduce libp2p noise // Stream output to both test log and capture buffer for verification diff --git a/test/cli/autoconf/ipns_test.go b/test/cli/autoconf/ipns_test.go index 71d8baeb330..f4067756381 100644 --- a/test/cli/autoconf/ipns_test.go +++ b/test/cli/autoconf/ipns_test.go @@ -1,12 +1,14 @@ package autoconf import ( + "bytes" "encoding/json" "fmt" "io" "maps" "net/http" "net/http/httptest" + "slices" "strings" "sync" "testing" @@ -72,26 +74,31 @@ func testIPNSPublishingWithWorkingEndpoint(t *testing.T) { require.Equal(t, 0, result.ExitCode(), "Publishing should succeed") assert.Contains(t, result.Stdout.String(), "Published to") - // Wait for async HTTP request to delegated publisher - time.Sleep(2 * time.Second) + // Wait for the async HTTP request to reach the delegated publisher + require.Eventually(t, func() bool { + return len(publisher.getRecordPayloads(peerIDBase36)) > 0 + }, 30*time.Second, 100*time.Millisecond, "HTTP PUT request should have been made to delegated publisher") // Verify HTTP PUT was made to delegated publisher publishedKeys := publisher.getPublishedKeys() assert.NotEmpty(t, publishedKeys, "HTTP PUT request should have been made to delegated publisher") - // Get the PUT payload that was sent to the delegated publisher - putPayload := publisher.getRecordPayload(peerIDBase36) - require.NotNil(t, putPayload, "Should have captured PUT payload") - require.Greater(t, len(putPayload), 0, "PUT payload should not be empty") - // Retrieve the IPNS record using routing get getResult := node.RunIPFS("routing", "get", "/ipns/"+peerID) require.Equal(t, 0, getResult.ExitCode(), "Should be able to retrieve IPNS record") getPayload := getResult.Stdout.Bytes() - // Compare the payloads - assert.Equal(t, putPayload, getPayload, - "PUT payload sent to delegated publisher should match what routing get returns") + // The record routing returns has to be one we PUT to the delegated + // publisher, not necessarily the first. A minute after startup the + // republisher re-signs every key and publishes it again, which yields a + // different signature and expiry for the same value, and routing prefers + // that newer record. Waiting lets the matching PUT arrive if it has not yet. + require.Eventually(t, func() bool { + return slices.ContainsFunc(publisher.getRecordPayloads(peerIDBase36), func(payload []byte) bool { + return bytes.Equal(payload, getPayload) + }) + }, 30*time.Second, 200*time.Millisecond, + "record returned by routing get should be one of the records PUT to the delegated publisher") // Also verify the record points to the expected content assert.Contains(t, getResult.Stdout.String(), testCID, @@ -252,7 +259,7 @@ type mockIPNSPublisher struct { server *httptest.Server mu sync.Mutex publishedKeys map[string]string // peerID -> published CID - recordPayloads map[string][]byte // peerID -> actual HTTP PUT record payload + recordPayloads map[string][][]byte // peerID -> every HTTP PUT record payload, in order responseFunc func(peerID string, record []byte) int // returns HTTP status code } @@ -260,7 +267,7 @@ func newMockIPNSPublisher(t *testing.T) *mockIPNSPublisher { m := &mockIPNSPublisher{ t: t, publishedKeys: make(map[string]string), - recordPayloads: make(map[string][]byte), + recordPayloads: make(map[string][][]byte), } // Default response function accepts all publishes @@ -301,9 +308,9 @@ func (m *mockIPNSPublisher) handleIPNS(w http.ResponseWriter, r *http.Request) { if status == http.StatusOK { if len(body) > 0 { - // Store the actual record payload - m.recordPayloads[peerID] = make([]byte, len(body)) - copy(m.recordPayloads[peerID], body) + // Keep every record we are sent, not just the last one. The + // republisher re-signs and re-publishes on its own schedule. + m.recordPayloads[peerID] = append(m.recordPayloads[peerID], bytes.Clone(body)) } // Mark as published @@ -335,15 +342,14 @@ func (m *mockIPNSPublisher) getPublishedKeys() map[string]string { return result } -func (m *mockIPNSPublisher) getRecordPayload(peerID string) []byte { +func (m *mockIPNSPublisher) getRecordPayloads(peerID string) [][]byte { m.mu.Lock() defer m.mu.Unlock() - if payload, exists := m.recordPayloads[peerID]; exists { - result := make([]byte, len(payload)) - copy(result, payload) - return result + payloads := make([][]byte, 0, len(m.recordPayloads[peerID])) + for _, payload := range m.recordPayloads[peerID] { + payloads = append(payloads, bytes.Clone(payload)) } - return nil + return payloads } func (m *mockIPNSPublisher) close() { diff --git a/test/cli/delegated_routing_v1_http_server_test.go b/test/cli/delegated_routing_v1_http_server_test.go index e6f5867fc49..311a9260147 100644 --- a/test/cli/delegated_routing_v1_http_server_test.go +++ b/test/cli/delegated_routing_v1_http_server_test.go @@ -7,7 +7,6 @@ import ( "time" "github.com/google/uuid" - "github.com/ipfs/boxo/autoconf" "github.com/ipfs/boxo/ipns" "github.com/ipfs/boxo/routing/http/client" "github.com/ipfs/boxo/routing/http/types" @@ -228,18 +227,26 @@ func TestRoutingV1Server(t *testing.T) { t.Parallel() routingTypes := []string{"auto", "autoclient", "dht", "dhtclient"} - for _, routingType := range routingTypes { + + // One node per routing type, all bootstrapping off the same set of + // in-process DHT peers on loopback. Pointing this at the public swarm + // instead made the test depend on a CI runner reaching + // bootstrap.libp2p.io from a cold repo, which is what used to fail. + h := harness.NewT(t) + nodes := h.NewNodes(len(routingTypes)).Init() + for i, routingType := range routingTypes { + nodes[i].UpdateConfig(func(cfg *config.Config) { + cfg.Gateway.ExposeRoutingAPI = config.True + cfg.Routing.Type = config.NewOptionalString(routingType) + }) + } + h.BootstrapWithStubDHT(nodes) + + for i, routingType := range routingTypes { + node := nodes[i] t.Run("routing_type="+routingType, func(t *testing.T) { t.Parallel() - // Single node with DHT and real bootstrap peers - node := harness.NewT(t).NewNode().Init() - node.UpdateConfig(func(cfg *config.Config) { - cfg.Gateway.ExposeRoutingAPI = config.True - cfg.Routing.Type = config.NewOptionalString(routingType) - // Set real bootstrap peers from boxo/autoconf - cfg.Bootstrap = autoconf.FallbackBootstrapPeers - }) node.StartDaemon() defer node.StopDaemon() @@ -249,23 +256,26 @@ func TestRoutingV1Server(t *testing.T) { // Query for closest peers to our own peer ID key := peer.ToCid(node.PeerID()) - // Wait for WAN DHT routing table to be populated. - // The server has a 30-second routing timeout, so we use 60 seconds - // per request to allow for network latency while preventing hangs. - // Total wait time is 5 minutes to accommodate slow CI DHT bootstrapping. - // Passing runs finish in 8-48s; failures are total bootstrap failures, - // not slow convergence, so extra headroom doesn't waste time on success. + // The stub peers are on loopback and always reachable, so the + // WAN routing table fills in as soon as the daemon finishes + // bootstrapping. The wait only covers that startup. var records []*types.PeerRecord require.EventuallyWithT(t, func(ct *assert.CollectT) { - ctx, cancel := context.WithTimeout(t.Context(), 60*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) defer cancel() resultsIter, err := c.GetClosestPeers(ctx, key) if !assert.NoError(ct, err) { return } - records, err = iter.ReadAllResults(resultsIter) - assert.NoError(ct, err) - }, 5*time.Minute, 5*time.Second) + res, err := iter.ReadAllResults(resultsIter) + if !assert.NoError(ct, err) { + return + } + if !assert.NotEmpty(ct, res) { + return + } + records = res + }, 60*time.Second, 500*time.Millisecond) // Verify we got some peers back from WAN DHT require.NotEmpty(t, records, "should return peers close to own peerid") diff --git a/test/cli/fuse/fuse_test.go b/test/cli/fuse/fuse_test.go index fb6915f85be..595ae7b419e 100644 --- a/test/cli/fuse/fuse_test.go +++ b/test/cli/fuse/fuse_test.go @@ -20,6 +20,7 @@ import ( "runtime" "sort" "strings" + "sync" "syscall" "testing" @@ -435,6 +436,9 @@ func TestFUSE(t *testing.T) { }) } +// mountLock serializes the mount step across parallel subtests. See mountAll. +var mountLock sync.Mutex + // mountAll creates mount directories and mounts IPFS, IPNS, and MFS. func mountAll(t *testing.T, node *harness.Node) (ipfsMount, ipnsMount, mfsMount string) { t.Helper() @@ -452,7 +456,18 @@ func mountAll(t *testing.T, node *harness.Node) (ipfsMount, ipnsMount, mfsMount lazyUnmount(ipnsMount) lazyUnmount(mfsMount) - result := node.IPFS("mount", "-f", ipfsMount, "-n", ipnsMount, "-m", mfsMount) + // One mount at a time. Each `ipfs mount` runs three setuid fusermount + // helpers, and the parallel subtests reach this point together, so without + // the lock a couple of dozen of them race to open /dev/fuse at once and one + // occasionally comes back with a bare exit status 1. + mountLock.Lock() + result := node.RunIPFS("mount", "-f", ipfsMount, "-n", ipnsMount, "-m", mfsMount) + mountLock.Unlock() + + // Report rather than panic, and include the daemon's stderr: fusermount + // explains itself there, and the exit code on its own says nothing. + require.NoError(t, result.Err, "ipfs mount failed\nclient stderr: %s\ndaemon stderr:\n%s", + result.Stderr.String(), node.Daemon.Stderr.String()) // Extra space after "MFS" matches the column-aligned output produced // by MountCmd in core/commands/mount_unix.go. diff --git a/test/cli/harness/peering.go b/test/cli/harness/peering.go index 2d538338ba2..3db2a7d0b44 100644 --- a/test/cli/harness/peering.go +++ b/test/cli/harness/peering.go @@ -15,32 +15,74 @@ type Peering struct { To int } +// Ports handed out by NewRandPort come from this range. It sits below the +// ephemeral range every platform we test on uses (Linux starts at 32768, macOS +// and Windows at 49152), so the kernel never assigns one of these as the source +// port of an outgoing connection. Without that, a port we probed and released +// can be taken by any of the many connections the daemons under test open, +// which used to surface as "bind: address already in use". +const ( + reservedRangeStart = 10000 + reservedRangeEnd = 32000 +) + var ( allocatedPorts = make(map[int]struct{}) portMutex sync.Mutex ) +// NewTCPListener binds a listener on a free loopback port and returns it along +// with the port number. Prefer this over NewRandPort whenever the test itself +// is the one listening: the socket stays bound the whole time, so nothing can +// take the port between picking it and using it. +func NewTCPListener(t *testing.T) (net.Listener, int) { + t.Helper() + + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("allocating a loopback listener: %v", err) + } + port := l.Addr().(*net.TCPAddr).Port + + portMutex.Lock() + allocatedPorts[port] = struct{}{} + portMutex.Unlock() + + t.Cleanup(func() { _ = l.Close() }) + return l, port +} + +// NewRandPort reserves a port for something else to bind later, typically an +// ipfs daemon we spawn. Because the port is only probed and then released, +// there is an unavoidable window where another process could take it; ports are +// picked from below the ephemeral range so that at least outgoing connections +// on a busy machine cannot land on one. +// +// When the test binds the port itself, use NewTCPListener instead. func NewRandPort() int { portMutex.Lock() defer portMutex.Unlock() for range 100 { - l, err := net.Listen("tcp", "localhost:0") - if err != nil { + port := reservedRangeStart + rand.Intn(reservedRangeEnd-reservedRangeStart) + if _, used := allocatedPorts[port]; used { continue } - port := l.Addr().(*net.TCPAddr).Port - l.Close() - if _, used := allocatedPorts[port]; !used { - allocatedPorts[port] = struct{}{} - return port + l, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + continue // in use by something outside this test binary } + l.Close() + + allocatedPorts[port] = struct{}{} + return port } - // Fallback to random port if we can't get a unique one from the OS + // Every candidate was either taken or already handed out. Give up on + // probing and return anything we have not used yet. for range 1000 { - port := 30000 + rand.Intn(10000) + port := reservedRangeStart + rand.Intn(reservedRangeEnd-reservedRangeStart) if _, used := allocatedPorts[port]; !used { allocatedPorts[port] = struct{}{} return port diff --git a/test/cli/http_retrieval_client_test.go b/test/cli/http_retrieval_client_test.go index 32628bfcea0..1a591f2d4c4 100644 --- a/test/cli/http_retrieval_client_test.go +++ b/test/cli/http_retrieval_client_test.go @@ -93,7 +93,9 @@ func TestHTTPRetrievalClient(t *testing.T) { // Ok, now attempt retrieval. // If there was no timeout and returned bytes match expected body, HTTP routing and retrieval worked end-to-end. catRes := node.IPFS("cat", testCid.String()) - assert.Equal(t, randStr, catRes.Stdout.Trimmed()) + // Compare the raw bytes. The payload is random, so roughly one run in 256 + // ends in a newline, and Trimmed() would eat it and fail the comparison. + assert.Equal(t, randStr, catRes.Stdout.String()) }) } diff --git a/test/cli/ipfswatch_test.go b/test/cli/ipfswatch_test.go index ce5798c6cd1..d8c55f0bda0 100644 --- a/test/cli/ipfswatch_test.go +++ b/test/cli/ipfswatch_test.go @@ -64,10 +64,17 @@ func TestIPFSWatch(t *testing.T) { stderrStr := result.Stderr.String() require.NotContains(t, stderrStr, "unknown datastore type", "ipfswatch should recognize datastore plugins") - // Create a test file with unique content based on timestamp + // Create a test file with unique content based on timestamp. + // Write it elsewhere and move it in, so the watcher sees a single event for + // a file that is already complete. Writing in place creates the file first + // and fills it after, and ipfswatch adds whatever is on disk when it wakes + // up, which on a busy machine is an empty file. testContent := fmt.Sprintf("ipfswatch test content generated at %s", time.Now().Format(time.RFC3339Nano)) + stagedFile := filepath.Join(h.Dir, "test.txt.staged") + err = os.WriteFile(stagedFile, []byte(testContent), 0o644) + require.NoError(t, err) testFile := filepath.Join(watchDir, "test.txt") - err = os.WriteFile(testFile, []byte(testContent), 0o644) + err = os.Rename(stagedFile, testFile) require.NoError(t, err) // Wait for ipfswatch to process the file and extract CID from log diff --git a/test/cli/p2p_test.go b/test/cli/p2p_test.go index 2400d7d8bb4..888010ef841 100644 --- a/test/cli/p2p_test.go +++ b/test/cli/p2p_test.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "io" - "net" "net/http" "os/exec" "slices" @@ -285,19 +284,16 @@ func TestP2PForeground(t *testing.T) { }) nodes.StartDaemons().Connect() - httpServerPort := harness.NewRandPort() + listener, httpServerPort := harness.NewTCPListener(t) forwardPort := harness.NewRandPort() // Start HTTP server expectedBody := "Hello from p2p tunnel!" httpServer := &http.Server{ - Addr: fmt.Sprintf("127.0.0.1:%d", httpServerPort), Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(expectedBody)) }), } - listener, err := net.Listen("tcp", httpServer.Addr) - require.NoError(t, err) go func() { _ = httpServer.Serve(listener) }() defer httpServer.Close() @@ -342,19 +338,16 @@ func TestP2PForeground(t *testing.T) { }) nodes.StartDaemons().Connect() - httpServerPort := harness.NewRandPort() + listener, httpServerPort := harness.NewTCPListener(t) forwardPort := harness.NewRandPort() // Start HTTP server expectedBody := "Hello from forward foreground tunnel!" httpServer := &http.Server{ - Addr: fmt.Sprintf("127.0.0.1:%d", httpServerPort), Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(expectedBody)) }), } - listener, err := net.Listen("tcp", httpServer.Addr) - require.NoError(t, err) go func() { _ = httpServer.Serve(listener) }() defer httpServer.Close() diff --git a/test/cli/provider_test.go b/test/cli/provider_test.go index 565db743525..9b17db32a9e 100644 --- a/test/cli/provider_test.go +++ b/test/cli/provider_test.go @@ -1446,10 +1446,15 @@ func TestProviderUniqueDedupLogging(t *testing.T) { // Single pin add with both CIDs shares one bloom. node.IPFS("pin", "add", "--fast-provide-dag", "--fast-provide-wait", cidDirA, cidDirB) + // The daemon writes this line before it answers the RPC, but the test + // reads a buffer filled by a goroutine copying the daemon's stderr, and + // that copy can lag behind the command returning. + require.True(t, waitForLogMessage(node.Daemon.Stderr, `"providedCIDs": 5`, 30*time.Second), + "fast-provide-dag should log how many CIDs it provided") + daemonLog := node.Daemon.Stderr.String() require.Contains(t, daemonLog, "bloom tracker created") require.NotContains(t, daemonLog, "bloom tracker autoscaled") - require.Contains(t, daemonLog, `"providedCIDs": 5`) require.Contains(t, daemonLog, `"skippedBranches": 1`) }) @@ -1497,10 +1502,14 @@ func TestProviderUniqueDedupLogging(t *testing.T) { waitForSweepReprovide(t, publisher, 90*time.Second, 6) + // waitForSweepReprovide polls `ipfs provide stat`, which can report the + // reprovide as done before the copy of the daemon's stderr catches up. + require.True(t, waitForLogMessage(publisher.Daemon.Stderr, `"providedCIDs": 6`, 30*time.Second), + "reprovide should log how many CIDs it provided") + daemonLog := publisher.Daemon.Stderr.String() require.Contains(t, daemonLog, "bloom tracker created") require.NotContains(t, daemonLog, "bloom tracker autoscaled") - require.Contains(t, daemonLog, `"providedCIDs": 6`) require.Contains(t, daemonLog, `"skippedBranches": 1`) }) } diff --git a/test/sharness/t0065-active-requests.sh b/test/sharness/t0065-active-requests.sh index e73e1198994..bbdcd7e60b8 100755 --- a/test/sharness/t0065-active-requests.sh +++ b/test/sharness/t0065-active-requests.sh @@ -11,6 +11,14 @@ test_description="Test active request commands" test_init_ipfs test_launch_ipfs_daemon +# By default the daemon drops every finished request from the log each time the +# log reaches a multiple of ten entries. The polling below makes several +# requests, so without this the entry we are waiting for could be swept away +# before we see it. +test_expect_success "keep finished requests in the log" ' + ipfs diag cmds set-time 60s +' + test_expect_success "command works" ' ipfs diag cmds > cmd_out ' @@ -22,11 +30,12 @@ test_expect_success "invoc shows up in output" ' test_expect_success "start longer running command" ' ipfs log tail & LOGPID=$! - go-sleep 100ms ' +# The daemon only lists the request once the backgrounded client has connected, +# which on a loaded machine takes longer than any fixed sleep we could pick. test_expect_success "long running command shows up" ' - ipfs diag cmds > cmd_out2 + test_run_repeat_60_sec "ipfs diag cmds > cmd_out2 && grep log/tail cmd_out2 | grep true" ' test_expect_success "output looks good" ' @@ -41,8 +50,10 @@ test_expect_success "kill log cmd" ' wait $LOGPID || true ' +# Same on the way out: the daemon marks the request inactive when it notices the +# client is gone, not when kill returns. test_expect_success "long running command inactive" ' - ipfs diag cmds > cmd_out3 + test_run_repeat_60_sec "ipfs diag cmds > cmd_out3 && grep log/tail cmd_out3 | grep false" ' test_expect_success "command shows up as inactive" ' diff --git a/test/sharness/t0140-swarm.sh b/test/sharness/t0140-swarm.sh index 37bb44b6440..56182c2ec38 100755 --- a/test/sharness/t0140-swarm.sh +++ b/test/sharness/t0140-swarm.sh @@ -172,8 +172,12 @@ test_expect_success "disconnect work without specifying a transport address" ' [ $(ipfsi 0 swarm peers | wc -l) -eq 0 ] ' +# No check for zero peers up front: disconnecting is not permanent, and the DHT +# keeps the other node in its routing table and re-dials it on any refresh, so +# by the time this case runs the two may be connected again. The disconnect +# itself is covered by the case above, and the full disconnect-then-connect +# cycle by the one below, both inside a single command chain. test_expect_success "connect work without specifying a transport address" ' - [ $(ipfsi 0 swarm peers | wc -l) -eq 0 ] && ipfsi 0 swarm connect "/p2p/$(iptb attr get 1 id)" && [ $(ipfsi 0 swarm peers | wc -l) -eq 1 ] '