@cryptotaxi247 / kubo / commits / 0507e8aba

fix: migration fetcher robustness (#11305)

## Problem On a repo from `go-ipfs` or Kubo older than v0.27, the one-time migration uses `http.DefaultClient` (no timeouts) against a single hardcoded `trustless-gateway.link`. If that gateway is slow or blocked, the daemon hangs indefinitely before the data store opens, with no fallback. Reported in ipfs/ipfs-desktop#3147, where a user with a v11 repo thought they had lost 4,444 added images. ## Fix - HTTP client gets dial, TLS, and response-header timeouts (15s, 15s, and boxo's `DefaultRetrievalTimeout` of 30s). - The `"HTTPS"` alias in `Migration.DownloadSources` expands to five trustless community gateways instead of one. Trust is in local per-block multihash verification, not the operator. - Outbound requests send `?format=car` (or `?format=ipns-record`) alongside `Accept`, since some gateways honor only one. - `MultiFetcher` gets a session-scoped quarantine: a failing fetcher moves to the back of the rotation; after three full failed loops it latches `ErrMultiFetcherExhausted` pointing the user at `Migration.DownloadSources`. A cancelled context exits the loop early so it never poisons the quarantine. - `RetryFetcher` is removed; rotation across distinct gateways replaces same-gateway retries. Also fixes two pre-existing bugs in the same path: `NewHttpFetcher` ignored the `userAgent` argument so every request shipped Go's default `Go-http-client/1.1`, and `resolveIPNS` leaked the response body. The `Migration` config and `"HTTPS"` alias keep working the same way for users; the alias just expands to more gateways internally. Closes #7933 Closes #3137 Closes #8911 Closes ipfs/ipfs-desktop#3147

Marcin Rataj committed May 16, 2026 at 19:33 UTC 0507e8aba004d20497b159b6c673fc7e246905c5

8 files changed +489 -64

cmd/ipfs/kubo/add_migrations.go

+1 -1

@@ -36,7 +36,7 @@ func addMigrations(ctx context.Context, node *core.IpfsNode, fetcher migrations.

36 if err != nil {

37 return err

38 }

39 - case *migrations.HttpFetcher, *migrations.RetryFetcher: // https://github.com/ipfs/kubo/issues/8780

39 + case *migrations.HttpFetcher: // https://github.com/ipfs/kubo/issues/8780

40 // Add the downloaded migration files directly

41 if migrations.DownloadDirectory != "" {

42 var paths []string

docs/changelogs/v0.42.md

@@ -13,6 +13,7 @@ This release was brought to you by the [Shipyard](https://ipshipyard.com/) team.

13 - [🎯 Announce CIDs on demand with `ipfs provide once`](#-announce-cids-on-demand-with-ipfs-provide-once)

14 - [⚙️ `Provide.DHT.Interval=0` no longer disables providing](#%EF%B8%8F-providedhtinterval0-no-longer-disables-providing)

15 - [🐛 Fixed pin operations hanging under pinned reprovide strategies](#-fixed-pin-operations-hanging-under-pinned-reprovide-strategies)

16 + - [🐛 Smoother first-run upgrades from very old repos](#-smoother-first-run-upgrades-from-very-old-repos)

17 - [🐛 Reliable shutdown and container health checks](#-reliable-shutdown-and-container-health-checks)

18 - [🚨 ERROR log for listeners blocked by `Swarm.AddrFilters` or `Addresses.NoAnnounce`](#-error-log-for-listeners-blocked-by-swarmaddrfilters-or-addressesnoannounce)

19 - [📊 OpenTelemetry: scope info now exposed as labels](#-opentelemetry-scope-info-now-exposed-as-labels)

@@ -62,6 +63,10 @@ In a terminal, the command shows a running count of queued CIDs. With `--enc=jso

64 The pinner now snapshots the index under the read lock and releases it before the reprovider starts, so pin operations are no longer blocked by the reprovide cycle. The default `Provide.Strategy=all` was not affected.

66 +#### 🐛 Smoother first-run upgrades from very old repos

67 +

68 +The one-time migration for repos from `go-ipfs` or Kubo older than v0.27 now retries across several gateways with HTTP timeouts, so a single slow or blocked gateway no longer hangs the daemon. Set [`Migration.DownloadSources`](https://github.com/ipfs/kubo/blob/master/docs/config.md#migrationdownloadsources) to use your own gateway list.

69 +

70 #### 🐛 Reliable shutdown and container health checks

72 Sending `SIGTERM` or `SIGINT` to kubo could leave the daemon stuck "half-shutdown": internal subsystems had stopped, but the process kept running and answering the RPC API. Docker and Kubernetes health checks reported the node as healthy while it had quietly stopped serving content. Recovery required a manual `docker restart`. Separately, the pinner could log a `pebble: closed` panic trace when the datastore closed before ongoing pin operations finished.

repo/fsrepo/migrations/fetch_test.go

+275

@@ -3,11 +3,17 @@ package migrations

3 import (

4 "bufio"

5 "bytes"

6 + "context"

7 + "errors"

8 "fmt"

9 + "net/http"

10 + "net/http/httptest"

11 "os"

12 "path/filepath"

13 "runtime"

14 "strings"

15 + "sync"

16 + "sync/atomic"

17 "testing"

18 )

@@ -158,6 +164,93 @@ func TestFetchBinary(t *testing.T) {

164 }

165 }

166

167 +// TestHttpFetcherUserAgent guards against a regression where NewHttpFetcher

168 +// accepts a userAgent parameter but forgets to store it on the struct,

169 +// silently sending Go's default "Go-http-client/1.1" instead of the

170 +// migration agent string.

171 +func TestHttpFetcherUserAgent(t *testing.T) {

172 + const wantUA = "kubo/migration"

173 +

174 + var gotUA string

175 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

176 + gotUA = r.Header.Get("User-Agent")

177 + w.WriteHeader(http.StatusNotFound)

178 + }))

179 + defer srv.Close()

180 +

181 + fetcher := NewHttpFetcher("/ipfs/bafyreigh2akiscaildcqabsyg3dfr6chu3fgpregiymsck7e7aqa4s52zy", srv.URL, wantUA, 0)

182 + _, _ = fetcher.Fetch(t.Context(), "/anything")

183 +

184 + if gotUA != wantUA {

185 + t.Fatalf("User-Agent: got %q, want %q", gotUA, wantUA)

186 + }

187 +}

188 +

189 +// TestMigrationDownloadSourcesFailover is an end-to-end check that two

190 +// gateways listed in Migration.DownloadSources (here passed straight into

191 +// GetMigrationFetcher, the same path ReadMigrationConfig feeds) cooperate via

192 +// MultiFetcher: when the first gateway either errors with 404 or returns

193 +// bytes that don't parse as a CAR, the second gateway is attempted and the

194 +// migration data flows through.

195 +func TestMigrationDownloadSourcesFailover(t *testing.T) {

196 + ctx := t.Context()

197 +

198 + t.Run("first gateway returns 404", func(t *testing.T) {

199 + var badHits atomic.Int64

200 + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {

201 + badHits.Add(1)

202 + http.Error(w, "not found", http.StatusNotFound)

203 + }))

204 + defer bad.Close()

205 +

206 + // Migration.DownloadSources order: bad first, real test gateway second.

207 + fetcher, err := GetMigrationFetcher([]string{bad.URL, testServer.URL}, testIpfsDist, nil)

208 + if err != nil {

209 + t.Fatalf("GetMigrationFetcher: %v", err)

210 + }

211 + defer fetcher.Close()

212 +

213 + out, err := fetcher.Fetch(ctx, "/kubo/versions")

214 + if err != nil {

215 + t.Fatalf("expected failover to second gateway, got: %v", err)

216 + }

217 + if len(out) < 6 {

218 + t.Fatalf("second gateway should have served the versions file, got %d bytes", len(out))

219 + }

220 + if badHits.Load() == 0 {

221 + t.Fatal("first gateway was never tried; the failover path did not actually run")

222 + }

223 + })

224 +

225 + t.Run("first gateway returns invalid CAR bytes", func(t *testing.T) {

226 + var badHits atomic.Int64

227 + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {

228 + badHits.Add(1)

229 + w.Header().Set("Content-Type", "application/vnd.ipld.car")

230 + w.WriteHeader(http.StatusOK)

231 + _, _ = w.Write([]byte("this is definitely not a valid CAR file"))

232 + }))

233 + defer bad.Close()

234 +

235 + fetcher, err := GetMigrationFetcher([]string{bad.URL, testServer.URL}, testIpfsDist, nil)

236 + if err != nil {

237 + t.Fatalf("GetMigrationFetcher: %v", err)

238 + }

239 + defer fetcher.Close()

240 +

241 + out, err := fetcher.Fetch(ctx, "/kubo/versions")

242 + if err != nil {

243 + t.Fatalf("expected failover to second gateway, got: %v", err)

244 + }

245 + if len(out) < 6 {

246 + t.Fatalf("second gateway should have served the versions file, got %d bytes", len(out))

247 + }

248 + if badHits.Load() == 0 {

249 + t.Fatal("first gateway was never tried; the failover path did not actually run")

250 + }

251 + })

252 +}

253 +

254 func TestMultiFetcher(t *testing.T) {

255 ctx := t.Context()

256

@@ -175,3 +268,185 @@ func TestMultiFetcher(t *testing.T) {

268 fmt.Println("unexpected more data")

269 }

270 }

271 +

272 +// TestMultiFetcherQuarantine verifies that a fetcher which fails once is

273 +// moved to the back of the rotation on subsequent calls, so a dead gateway

274 +// does not cost the full HTTP timeout on every parallel migration download.

275 +func TestMultiFetcherQuarantine(t *testing.T) {

276 + ctx := t.Context()

277 +

278 + tracker := &countingFetcher{}

279 + good := NewHttpFetcher(testIpfsDist, testServer.URL, "", 0)

280 + mf := NewMultiFetcher(tracker, good)

281 +

282 + // First call: tracker is healthy, gets tried first, fails. good takes over.

283 + if _, err := mf.Fetch(ctx, "/kubo/versions"); err != nil {

284 + t.Fatalf("first fetch: %v", err)

285 + }

286 + if got := tracker.calls.Load(); got != 1 {

287 + t.Fatalf("expected tracker to be tried once, got %d", got)

288 + }

289 +

290 + // Second call: tracker is quarantined and must not be tried while good

291 + // is still healthy.

292 + if _, err := mf.Fetch(ctx, "/kubo/versions"); err != nil {

293 + t.Fatalf("second fetch: %v", err)

294 + }

295 + if got := tracker.calls.Load(); got != 1 {

296 + t.Fatalf("expected tracker to stay quarantined, got %d calls", got)

297 + }

298 +}

299 +

300 +// TestMultiFetcherQuarantineReset verifies that when every fetcher fails in a

301 +// single Fetch call, the quarantine resets so the next call retries all

302 +// fetchers from scratch rather than inheriting a fully poisoned set.

303 +func TestMultiFetcherQuarantineReset(t *testing.T) {

304 + ctx := t.Context()

305 +

306 + a := &countingFetcher{}

307 + b := &countingFetcher{}

308 + mf := NewMultiFetcher(a, b)

309 +

310 + if _, err := mf.Fetch(ctx, "/anything"); err == nil {

311 + t.Fatal("expected error when all fetchers fail")

312 + }

313 + if ca, cb := a.calls.Load(), b.calls.Load(); ca != 1 || cb != 1 {

314 + t.Fatalf("first call: expected each fetcher tried once, got a=%d b=%d", ca, cb)

315 + }

316 +

317 + if _, err := mf.Fetch(ctx, "/anything"); err == nil {

318 + t.Fatal("expected error when all fetchers fail")

319 + }

320 + // After the total wipeout, both should have been retried fresh, not

321 + // skipped as quarantined.

322 + if ca, cb := a.calls.Load(), b.calls.Load(); ca != 2 || cb != 2 {

323 + t.Fatalf("second call after reset: expected each fetcher tried again, got a=%d b=%d", ca, cb)

324 + }

325 +}

326 +

327 +// TestMultiFetcherExhaustionCap verifies the MultiFetcher gives up after

328 +// maxMultiFetcherFullLoopFailures full rotations, returning

329 +// ErrMultiFetcherExhausted without trying inner fetchers again.

330 +func TestMultiFetcherExhaustionCap(t *testing.T) {

331 + ctx := t.Context()

332 +

333 + a := &countingFetcher{}

334 + b := &countingFetcher{}

335 + mf := NewMultiFetcher(a, b)

336 +

337 + // Three failed full rotations should latch the breaker.

338 + for i := range maxMultiFetcherFullLoopFailures {

339 + if _, err := mf.Fetch(ctx, "/x"); err == nil {

340 + t.Fatalf("rotation %d: expected error", i+1)

341 + }

342 + }

343 +

344 + expectedCalls := int64(maxMultiFetcherFullLoopFailures)

345 + if ca, cb := a.calls.Load(), b.calls.Load(); ca != expectedCalls || cb != expectedCalls {

346 + t.Fatalf("after %d rotations: expected each fetcher called %d times, got a=%d b=%d",

347 + maxMultiFetcherFullLoopFailures, expectedCalls, ca, cb)

348 + }

349 +

350 + // Subsequent calls must hard-error with ErrMultiFetcherExhausted and

351 + // must not invoke the inner fetchers again.

352 + _, err := mf.Fetch(ctx, "/x")

353 + if !errors.Is(err, ErrMultiFetcherExhausted) {

354 + t.Fatalf("expected ErrMultiFetcherExhausted, got %v", err)

355 + }

356 + if ca, cb := a.calls.Load(), b.calls.Load(); ca != expectedCalls || cb != expectedCalls {

357 + t.Fatalf("inner fetchers called after exhaustion: a=%d b=%d", ca, cb)

358 + }

359 +}

360 +

361 +// TestMultiFetcherConcurrent exercises the locking paths under -race by

362 +// hammering one MultiFetcher from many goroutines, mirroring how

363 +// fetchMigrations spawns parallel downloads against a shared fetcher.

364 +func TestMultiFetcherConcurrent(t *testing.T) {

365 + ctx := t.Context()

366 +

367 + bad := &countingFetcher{}

368 + good := NewHttpFetcher(testIpfsDist, testServer.URL, "", 0)

369 + mf := NewMultiFetcher(bad, good)

370 +

371 + const goroutines = 16

372 + const callsPerGoroutine = 8

373 +

374 + var wg sync.WaitGroup

375 + wg.Add(goroutines)

376 + for range goroutines {

377 + go func() {

378 + defer wg.Done()

379 + for range callsPerGoroutine {

380 + if _, err := mf.Fetch(ctx, "/kubo/versions"); err != nil {

381 + t.Errorf("unexpected fetch error: %v", err)

382 + return

383 + }

384 + }

385 + }()

386 + }

387 + wg.Wait()

388 +}

389 +

390 +// TestMultiFetcherSuccessResetsCounter verifies that any successful Fetch

391 +// resets the loop-failure counter, so transient blips during a long session

392 +// don't accumulate toward the exhaustion cap.

393 +func TestMultiFetcherSuccessResetsCounter(t *testing.T) {

394 + ctx := t.Context()

395 +

396 + bad := &countingFetcher{}

397 + good := NewHttpFetcher(testIpfsDist, testServer.URL, "", 0)

398 + mf := NewMultiFetcher(bad, good)

399 +

400 + // Many alternating success calls must not trip the breaker.

401 + for i := range maxMultiFetcherFullLoopFailures * 3 {

402 + if _, err := mf.Fetch(ctx, "/kubo/versions"); err != nil {

403 + t.Fatalf("call %d: %v", i+1, err)

404 + }

405 + }

406 + if err := mf.exhaustedErr(); err != nil {

407 + t.Fatalf("breaker tripped despite repeated successes: %v", err)

408 + }

409 +}

410 +

411 +// TestMultiFetcherContextCancelled verifies that a cancelled context exits

412 +// the rotation early without quarantining every fetcher or counting toward

413 +// the exhaustion cap. Otherwise three user Ctrl-Cs in a row would latch

414 +// ErrMultiFetcherExhausted on a perfectly healthy gateway list.

415 +func TestMultiFetcherContextCancelled(t *testing.T) {

416 + a := &countingFetcher{}

417 + b := &countingFetcher{}

418 + mf := NewMultiFetcher(a, b)

419 +

420 + ctx, cancel := context.WithCancel(t.Context())

421 + cancel()

422 +

423 + for i := range maxMultiFetcherFullLoopFailures + 2 {

424 + _, err := mf.Fetch(ctx, "/x")

425 + if !errors.Is(err, context.Canceled) {

426 + t.Fatalf("call %d: expected context.Canceled, got %v", i+1, err)

427 + }

428 + }

429 +

430 + // Each call should have exited after the first fetcher returned the

431 + // cancellation error, so b is never tried and the breaker never latches.

432 + if got := b.calls.Load(); got != 0 {

433 + t.Fatalf("second fetcher should not be tried after cancellation, got %d calls", got)

434 + }

435 + if err := mf.exhaustedErr(); err != nil {

436 + t.Fatalf("breaker latched on cancelled-context loops: %v", err)

437 + }

438 +}

439 +

440 +// countingFetcher always errors and records how many times it was called.

441 +// The counter is atomic so the fetcher is safe to share across goroutines

442 +// during -race tests.

443 +type countingFetcher struct {

444 + calls atomic.Int64

445 +}

446 +

447 +func (c *countingFetcher) Fetch(ctx context.Context, _ string) ([]byte, error) {

448 + c.calls.Add(1)

449 + return nil, fmt.Errorf("countingFetcher always fails")

450 +}

451 +

452 +func (c *countingFetcher) Close() error { return nil }

repo/fsrepo/migrations/fetcher.go

+112 -5

@@ -6,6 +6,7 @@ import (

6 "fmt"

7 "io"

8 "os"

9 + "sync"

10 )

12 const (

@@ -16,8 +17,20 @@ const (

18 // Distribution environ variable.

19 envIpfsDistPath = "IPFS_DIST_PATH"

20 +

21 + // maxMultiFetcherFullLoopFailures caps how many times a MultiFetcher

22 + // may exhaust every fetcher before it gives up for the rest of its

23 + // lifetime. Without a cap, a fully unreachable network would keep

24 + // every Fetch call paying the full per-gateway timeout once per call.

25 + maxMultiFetcherFullLoopFailures = 3

26 )

28 +// ErrMultiFetcherExhausted is returned by MultiFetcher.Fetch after every

29 +// fetcher has failed maxMultiFetcherFullLoopFailures full rotations in a row.

30 +// The message points the user at the recovery path: replacing the gateway

31 +// list via Migration.DownloadSources in the Kubo config.

32 +var ErrMultiFetcherExhausted = errors.New("migration download exhausted: every configured gateway failed; add a reachable HTTPS gateway to Migration.DownloadSources in your Kubo config and retry")

33 +

34 type Fetcher interface {

35 // Fetch attempts to fetch the file at the given ipfs path.

36 Fetch(ctx context.Context, filePath string) ([]byte, error)

@@ -25,10 +38,25 @@ type Fetcher interface {

38 Close() error

39 }

28 -// MultiFetcher holds multiple Fetchers and provides a Fetch that tries each

29 -// until one succeeds.

41 +// MultiFetcher tries each Fetcher in turn until one succeeds. A fetcher that

42 +// has already failed in this MultiFetcher's lifetime moves to the back of the

43 +// rotation; if every healthy fetcher fails, the quarantined ones run as a

44 +// fallback so a transient outage can self-heal. If every fetcher fails in a

45 +// single call, the quarantine resets and the next call starts fresh, but

46 +// only up to maxMultiFetcherFullLoopFailures times: after that the

47 +// MultiFetcher returns ErrMultiFetcherExhausted without trying again.

48 +//

49 +// This acts as a session-scoped circuit breaker: when migrations issue many

50 +// parallel downloads through one MultiFetcher, the first failure drops a

51 +// dead gateway from rotation for the rest of the session instead of charging

52 +// every goroutine the full HTTP timeout against it.

53 type MultiFetcher struct {

54 fetchers []Fetcher

55 +

56 + mu sync.Mutex

57 + failed map[int]struct{}

58 + loopFailures int

59 + exhausted error

60 }

62 type limitReadCloser struct {

@@ -41,25 +69,104 @@ type limitReadCloser struct {

69 func NewMultiFetcher(f ...Fetcher) *MultiFetcher {

70 mf := &MultiFetcher{

71 fetchers: make([]Fetcher, len(f)),

72 + failed: make(map[int]struct{}),

73 }

74 copy(mf.fetchers, f)

75 return mf

76 }

49 -// Fetch attempts to fetch the file at each of its fetchers until one succeeds.

78 +// Fetch tries each fetcher until one succeeds. Fetchers that have already

79 +// failed in this session are tried last. Once every fetcher has failed

80 +// maxMultiFetcherFullLoopFailures full loops in a row, Fetch returns

81 +// ErrMultiFetcherExhausted without further attempts.

82 func (f *MultiFetcher) Fetch(ctx context.Context, ipfsPath string) ([]byte, error) {

83 + if err := f.exhaustedErr(); err != nil {

84 + return nil, err

85 + }

86 +

87 var errs []error

52 - for _, fetcher := range f.fetchers {

53 - out, err := fetcher.Fetch(ctx, ipfsPath)

88 + for _, i := range f.tryOrder() {

89 + out, err := f.fetchers[i].Fetch(ctx, ipfsPath)

90 if err == nil {

91 + f.markOutcome(i, true)

92 return out, nil

93 }

94 + // A cancelled or timed-out context is not the gateway's fault.

95 + // Returning early avoids quarantining every fetcher and bumping

96 + // the loop-failure counter, which could latch the exhaustion

97 + // breaker after a few user-initiated cancellations.

98 + if ctxErr := ctx.Err(); ctxErr != nil {

99 + return nil, ctxErr

100 + }

101 fmt.Printf("Error fetching: %s\n", err.Error())

102 errs = append(errs, err)

103 + f.markOutcome(i, false)

104 + }

105 +

106 + // Every fetcher failed in this call. Bump the loop-failure counter

107 + // and decide whether to give up entirely or let the next call retry.

108 + if err := f.recordFullLoopFailure(errs); err != nil {

109 + return nil, err

110 }

111 return nil, errors.Join(errs...)

112 }

113

114 +// tryOrder returns the indices of all fetchers in the order they should be

115 +// tried this call: never-failed fetchers first, previously-failed ones last,

116 +// each group keeping its original order.

117 +func (f *MultiFetcher) tryOrder() []int {

118 + f.mu.Lock()

119 + defer f.mu.Unlock()

120 + order := make([]int, 0, len(f.fetchers))

121 + var quarantined []int

122 + for i := range f.fetchers {

123 + if _, bad := f.failed[i]; bad {

124 + quarantined = append(quarantined, i)

125 + } else {

126 + order = append(order, i)

127 + }

128 + }

129 + return append(order, quarantined...)

130 +}

131 +

132 +// markOutcome records the result of a single fetcher attempt: a success

133 +// clears any quarantine bit on i and resets the loop-failure streak, while a

134 +// failure puts i in quarantine. Both operations are idempotent.

135 +func (f *MultiFetcher) markOutcome(i int, success bool) {

136 + f.mu.Lock()

137 + defer f.mu.Unlock()

138 + if success {

139 + delete(f.failed, i)

140 + f.loopFailures = 0

141 + return

142 + }

143 + f.failed[i] = struct{}{}

144 +}

145 +

146 +// recordFullLoopFailure increments the full-loop counter. If the cap is

147 +// reached, the MultiFetcher latches into the exhausted state and returns

148 +// ErrMultiFetcherExhausted (wrapping the last batch of errors). Otherwise

149 +// the quarantine is cleared so the next call retries every fetcher fresh.

150 +func (f *MultiFetcher) recordFullLoopFailure(errs []error) error {

151 + f.mu.Lock()

152 + defer f.mu.Unlock()

153 + f.loopFailures++

154 + if f.loopFailures >= maxMultiFetcherFullLoopFailures {

155 + f.exhausted = fmt.Errorf("%w: %w", ErrMultiFetcherExhausted, errors.Join(errs...))

156 + return f.exhausted

157 + }

158 + clear(f.failed)

159 + return nil

160 +}

161 +

162 +// exhaustedErr returns the latched exhaustion error, or nil if the

163 +// MultiFetcher is still in service.

164 +func (f *MultiFetcher) exhaustedErr() error {

165 + f.mu.Lock()

166 + defer f.mu.Unlock()

167 + return f.exhausted

168 +}

169 +

170 func (f *MultiFetcher) Close() error {

171 var errs error

172 for _, fetcher := range f.fetchers {

repo/fsrepo/migrations/httpfetcher.go

+75 -9

@@ -5,9 +5,11 @@ import (

5 "errors"

6 "fmt"

7 "io"

8 + "net"

9 "net/http"

10 gopath "path"

11 "strings"

12 + "time"

14 "github.com/ipfs/boxo/blockservice"

15 "github.com/ipfs/boxo/blockstore"

@@ -23,18 +25,73 @@ import (

25 "github.com/ipfs/go-datastore"

26 dssync "github.com/ipfs/go-datastore/sync"

27 "github.com/ipfs/go-unixfsnode"

28 + config "github.com/ipfs/kubo/config"

29 gocarv2 "github.com/ipld/go-car/v2"

30 dagpb "github.com/ipld/go-codec-dagpb"

31 madns "github.com/multiformats/go-multiaddr-dns"

32 )

34 const (

32 - // default is different name than ipfs.io which is being blocked by some ISPs

35 + // defaultGatewayURL is used when no gateway is configured. Some ISPs

36 + // block ipfs.io, so we use a different hostname.

37 defaultGatewayURL = "https://trustless-gateway.link"

38 // Default maximum download size.

39 defaultFetchLimit = 1024 * 1024 * 512

40 +

41 + // Sized for users on slow / high-latency networks (e.g. VPNs through

42 + // congested links): a 3-RTT TLS handshake at 1-2s RTT plus packet

43 + // loss can legitimately approach 10s. 15s leaves headroom while

44 + // still failing fast against truly dead gateways.

45 + dialTimeout = 15 * time.Second

46 + tlsHandshakeTimeout = 15 * time.Second

47 )

49 +// defaultMigrationGateways is a last-resort fallback used when

50 +// Migration.DownloadSources expands the "HTTPS" alias. The first entry

51 +// (trustless-gateway.link) serves nearly all users; the rest are tried

52 +// only when it is blocked or unreachable.

53 +//

54 +// Including third-party gateways is safe: each block is fetched as CAR

55 +// and verified against the requested CID's multihash, so a malicious

56 +// operator cannot substitute different content.

57 +//

58 +// TODO: replace this static list with a dynamic source, either the public

59 +// gateway checker list at

60 +// https://github.com/ipfs/public-gateway-checker/raw/refs/heads/main/gateways.json

61 +// or AutoConf. Not done yet because this code path only runs for repos

62 +// from go-ipfs or Kubo older than v0.27 (roughly 2020 vintage). Modern

63 +// Kubo ships embedded migrations and never reaches it, so the impact and

64 +// risk of leaving the list hard-coded are both low.

65 +var defaultMigrationGateways = []string{

66 + defaultGatewayURL,

67 + "https://gateway.pinata.cloud",

68 + "https://ipfs.filebase.io",

69 + "https://4everland.io",

70 + "https://dget.top",

71 +}

72 +

73 +// migrationHTTPClient is the HTTP client for migration downloads. Its timeouts

74 +// fail fast on unreachable or stalled gateways so MultiFetcher can rotate.

75 +var migrationHTTPClient = &http.Client{

76 + Transport: &http.Transport{

77 + Proxy: http.ProxyFromEnvironment,

78 + DialContext: (&net.Dialer{

79 + Timeout: dialTimeout,

80 + }).DialContext,

81 + TLSHandshakeTimeout: tlsHandshakeTimeout,

82 + // ResponseHeaderTimeout matches boxo's server-side

83 + // DefaultRetrievalTimeout (re-exported via Kubo config): a healthy

84 + // gateway returns the first byte within this budget or 504s.

85 + // Mirroring it avoids drift if boxo retunes.

86 + ResponseHeaderTimeout: config.DefaultRetrievalTimeout,

87 + IdleConnTimeout: 90 * time.Second,

88 + ExpectContinueTimeout: 1 * time.Second,

89 + ForceAttemptHTTP2: true,

90 + },

91 + // No overall Timeout: cancellation flows from the request context, so

92 + // streaming bodies run to completion under context control.

93 +}

94 +

95 // HttpFetcher fetches files over HTTP using verifiable CAR archives.

96 type HttpFetcher struct { //nolint

97 distPath string

@@ -52,9 +109,10 @@ var _ Fetcher = (*HttpFetcher)(nil)

109 // Specifying 0 for fetchLimit sets the default, -1 means no limit.

110 func NewHttpFetcher(distPath, gateway, userAgent string, fetchLimit int64) *HttpFetcher { //nolint

111 f := &HttpFetcher{

55 - distPath: LatestIpfsDist,

56 - gateway: defaultGatewayURL,

57 - limit: defaultFetchLimit,

112 + distPath: LatestIpfsDist,

113 + gateway: defaultGatewayURL,

114 + limit: defaultFetchLimit,

115 + userAgent: userAgent,

116 }

117

118 if distPath != "" {

@@ -86,7 +144,7 @@ func (f *HttpFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error

144 return nil, fmt.Errorf("path could not be resolved: %w", err)

145 }

146

89 - rc, err := f.httpRequest(ctx, imPath, "application/vnd.ipld.car")

147 + rc, err := f.httpRequest(ctx, imPath, "application/vnd.ipld.car", "car")

148 if err != nil {

149 return nil, fmt.Errorf("failed to fetch CAR: %w", err)

150 }

@@ -123,10 +181,11 @@ func (f *HttpFetcher) resolvePath(ctx context.Context, pathStr string) (path.Imm

181 }

182

183 func (f *HttpFetcher) resolveIPNS(ctx context.Context, name ipns.Name) (path.Path, error) {

126 - rc, err := f.httpRequest(ctx, name.AsPath(), "application/vnd.ipfs.ipns-record")

184 + rc, err := f.httpRequest(ctx, name.AsPath(), "application/vnd.ipfs.ipns-record", "ipns-record")

185 if err != nil {

186 return path.ImmutablePath{}, err

187 }

188 + defer rc.Close()

189

190 rc = NewLimitReadCloser(rc, int64(ipns.MaxRecordSize))

191 rawRecord, err := io.ReadAll(rc)

@@ -156,8 +215,15 @@ func (f *HttpFetcher) resolveDNSLink(ctx context.Context, p path.Path) (path.Pat

215 return res.Path, nil

216 }

217

159 -func (f *HttpFetcher) httpRequest(ctx context.Context, p path.Path, accept string) (io.ReadCloser, error) {

218 +func (f *HttpFetcher) httpRequest(ctx context.Context, p path.Path, accept, format string) (io.ReadCloser, error) {

219 url := f.gateway + p.String()

220 + // Pass the format hint as both an Accept header and a ?format= query

221 + // parameter. The trustless gateway spec defines both as valid

222 + // signaling mechanisms, and some gateway implementations honor one

223 + // but not the other; sending both maximizes compatibility.

224 + if format != "" {

225 + url += "?format=" + format

226 + }

227 fmt.Printf("Fetching with HTTP: %q\n", url)

228 req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)

229 if err != nil {

@@ -169,9 +235,9 @@ func (f *HttpFetcher) httpRequest(ctx context.Context, p path.Path, accept strin

235 req.Header.Set("User-Agent", f.userAgent)

236 }

237

172 - resp, err := http.DefaultClient.Do(req)

238 + resp, err := migrationHTTPClient.Do(req)

239 if err != nil {

174 - return nil, fmt.Errorf("http.DefaultClient.Do error: %w", err)

240 + return nil, fmt.Errorf("migration http request error: %w", err)

241 }

242

243 if resp.StatusCode >= 400 {

repo/fsrepo/migrations/migrations.go

+10 -5

@@ -157,21 +157,26 @@ func ReadMigrationConfig(repoRoot string, userConfigFile string) (*config.Migrat

157 return &cfg.Migration, nil

158 }

159

160 -// GetMigrationFetcher creates one or more fetchers according to

161 -// downloadSources.

160 +// GetMigrationFetcher creates one or more fetchers from downloadSources.

161 +// Multiple fetchers are wrapped in a MultiFetcher that rotates to the next

162 +// gateway when one errors and quarantines failed gateways for the session.

163 //

164 // Deprecated: This function is used by legacy migration downloads and will be removed

165 // in a future version. Use RunHybridMigrations or RunEmbeddedMigrations instead.

166 func GetMigrationFetcher(downloadSources []string, distPath string, newIpfsFetcher func(string) Fetcher) (Fetcher, error) {

167 const httpUserAgent = "kubo/migration"

167 - const numTriesPerHTTP = 3

168

169 var fetchers []Fetcher

170 for _, src := range downloadSources {

171 src := strings.TrimSpace(src)

172 switch src {

173 case "HTTPS", "https", "HTTP", "http":

174 - fetchers = append(fetchers, &RetryFetcher{NewHttpFetcher(distPath, "", httpUserAgent, 0), numTriesPerHTTP})

174 + // Expand the alias into the full ordered list of trustless

175 + // community-provided gateways so migration survives a

176 + // single-gateway outage.

177 + for _, gw := range defaultMigrationGateways {

178 + fetchers = append(fetchers, NewHttpFetcher(distPath, gw, httpUserAgent, 0))

179 + }

180 case "IPFS", "ipfs":

181 return nil, errors.New("IPFS downloads are not supported for legacy migrations (repo versions <16). Please use only HTTPS in Migration.DownloadSources")

182 case "":

@@ -188,7 +193,7 @@ func GetMigrationFetcher(downloadSources []string, distPath string, newIpfsFetch

193 default:

194 return nil, errors.New("bad gateway address: url scheme must be http or https")

195 }

191 - fetchers = append(fetchers, &RetryFetcher{NewHttpFetcher(distPath, u.String(), httpUserAgent, 0), numTriesPerHTTP})

196 + fetchers = append(fetchers, NewHttpFetcher(distPath, u.String(), httpUserAgent, 0))

197 }

198 }

199

repo/fsrepo/migrations/migrations_test.go

+11 -11

@@ -310,10 +310,8 @@ func TestGetMigrationFetcher(t *testing.T) {

310 if err != nil {

311 t.Fatal(err)

312 }

313 - if rf, ok := f.(*RetryFetcher); !ok {

314 - t.Fatal("expected RetryFetcher")

315 - } else if _, ok := rf.Fetcher.(*HttpFetcher); !ok {

316 - t.Fatal("expected HttpFetcher")

313 + if _, ok := f.(*HttpFetcher); !ok {

314 + t.Fatalf("expected HttpFetcher, got %T", f)

315 }

316

317 downloadSources = []string{"ipfs"}

@@ -327,10 +325,12 @@ func TestGetMigrationFetcher(t *testing.T) {

325 if err != nil {

326 t.Fatal(err)

327 }

330 - if rf, ok := f.(*RetryFetcher); !ok {

331 - t.Fatal("expected RetryFetcher")

332 - } else if _, ok := rf.Fetcher.(*HttpFetcher); !ok {

333 - t.Fatal("expected HttpFetcher")

328 + mf, ok := f.(*MultiFetcher)

329 + if !ok {

330 + t.Fatal("expected MultiFetcher for HTTPS alias expansion")

331 + }

332 + if mf.Len() != len(defaultMigrationGateways) {

333 + t.Fatalf("expected %d fetchers from HTTPS alias, got %d", len(defaultMigrationGateways), mf.Len())

334 }

335

336 downloadSources = []string{"IPFS", "HTTPS"}

@@ -344,12 +344,12 @@ func TestGetMigrationFetcher(t *testing.T) {

344 if err != nil {

345 t.Fatal(err)

346 }

347 - mf, ok := f.(*MultiFetcher)

347 + mf, ok = f.(*MultiFetcher)

348 if !ok {

349 t.Fatal("expected MultiFetcher")

350 }

351 - if mf.Len() != 2 {

352 - t.Fatal("expected 2 fetchers in MultiFetcher")

351 + if mf.Len() != len(defaultMigrationGateways)+1 {

352 + t.Fatalf("expected %d fetchers in MultiFetcher, got %d", len(defaultMigrationGateways)+1, mf.Len())

353 }

354

355 downloadSources = nil

repo/fsrepo/migrations/retryfetcher.go deleted

-33

@@ -1,33 +0,0 @@

1 -package migrations

2 -

3 -import (

4 - "context"

5 - "fmt"

6 -)

7 -

8 -type RetryFetcher struct {

9 - Fetcher

10 - MaxTries int

11 -}

12 -

13 -var _ Fetcher = (*RetryFetcher)(nil)

14 -

15 -func (r *RetryFetcher) Fetch(ctx context.Context, filePath string) ([]byte, error) {

16 - var lastErr error

17 - for i := 0; i < r.MaxTries; i++ {

18 - out, err := r.Fetcher.Fetch(ctx, filePath)

19 - if err == nil {

20 - return out, nil

21 - }

22 -

23 - if ctx.Err() != nil {

24 - return nil, ctx.Err()

25 - }

26 - lastErr = err

27 - }

28 - return nil, fmt.Errorf("exceeded number of retries. last error was %w", lastErr)

29 -}

30 -

31 -func (r *RetryFetcher) Close() error {

32 - return r.Fetcher.Close()

33 -}