Large file storage for AT Protocol: an IPFS server & gateway with XRPC upload & admin. atfs.dev
ipfs atproto xrpc
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780package ipfs
import ( "context" "errors" "fmt" "log/slog" "net/http" "net/netip" "path/filepath" "sync" "time"
boxobitswap "github.com/ipfs/boxo/bitswap" bsnet "github.com/ipfs/boxo/bitswap/network/bsnet" "github.com/ipfs/go-cid" ds "github.com/ipfs/go-datastore" dsync "github.com/ipfs/go-datastore/sync" "github.com/libp2p/go-libp2p" dht "github.com/libp2p/go-libp2p-kad-dht" "github.com/libp2p/go-libp2p/core/crypto" "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager" quic "github.com/libp2p/go-libp2p/p2p/transport/quic" "github.com/libp2p/go-libp2p/p2p/transport/tcp" libp2pwebrtc "github.com/libp2p/go-libp2p/p2p/transport/webrtc" "github.com/libp2p/go-libp2p/p2p/transport/websocket" webtransport "github.com/libp2p/go-libp2p/p2p/transport/webtransport" "github.com/libp2p/go-libp2p/x/rate" "github.com/multiformats/go-multiaddr"
"atfs.dev/internal/store")
const ( identityFileName = "identity.key" unixfsDirName = "unixfs"
// DefaultPort is the libp2p swarm (TCP+QUIC) listen port every atfs // instance binds — the standard kubo/IPFS default. It's no longer // configurable (see CLAUDE.md's "Bare-minimum configuration" rule): // never needed changing in practice, and dropping the knob leaves the // did:web document as the only setting still bound at startup. DefaultPort = 4001
// reprovideInterval re-announces content periodically because DHT // provider records expire (~24-48h) — 12h keeps comfortable margin. reprovideInterval = 12 * time.Hour // retryInterval replaces reprovideInterval as the next pass's delay // when every provide in a pass failed — most likely a still-cold DHT // routing table — so content doesn't stay unannounced for a full // reprovideInterval. retryInterval = 5 * time.Minute // provideTimeout bounds a single CID's provide call, so one slow or // unreachable DHT lookup can't stall the whole announce pass. provideTimeout = time.Minute // bootstrapConnectTimeout bounds a single bootstrap peer dial. bootstrapConnectTimeout = 30 * time.Second // routingTableWaitTimeout bounds how long the first provide pass // waits for the DHT routing table to gain any peers before proceeding // anyway — see waitForRoutingTablePeers. routingTableWaitTimeout = 2 * time.Minute // routingTableWaitPoll is how often the wait re-checks routing table // size. routingTableWaitPoll = 2 * time.Second // provideCoalesceDelay is how long a provide-loop kick (see kickProvide) // waits, absorbing any further kicks, before triggering a pass. This // gives a burst of reachability transitions — and the identify pushes // that follow an address change — time to settle so the pass sees the // final address set rather than a half-updated one. provideCoalesceDelay = 10 * time.Second // announceConcurrency bounds how many AnnounceRoot-triggered DHT // Provide walks run at once — see the announceSem field. provideAll's // own periodic pass needs no such bound: it already announces one CID // at a time in a single sequential loop. AnnounceRoot is different // precisely because its callers don't share that loop — a burst of // uploads or completed pins can call it in quick succession, and // ATFS-d01i showed what an unauthenticated caller able to trigger it // once per listed file, per request, turns that into: a small cap // here is what stops any future caller from reintroducing an // unbounded fan-out, the same role internal/pin's fetchSem plays for // fetch attempts. announceConcurrency = 4
// wsAnnounceTemplate builds the /tls/ws multiaddr announced for a // serviceDid domain (see announceWebSocket) — port 443, since that's // the only port a browser's wss:// can reach through an ordinary // TLS-terminating ingress (Cloudflare Tunnel and the like). wsAnnounceTemplate = "/dns4/%s/tcp/443/tls/ws"
// wsSubnetConnLimit and wsSubnetConnRPS raise go-libp2p's own // defaults for inbound connections that share one address (see // newResourceManager) now that a browser can dial in over /ws: // go-libp2p allows 8 concurrent connections and 0.2 new ones per // second (burst 16) per /32 out of the box, sized for a swarm where // every peer is its own IP address. Behind a tunnel every browser // shares the tunnel's own address instead — go-libp2p has no // X-Forwarded-For handling anywhere — so the unmodified defaults // would refuse the 9th simultaneous uploader outright. These numbers // give real headroom for that without removing the limit altogether; // whether they hold up against the real Cloudflare tunnel is // unverified (see ATFS-twv3's MEASURE todos). wsSubnetConnLimit = 64 wsSubnetConnRPS = 2.0)
// Node is atfs's IPFS-network participant: a libp2p host serving Bitswap// over atfs's blob store and its UnixFS index, participating in the Amino// DHT, periodically announcing its content's fetchable roots, and — via// Pin — fetching content the other way, from providers or the network into// the store.type Node struct { host host.Host dht *dht.IpfsDHT bs *boxobitswap.Bitswap index *Indexer store *store.Store
// provideKick signals selfHealAndProvide's provide loop (via // runProvideLoop) to run a pass soon rather than waiting out the full // reprovideInterval — see kickProvide and logReachability. Buffered to // capacity 1; kickProvide always sends non-blockingly, so a burst of // reachability transitions collapses to a single pending kick. provideKick chan struct{}
// ctx is the node's own long-lived context — derived from Start's ctx, // cancelled by cancel/Close — kept separately from any caller's // request-scoped context. AnnounceRoot needs it precisely because its // callers don't have one of their own that outlives them: an HTTP // handler's context dies the moment it returns, which is before // AnnounceRoot's background announce has any chance to run. ctx context.Context
// providers is the HTTP client every provider attempt uses (see // newProviderClient), including the guard that keeps a caller-supplied // origin from reaching an address inside the operator's network. A // field rather than a package-level client so this package's own tests // can reach an httptest origin — over TLS, and by redirecting the // connection rather than by relaxing the guard, which is the only way // to exercise the real fetch path without a rule of it going untested. // Nothing outside this package can replace it: no configuration, no // record and no request reaches this field. providers *http.Client
// announce performs a single CID's DHT provide — n.provideOne in // production, wired once n exists (see start) — used by AnnounceRoot. A // field rather than a direct call so tests can observe or replace it // without waiting out a real, or even a failing-fast offline, DHT // lookup. announce func(ctx context.Context, root cid.Cid) bool
// announceSem bounds how many AnnounceRoot calls are actually running // n.announce at once — see announceConcurrency. Acquired inside // AnnounceRoot's own goroutine, never by its caller, so the wait for a // slot can't turn AnnounceRoot itself into something that blocks; the // acquire races n.ctx.Done() so a call still queued for a slot when // Close cancels the node gives up on it instead of leaking past // shutdown, the same shape internal/pin's fetchSem uses against a // fetch attempt's own deadline. announceSem chan struct{}
cancel context.CancelFunc wg sync.WaitGroup
// closeMu guards AnnounceRoot's wg.Add against a concurrent Close. // AnnounceRoot is called from arbitrary caller goroutines, unlike the // two loops start() itself registers with wg before anyone could call // Close — so without this, a wg.Add landing after Close's own wg.Wait // has already returned zero would leak that goroutine past shutdown // (sync.WaitGroup's own documented hazard: adds with a positive delta // must happen before a Wait call they're meant to be counted by). Close // holds closeMu across setting closed and its own cancel, so an // AnnounceRoot that loses the race sees closed and skips instead. closeMu sync.Mutex closed bool
closeOnce sync.Once closeErr error}
// WebSocketConfig configures the browser-dialable /ws address this node// mounts on its libp2p host, if any — see ATFS-twv3 and CLAUDE.md's// "Serving" section. libp2p's WebSocket transport cannot be mounted into// an existing mux (newListener and its ServeHTTP are unexported, and the// transport always owns its own http.Server), so it's the other way// round here too: the transport owns Port, and Handler serves everything// on it that isn't a WebSocket upgrade. The zero value disables the whole// thing — no /ws listen address, no fallback HTTP surface — which is what// ATFS_HTTP_PORT=0 now means (see daemon.go).type WebSocketConfig struct { // Port is the port the WebSocket listener binds — the same port a // bare net.Listen used to open directly (config.HTTP.Port). 0 // disables the /ws listener entirely; Handler and AnnounceDomain are // then ignored. Port int // Handler serves every request on Port that isn't a WebSocket // upgrade: atfs's whole ordinary HTTP surface (xrpc, gateway, // did:web, the identity line), already wrapped in atfs's own // middleware (see internal/serve.WrapHandler). Required whenever // Port != 0. Handler http.Handler // AnnounceDomain, when non-empty, is the serviceDid domain (see // record.ParseDIDWebDomain) this instance's /ws address is announced // under, as /dns4/<AnnounceDomain>/tcp/443/tls/ws — added alongside, // not instead of, the plain listen address: a TLS-terminating // ingress (a Cloudflare Tunnel, say) still reaches this instance over // plain HTTP even though the browser dialed wss://, since libp2p's // Noise handshake runs inside the WebSocket frame regardless of where // TLS is terminated. Bound at startup, exactly like the served // did:web document: a later serviceDid change needs a restart to move // this (see daemon.go's did:web advisory log). AnnounceDomain string}
// Start brings up the full IPFS stack: a libp2p host (TCP+QUIC always, plus// a browser-dialable WebSocket listener when ws.Port != 0 — see// WebSocketConfig), the Amino DHT (auto mode — server whenever this// host is publicly reachable, client otherwise; standard IPFS bootstrap// peers), Bitswap serving over st and its UnixFS index, and periodic DHT// provider announcements. It walks the store to index any blob still// missing a UnixFS DAG (self-healing) before its first announce pass.//// priv is loaded separately via LoadIdentity — see internal/daemon.Run's// boot order, which needs the peer ID before the rest of the stack starts.// dataDir is used only for the UnixFS index, stored alongside the identity// file LoadIdentity(dataDir) reads.//// The returned Node shuts down cleanly when ctx is cancelled, or// explicitly via Node.Close.func Start(ctx context.Context, dataDir string, priv crypto.PrivKey, port int, st *store.Store, ws WebSocketConfig) (*Node, error) { return start(ctx, dataDir, priv, port, st, true, ws)}
// start is Start's shared implementation. network=false skips dialing the// real IPFS bootstrap peers, DHT bootstrapping, and the provide loop, so// tests can exercise the libp2p host and Bitswap serving stack entirely// offline and in-process — connected directly to another test host —// without ever touching live DHT/bootstrap infrastructure.func start(ctx context.Context, dataDir string, priv crypto.PrivKey, port int, st *store.Store, network bool, ws WebSocketConfig) (*Node, error) { index, err := OpenIndexer(filepath.Join(dataDir, unixfsDirName), st) if err != nil { return nil, err }
// relay is the AutoRelay peer source's late-bound link to the DHT (see // relaySource) — the DHT can't exist yet at this point since dht.New // needs the host we're about to build, but AutoRelay needs its peer // source at host-construction time. var relay relaySource
rm, err := newResourceManager() if err != nil { return nil, fmt.Errorf("ipfs: building resource manager: %w", err) }
h, err := libp2p.New( libp2p.Identity(priv), libp2p.ListenAddrStrings(listenAddrs(port, ws.Port)...), // Explicit rather than libp2p.DefaultTransports: the /ws entry // needs WithHTTPHandler wired in (see wsTransportOpts), and // specifying any Transport option at all disables libp2p's own // "fill in DefaultTransports when none is set" fallback — so // every transport DefaultTransports would have registered has to // be listed here too, or TCP/QUIC/WebTransport/WebRTC silently // stop working. libp2p.Transport(tcp.NewTCPTransport), libp2p.Transport(quic.NewTransport), libp2p.Transport(websocket.New, wsTransportOpts(ws)...), libp2p.Transport(webtransport.New), libp2p.Transport(libp2pwebrtc.New), libp2p.AddrsFactory(announceWebSocket(ws)), libp2p.ResourceManager(rm), libp2p.NATPortMap(), libp2p.EnableHolePunching(), // AutoNAT v1's client (reachability detection) runs unconditionally // on any host; EnableNATService additionally serves dial-backs for // other peers. EnableAutoNATv2 does the same for the newer, more // precise per-address protocol — both are trivial-cost good-citizen // additions, not required for our own reachability detection. libp2p.EnableNATService(), libp2p.EnableAutoNATv2(), // Feeds AutoRelay candidate peers from the DHT routing table once it // exists; harmlessly empty until relay.set is called below, and // permanently in offline/test mode. libp2p.EnableAutoRelayWithPeerSource(autoRelayPeerSource(relay.candidates)), ) if err != nil { return nil, fmt.Errorf("ipfs: starting libp2p host: %w", err) }
// dht.ModeAuto — kubo's default — runs as a DHT client until AutoNAT // reports this host publicly reachable, then switches to server (and // back, if reachability regresses); see the mode switch in // go-libp2p-kad-dht's handleLocalReachabilityChangedEvent. That // depends on real AutoNAT reachability events, which the offline test // path (network=false) never gets — a never-connected host settles in // (and stays in) client mode. Tests connect two hosts directly and // don't exercise DHT routing at all, but pin ModeServer explicitly // there anyway: it's the honest, unsurprising choice for a "real DHT, // just offline" test double, and keeps prior behavior unchanged. mode := dht.ModeAuto if !network { mode = dht.ModeServer } memDS := dsync.MutexWrap(ds.NewMapDatastore()) dhtOpts := []dht.Option{dht.Mode(mode), dht.Datastore(memDS)} if network { dhtOpts = append(dhtOpts, dht.BootstrapPeers(dht.GetDefaultBootstrapPeerAddrInfos()...)) } kadDHT, err := dht.New(h, dhtOpts...) if err != nil { _ = h.Close() return nil, fmt.Errorf("ipfs: starting dht: %w", err) } relay.set(kadDHT)
nodeCtx, cancel := context.WithCancel(ctx)
// bitswap.New builds a client as well as a server. The client is what // Node.Pin fetches through; it never writes to bstore (it only reads it // for duplicate accounting), so handing it the read-only serving view // is safe — see newFetcher. bstore := NewBlockstore(st, index) bsNet := bsnet.NewFromIpfsHost(h) bs := boxobitswap.New(nodeCtx, bsNet, kadDHT, bstore)
n := &Node{host: h, dht: kadDHT, bs: bs, index: index, store: st, providers: newProviderClient(), provideKick: make(chan struct{}, 1), announceSem: make(chan struct{}, announceConcurrency), ctx: nodeCtx, cancel: cancel} n.announce = n.provideOne index.announce = n.AnnounceRoot
// Belt-and-braces: nodeCtx already dies when ctx does (it's derived // from ctx), which stops the background loop below, but Close also // needs to run to actually shut the host/DHT/bitswap down. This // watcher isn't tracked in n.wg — Close waits on n.wg, and Close is // what this goroutine calls, so it must stay outside that wait. go func() { <-ctx.Done() _ = n.Close() }()
if network { n.connectBootstrapPeers(nodeCtx) }
n.wg.Add(1) go func() { defer n.wg.Done() n.selfHealAndProvide(nodeCtx, network) }()
n.wg.Add(1) go func() { defer n.wg.Done() n.logReachability(nodeCtx) }()
return n, nil}
// listenAddrs returns the swarm's TCP+QUIC listen addresses on port, plus// a plain /ws listen address on wsPort when it's non-zero (see// WebSocketConfig) — a browser can't speak TCP or QUIC, so /ws is the one// address that makes this node dialable from one at all. TLS is not// listened for directly: /ws is meant to sit behind a reverse proxy that// terminates TLS upstream (a Cloudflare Tunnel, for atfs), and libp2p's own// Noise handshake runs inside the WebSocket frame regardless — see// announceWebSocket for the /tls/ws address actually announced to peers.func listenAddrs(port, wsPort int) []string { addrs := []string{ fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", port), fmt.Sprintf("/ip6/::/tcp/%d", port), fmt.Sprintf("/ip4/0.0.0.0/udp/%d/quic-v1", port), fmt.Sprintf("/ip6/::/udp/%d/quic-v1", port), } if wsPort != 0 { addrs = append(addrs, fmt.Sprintf("/ip4/0.0.0.0/tcp/%d/ws", wsPort)) } return addrs}
// wsTransportOpts returns the websocket.Option list for libp2p.Transport// to apply to the WebSocket transport — WithHTTPHandler wired to ws.Handler// when the listener is enabled, or none at all when it's disabled (see// WebSocketConfig), which leaves the transport registered but inert,// exactly as it is today with no /ws listen address ever added.//// []any, not []websocket.Option: libp2p.Transport's variadic parameter is// untyped (it validates each option against the constructor's real// parameter type by reflection), so the options must already be boxed as// any to spread into it.func wsTransportOpts(ws WebSocketConfig) []any { if ws.Port == 0 { return nil } return []any{websocket.WithHTTPHandler(ws.Handler)}}
// announceWebSocket builds the AddrsFactory that appends this instance's// public /tls/ws address (see WebSocketConfig.AnnounceDomain) on top of// whatever addresses the host would otherwise report — or leaves the// address list untouched when there's no announce domain yet, or the /ws// listener itself is disabled. AddrsFactory runs last, after libp2p's own// reachability filtering (addrsManager.Addrs applies it after computing// the dialable set), so this address is never pruned as unreachable the// way a raw private/NATed address would be.func announceWebSocket(ws WebSocketConfig) func([]multiaddr.Multiaddr) []multiaddr.Multiaddr { identity := func(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr { return addrs } if ws.Port == 0 || ws.AnnounceDomain == "" { return identity } announce, err := multiaddr.NewMultiaddr(fmt.Sprintf(wsAnnounceTemplate, ws.AnnounceDomain)) if err != nil { // AnnounceDomain comes from record.ParseDIDWebDomain, already // validated as a bare hostname, so this would be a bug rather // than an operator mistake — log it and announce nothing extra // rather than fail startup over it. slog.Error("ipfs: building /ws announce address", "domain", ws.AnnounceDomain, "err", err) return identity } return func(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr { return append(addrs, announce) }}
// newResourceManager builds the same resource manager go-libp2p would use// by default (rcmgr.DefaultLimits, autoscaled, plus the bundled services'// own limits — see libp2p.SetDefaultServiceLimits), except for a raised// per-subnet inbound-connection ceiling and a matching connection-rate// ceiling — see wsSubnetConnLimit/wsSubnetConnRPS.func newResourceManager() (network.ResourceManager, error) { limits := rcmgr.DefaultLimits libp2p.SetDefaultServiceLimits(&limits)
loopbackV4 := netip.MustParsePrefix("127.0.0.0/8") loopbackV6 := netip.MustParsePrefix("::1/128")
return rcmgr.NewResourceManager( rcmgr.NewFixedLimiter(limits.AutoScale()), rcmgr.WithLimitPerSubnet( []rcmgr.ConnLimitPerSubnet{{PrefixLength: 32, ConnCount: wsSubnetConnLimit}}, []rcmgr.ConnLimitPerSubnet{ {PrefixLength: 56, ConnCount: wsSubnetConnLimit}, {PrefixLength: 48, ConnCount: 8 * wsSubnetConnLimit}, }, ), // WithConnRateLimiters replaces the whole rate limiter, unlike // WithLimitPerSubnet (which only overwrites the fields it's // given), so the loopback exemption go-libp2p's own default // carries has to be restated here too, or loopback traffic (every // test in this package included) would start getting rate-limited. rcmgr.WithConnRateLimiters(&rate.Limiter{ NetworkPrefixLimits: []rate.PrefixLimit{ {Prefix: loopbackV4, Limit: rate.Limit{}}, {Prefix: loopbackV6, Limit: rate.Limit{}}, }, SubnetRateLimiter: rate.SubnetLimiter{ IPv4SubnetLimits: []rate.SubnetLimit{ {PrefixLength: 32, Limit: rate.Limit{RPS: wsSubnetConnRPS, Burst: wsSubnetConnLimit}}, }, IPv6SubnetLimits: []rate.SubnetLimit{ {PrefixLength: 56, Limit: rate.Limit{RPS: wsSubnetConnRPS, Burst: wsSubnetConnLimit}}, {PrefixLength: 48, Limit: rate.Limit{RPS: 4 * wsSubnetConnRPS, Burst: 8 * wsSubnetConnLimit}}, }, GracePeriod: time.Minute, }, }), )}
// connectBootstrapPeers fires off best-effort connects to the standard// IPFS bootstrap peers so the DHT's routing table has somewhere to start.// A peer that can't be reached (most commonly: this node is offline) is// logged and otherwise ignored — startup must never depend on the// network being reachable.func (n *Node) connectBootstrapPeers(ctx context.Context) { for _, ai := range dht.GetDefaultBootstrapPeerAddrInfos() { go func(ai peer.AddrInfo) { cctx, cancel := context.WithTimeout(ctx, bootstrapConnectTimeout) defer cancel() if err := n.host.Connect(cctx, ai); err != nil { slog.Debug("ipfs: bootstrap connect failed", "peer", ai.ID, "err", err) } }(ai) }}
// selfHealAndProvide indexes any blob still missing a complete UnixFS DAG// (including healing a legacy manifest — see Indexer.isIndexed), finishes// any blob deletion that crashed between unpin and delete, then (if// network) bootstraps the DHT, waits for the routing table to gain peers,// announces every blob's fetchable root, and re-announces on a ticker// until ctx is cancelled — reprovideInterval normally, or the shorter// retryInterval if a pass's every provide failed (see nextProvideInterval).// It also runs an out-of-band pass promptly whenever logReachability// signals a material address change via n.provideKick, so provider records// pick up newly-dialable addresses well before the next scheduled pass.//// The index walk runs before the deletion sweep — see SweepDeletions —// so the sweep's shared-chunk keep-set scan never reads a stale or legacy// manifest.func (n *Node) selfHealAndProvide(ctx context.Context, network bool) { if err := n.index.WalkAndIndex(ctx); err != nil { slog.Error("ipfs: startup index walk", "err", err) } if err := n.index.SweepDeletions(ctx); err != nil { slog.Error("ipfs: startup deletion sweep", "err", err) } if !network { return }
if err := n.dht.Bootstrap(ctx); err != nil { slog.Warn("ipfs: dht bootstrap", "err", err) }
// A provide pass fired the instant the DHT is bootstrapped routinely // finds an empty routing table — Provide then fails every CID with // "failed to find any peer in table" — so give the table a bounded // window to gain peers first. Proceeding after a timeout (rather than // waiting indefinitely) is deliberate: nextProvideInterval below still // catches an all-failed pass and retries soon. if !waitForRoutingTablePeers(ctx, n.dht.RoutingTable().Size, routingTableWaitPoll, routingTableWaitTimeout) && ctx.Err() == nil { slog.Info("ipfs: routing table still empty, proceeding with first provide pass anyway", "waited", routingTableWaitTimeout) }
interval := nextProvideInterval(n.provideAll(ctx)) runProvideLoop(ctx, interval, n.provideAll, n.provideKick, provideCoalesceDelay)}
// runProvideLoop is selfHealAndProvide's post-bootstrap scheduling loop,// factored out with injectable durations/channels so it's testable without// a real DHT or real-length intervals. It runs provide (normally// n.provideAll) on a ticker starting at initial and reset to// nextProvideInterval(outcome) after every pass, plus an extra pass// whenever kicks fires — coalesced via coalesceKicks so a burst of kicks in// close succession triggers at most one extra pass. It returns when ctx is// cancelled, including mid-coalesce.func runProvideLoop(ctx context.Context, initial time.Duration, provide func(context.Context) provideOutcome, kicks <-chan struct{}, coalesceDelay time.Duration) { ticker := time.NewTicker(initial) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: ticker.Reset(nextProvideInterval(provide(ctx))) case <-kicks: if !coalesceKicks(ctx, kicks, coalesceDelay) { return } ticker.Reset(nextProvideInterval(provide(ctx))) } }}
// coalesceKicks waits out delay after a provide-loop kick, absorbing any// further kicks that arrive in that window into the same pending pass,// before reporting true to trigger it. It reports false if ctx is// cancelled first, so a kick that arrives during shutdown never blocks it.func coalesceKicks(ctx context.Context, kicks <-chan struct{}, delay time.Duration) bool { timer := time.NewTimer(delay) defer timer.Stop() for { select { case <-ctx.Done(): return false case <-timer.C: return true case <-kicks: // Absorbed into the same pending pass; keep waiting out delay. } }}
// waitForRoutingTablePeers polls sizeFn (the DHT routing table's peer// count, injected so this is testable without a real DHT) until it reports// at least one peer, ctx is cancelled, or timeout elapses. It reports// whether peers were found.func waitForRoutingTablePeers(ctx context.Context, sizeFn func() int, poll, timeout time.Duration) bool { if sizeFn() > 0 { return true }
deadline := time.NewTimer(timeout) defer deadline.Stop() ticker := time.NewTicker(poll) defer ticker.Stop() for { select { case <-ctx.Done(): return false case <-deadline.C: return false case <-ticker.C: if sizeFn() > 0 { return true } } }}
// provideOutcome summarizes one provideAll pass: how many roots it tried// to announce and how many of those succeeded. selfHealAndProvide feeds it// to nextProvideInterval to pick the next pass's delay.type provideOutcome struct { attempted int succeeded int}
// provideAll announces the fetchable root of every blob in the store: the// blessed CID itself for blobs at or under MaxRawBlockSize, the UnixFS// root for larger, indexed ones. A failure announcing any single CID// (most commonly: this node is offline) is logged and skipped — it must// never crash the node or abort the rest of the pass.func (n *Node) provideAll(ctx context.Context) provideOutcome { var out provideOutcome for c, err := range n.store.List() { if ctx.Err() != nil { return out } if err != nil { slog.Error("ipfs: provide: listing store", "err", err) continue }
root, ok, err := n.rootToProvide(c) if err != nil { slog.Error("ipfs: provide: resolving root", "cid", c, "err", err) continue } if !ok { continue }
out.attempted++ if n.provideOne(ctx, root) { out.succeeded++ } } return out}
// nextProvideInterval picks the delay before the next provide pass: the// short retryInterval if the pass had at least one CID to announce and// every one of them failed (most likely a DHT routing table that's still// cold), so content doesn't sit unannounced for a full reprovideInterval;// reprovideInterval otherwise, including when there was nothing to// provide.func nextProvideInterval(out provideOutcome) time.Duration { if out.attempted > 0 && out.succeeded == 0 { return retryInterval } return reprovideInterval}
// rootToProvide returns the CID that should be announced for blob c: its// UnixFS root if it's been indexed, or the blessed CID itself if it's// small enough to need no DAG. It reports ok=false for a large blob that// hasn't been indexed yet (self-heal hasn't caught up, or keeps failing)// — there's nothing valid to announce for it yet.func (n *Node) rootToProvide(c cid.Cid) (cid.Cid, bool, error) { if root, ok, err := n.index.RootFor(c); err != nil { return cid.Undef, false, err } else if ok { return root, true, nil }
stat, err := n.store.Stat(c) if err != nil { return cid.Undef, false, err } if stat.Size > MaxRawBlockSize { return cid.Undef, false, nil } return c, true, nil}
// provideOne announces a single root, bounded by provideTimeout, and// reports whether it succeeded.func (n *Node) provideOne(ctx context.Context, root cid.Cid) bool { pctx, cancel := context.WithTimeout(ctx, provideTimeout) defer cancel() if err := n.dht.Provide(pctx, root, true); err != nil { slog.Warn("ipfs: provide failed", "cid", root, "err", err) return false } return true}
// AnnounceRoot triggers an immediate, best-effort DHT announcement of// root — a blob's fetchable IPFS root, the same CID uploadFile reports as// ipfsRoot and a completed pin ends up serving — right after it becomes// available, rather than leaving it to wait for the next periodic provide// pass (up to reprovideInterval away; direct-connect Bitswap still works// in the meantime regardless, since it never depends on the DHT). See// uploadfile.go's resolveIPFSRoot (via Indexer.AnnounceRoot) and this// package's own Pin for its two callers — deliberately not xrpc's// listFiles, which resolves a root for its response the same way but as a// pure read (see xrpc's rootForListing): ATFS-d01i is what happened when a// shared helper blurred that line and let a public, unauthenticated GET// reach this method once per listed file, per request.//// It never blocks the caller and never fails visibly: the announce runs on// its own tracked goroutine, through n.announce (n.provideOne in// production — the same single-CID provide every periodic pass itself// uses), so it's bounded by provideTimeout and any failure — most commonly// a still-cold DHT, or this node being offline — is logged there and// otherwise dropped. That goroutine waits for a slot in announceSem before// calling n.announce (see announceConcurrency), so a burst of simultaneous// callers still runs only a handful of concurrent DHT walks rather than one// per call; the wait races n.ctx.Done(), so a call still queued for a slot// when the node starts closing gives up on it rather than outliving// shutdown. The periodic loop already covers the CID from here on: by the// time this is called the blob is in the store, and provideAll walks the// whole store on every pass.//// A no-op once the node is closing, so Close never waits on an// announcement that only just started (see closeMu).func (n *Node) AnnounceRoot(root cid.Cid) { n.closeMu.Lock() if n.closed { n.closeMu.Unlock() return } n.wg.Add(1) n.closeMu.Unlock()
go func() { defer n.wg.Done() select { case n.announceSem <- struct{}{}: defer func() { <-n.announceSem }() case <-n.ctx.Done(): return } n.announce(n.ctx, root) }()}
// Index is this node's UnixFS index. It satisfies both xrpc.Indexer and// gateway.Resolver, so the HTTP transport wiring (separate, later work)// can pass it straight to xrpc.New and gateway.New.func (n *Node) Index() *Indexer { return n.index }
// PeerID is this node's libp2p peer ID.func (n *Node) PeerID() peer.ID { return n.host.ID() }
// Addrs is this node's current listen addresses.func (n *Node) Addrs() []multiaddr.Multiaddr { return n.host.Addrs() }
// Close shuts the node down: the background index/provide loop, then// Bitswap, the DHT, and finally the libp2p host. It's idempotent and safe// to call multiple times, or concurrently with ctx cancellation (Start's// own ctx-triggered shutdown calls this too).func (n *Node) Close() error { n.closeOnce.Do(func() { n.closeMu.Lock() n.closed = true n.closeMu.Unlock()
n.cancel() n.wg.Wait()
var errs []error if err := n.bs.Close(); err != nil { errs = append(errs, err) } if err := n.dht.Close(); err != nil { errs = append(errs, err) } if err := n.host.Close(); err != nil { errs = append(errs, err) } n.closeErr = errors.Join(errs...) }) return n.closeErr}