| 1 | package migrations |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "os" |
| 9 | "sync" |
| 10 | ) |
| 11 | |
| 12 | const ( |
| 13 | // Current distribution to fetch migrations from. |
| 14 | CurrentIpfsDist = "/ipfs/QmRzRGJEjYDfbHHaALnHBuhzzrkXGdwcPMrgd5fgM7hqbe" // fs-repo-15-to-16 v1.0.1 |
| 15 | // Latest distribution path. Default for fetchers. |
| 16 | LatestIpfsDist = "/ipns/dist.ipfs.tech" |
| 17 | |
| 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 | ) |
| 27 | |
| 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) |
| 37 | // Close performs any cleanup after the fetcher is not longer needed. |
| 38 | Close() error |
| 39 | } |
| 40 | |
| 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 | } |
| 61 | |
| 62 | type limitReadCloser struct { |
| 63 | io.Reader |
| 64 | io.Closer |
| 65 | } |
| 66 | |
| 67 | // NewMultiFetcher creates a MultiFetcher with the given Fetchers. The |
| 68 | // Fetchers are tried in order, then passed to this function. |
| 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 | } |
| 77 | |
| 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 |
| 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 { |
| 173 | if err := fetcher.Close(); err != nil { |
| 174 | errs = errors.Join(errs, err) |
| 175 | } |
| 176 | } |
| 177 | return errs |
| 178 | } |
| 179 | |
| 180 | func (f *MultiFetcher) Len() int { |
| 181 | return len(f.fetchers) |
| 182 | } |
| 183 | |
| 184 | func (f *MultiFetcher) Fetchers() []Fetcher { |
| 185 | return f.fetchers |
| 186 | } |
| 187 | |
| 188 | // NewLimitReadCloser returns a new io.ReadCloser with the reader wrapped in a |
| 189 | // io.LimitedReader limited to reading the amount specified. |
| 190 | func NewLimitReadCloser(rc io.ReadCloser, limit int64) io.ReadCloser { |
| 191 | return limitReadCloser{ |
| 192 | Reader: io.LimitReader(rc, limit), |
| 193 | Closer: rc, |
| 194 | } |
| 195 | } |
| 196 | |
| 197 | // GetDistPathEnv returns the IPFS path to the distribution site, using |
| 198 | // the value of environ variable specified by envIpfsDistPath. If the environ |
| 199 | // variable is not set, then returns the provided distPath, and if that is not set |
| 200 | // then returns the IPNS path. |
| 201 | // |
| 202 | // To get the IPFS path of the latest distribution, if not overridden by the |
| 203 | // environ variable: GetDistPathEnv(CurrentIpfsDist). |
| 204 | func GetDistPathEnv(distPath string) string { |
| 205 | if dist := os.Getenv(envIpfsDistPath); dist != "" { |
| 206 | return dist |
| 207 | } |
| 208 | if distPath == "" { |
| 209 | return LatestIpfsDist |
| 210 | } |
| 211 | return distPath |
| 212 | } |