| 1 | package cmdenv |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "strconv" |
| 7 | "strings" |
| 8 | |
| 9 | "github.com/ipfs/boxo/blockstore" |
| 10 | "github.com/ipfs/boxo/dag/walker" |
| 11 | "github.com/ipfs/go-cid" |
| 12 | cmds "github.com/ipfs/go-ipfs-cmds" |
| 13 | logging "github.com/ipfs/go-log/v2" |
| 14 | "github.com/ipfs/kubo/commands" |
| 15 | "github.com/ipfs/kubo/config" |
| 16 | "github.com/ipfs/kubo/core" |
| 17 | coreiface "github.com/ipfs/kubo/core/coreiface" |
| 18 | options "github.com/ipfs/kubo/core/coreiface/options" |
| 19 | "github.com/ipfs/kubo/core/node" |
| 20 | routing "github.com/libp2p/go-libp2p/core/routing" |
| 21 | ) |
| 22 | |
| 23 | var log = logging.Logger("core/commands/cmdenv") |
| 24 | |
| 25 | // GetNode extracts the node from the environment. |
| 26 | func GetNode(env any) (*core.IpfsNode, error) { |
| 27 | ctx, ok := env.(*commands.Context) |
| 28 | if !ok { |
| 29 | return nil, fmt.Errorf("expected env to be of type %T, got %T", ctx, env) |
| 30 | } |
| 31 | |
| 32 | return ctx.GetNode() |
| 33 | } |
| 34 | |
| 35 | // GetApi extracts CoreAPI instance from the environment. |
| 36 | func GetApi(env cmds.Environment, req *cmds.Request) (coreiface.CoreAPI, error) { //nolint |
| 37 | ctx, ok := env.(*commands.Context) |
| 38 | if !ok { |
| 39 | return nil, fmt.Errorf("expected env to be of type %T, got %T", ctx, env) |
| 40 | } |
| 41 | |
| 42 | offline, _ := req.Options["offline"].(bool) |
| 43 | if !offline { |
| 44 | offline, _ = req.Options["local"].(bool) |
| 45 | if offline { |
| 46 | log.Errorf("Command '%s', --local is deprecated, use --offline instead", strings.Join(req.Path, " ")) |
| 47 | } |
| 48 | } |
| 49 | api, err := ctx.GetAPI() |
| 50 | if err != nil { |
| 51 | return nil, err |
| 52 | } |
| 53 | if offline { |
| 54 | return api.WithOptions(options.Api.Offline(offline)) |
| 55 | } |
| 56 | |
| 57 | return api, nil |
| 58 | } |
| 59 | |
| 60 | // GetConfigRoot extracts the config root from the environment |
| 61 | func GetConfigRoot(env cmds.Environment) (string, error) { |
| 62 | ctx, ok := env.(*commands.Context) |
| 63 | if !ok { |
| 64 | return "", fmt.Errorf("expected env to be of type %T, got %T", ctx, env) |
| 65 | } |
| 66 | |
| 67 | return ctx.ConfigRoot, nil |
| 68 | } |
| 69 | |
| 70 | // EscNonPrint converts non-printable characters and backslash into Go escape |
| 71 | // sequences. This is done to display all characters in a string, including |
| 72 | // those that would otherwise not be displayed or have an undesirable effect on |
| 73 | // the display. |
| 74 | func EscNonPrint(s string) string { |
| 75 | if !needEscape(s) { |
| 76 | return s |
| 77 | } |
| 78 | |
| 79 | esc := strconv.Quote(s) |
| 80 | // Remove first and last quote, and unescape quotes. |
| 81 | return strings.ReplaceAll(esc[1:len(esc)-1], `\"`, `"`) |
| 82 | } |
| 83 | |
| 84 | func needEscape(s string) bool { |
| 85 | if strings.ContainsRune(s, '\\') { |
| 86 | return true |
| 87 | } |
| 88 | for _, r := range s { |
| 89 | if !strconv.IsPrint(r) { |
| 90 | return true |
| 91 | } |
| 92 | } |
| 93 | return false |
| 94 | } |
| 95 | |
| 96 | // provideCIDSync performs a synchronous/blocking provide operation to announce |
| 97 | // the given CID to the DHT. |
| 98 | // |
| 99 | // - If the accelerated DHT client is used, a DHT lookup isn't needed, we |
| 100 | // directly allocate provider records to closest peers. |
| 101 | // - If Provide.DHT.SweepEnabled=true or OptimisticProvide=true, we make an |
| 102 | // optimistic provide call. |
| 103 | // - Else we make a standard provide call (much slower). |
| 104 | // |
| 105 | // IMPORTANT: The caller MUST verify DHT availability using HasActiveDHTClient() |
| 106 | // before calling this function. Calling with a nil or invalid router will cause |
| 107 | // a panic - this is the caller's responsibility to prevent. |
| 108 | func provideCIDSync(ctx context.Context, router routing.Routing, c cid.Cid) error { |
| 109 | return router.Provide(ctx, c, true) |
| 110 | } |
| 111 | |
| 112 | // ExecuteFastProvideRoot immediately provides a root CID to the DHT, bypassing the regular |
| 113 | // provide queue for faster content discovery. This function is reusable across commands |
| 114 | // that add or import content, such as ipfs add and ipfs dag import. |
| 115 | // |
| 116 | // Parameters: |
| 117 | // - ctx: context for synchronous provides |
| 118 | // - ipfsNode: the IPFS node instance |
| 119 | // - cfg: node configuration |
| 120 | // - rootCid: the CID to provide |
| 121 | // - wait: whether to block until provide completes (sync mode) |
| 122 | // - isPinned: whether content is pinned |
| 123 | // - isPinnedRoot: whether this is a pinned root CID |
| 124 | // - isMFS: whether content is in MFS |
| 125 | // |
| 126 | // Return value: |
| 127 | // - Returns nil if operation succeeded or was skipped (preconditions not met) |
| 128 | // - Returns error only in sync mode (wait=true) when provide operation fails |
| 129 | // - In async mode (wait=false), always returns nil (errors logged in goroutine) |
| 130 | // |
| 131 | // The function handles all precondition checks (Provide.Enabled, DHT availability, |
| 132 | // strategy matching) and logs appropriately. In async mode, it launches a goroutine |
| 133 | // with a detached context and timeout. |
| 134 | func ExecuteFastProvideRoot( |
| 135 | ctx context.Context, |
| 136 | ipfsNode *core.IpfsNode, |
| 137 | cfg *config.Config, |
| 138 | rootCid cid.Cid, |
| 139 | wait bool, |
| 140 | isPinned bool, |
| 141 | isPinnedRoot bool, |
| 142 | isMFS bool, |
| 143 | ) error { |
| 144 | log.Debugw("fast-provide-root: enabled", "wait", wait) |
| 145 | |
| 146 | // Check preconditions for providing |
| 147 | switch { |
| 148 | case !cfg.Provide.Enabled.WithDefault(config.DefaultProvideEnabled): |
| 149 | log.Debugw("fast-provide-root: skipped", "reason", "Provide.Enabled is false") |
| 150 | return nil |
| 151 | case !ipfsNode.HasActiveDHTClient(): |
| 152 | log.Debugw("fast-provide-root: skipped", "reason", "DHT not available") |
| 153 | return nil |
| 154 | } |
| 155 | |
| 156 | // Check if strategy allows providing this content |
| 157 | strategyStr := cfg.Provide.Strategy.WithDefault(config.DefaultProvideStrategy) |
| 158 | strategy := config.MustParseProvideStrategy(strategyStr) |
| 159 | shouldProvide := config.ShouldProvideForStrategy(strategy, isPinned, isPinnedRoot, isMFS) |
| 160 | |
| 161 | if !shouldProvide { |
| 162 | log.Debugw("fast-provide-root: skipped", "reason", "strategy does not match content", "strategy", strategyStr, "pinned", isPinned, "pinnedRoot", isPinnedRoot, "mfs", isMFS) |
| 163 | return nil |
| 164 | } |
| 165 | |
| 166 | // Execute provide operation |
| 167 | if wait { |
| 168 | // Synchronous mode: block until provide completes, return error on failure |
| 169 | log.Debugw("fast-provide-root: providing synchronously", "cid", rootCid) |
| 170 | if err := provideCIDSync(ctx, ipfsNode.DHTClient, rootCid); err != nil { |
| 171 | log.Warnw("fast-provide-root: sync provide failed", "cid", rootCid, "error", err) |
| 172 | return fmt.Errorf("fast-provide: %w", err) |
| 173 | } |
| 174 | log.Debugw("fast-provide-root: sync provide completed", "cid", rootCid) |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | // Asynchronous mode (default): fire-and-forget, don't block, always return nil. |
| 179 | // Parent off the node's lifetime context (not context.Background) so the |
| 180 | // goroutine cancels on daemon shutdown instead of potentially outliving |
| 181 | // the node and touching a closed DHT client. The timeout still bounds |
| 182 | // stuck DHT operations. |
| 183 | log.Debugw("fast-provide-root: providing asynchronously", "cid", rootCid) |
| 184 | go func() { |
| 185 | ctx, cancel := context.WithTimeout(ipfsNode.Context(), config.DefaultFastProvideTimeout) |
| 186 | defer cancel() |
| 187 | if err := provideCIDSync(ctx, ipfsNode.DHTClient, rootCid); err != nil { |
| 188 | log.Warnw("fast-provide-root: async provide failed", "cid", rootCid, "error", err) |
| 189 | } else { |
| 190 | log.Debugw("fast-provide-root: async provide completed", "cid", rootCid) |
| 191 | } |
| 192 | }() |
| 193 | return nil |
| 194 | } |
| 195 | |
| 196 | // ExecuteFastProvideDAG walks the DAGs rooted at roots and provides |
| 197 | // CIDs according to the active Provide.Strategy. A single bloom |
| 198 | // tracker is shared across all roots so shared sub-DAGs are |
| 199 | // deduplicated. Uses an unbuffered channel for backpressure. |
| 200 | // |
| 201 | // Context handling: |
| 202 | // - wait=true: the walk runs inline under cmdCtx (the request |
| 203 | // context), so a user Ctrl+C on the command cancels the walk. |
| 204 | // - wait=false: the walk runs in a background goroutine under |
| 205 | // nodeCtx (the IpfsNode lifetime context). This lets the walk |
| 206 | // survive the command handler returning (go-ipfs-cmds cancels |
| 207 | // req.Context on handler exit) while still being cancelled on |
| 208 | // daemon shutdown, so the goroutine does not outlive the node |
| 209 | // and keep the blockstore/provider pinned open. |
| 210 | // |
| 211 | // fpRate is the bloom filter target false-positive rate (1/N), normally |
| 212 | // resolved from cfg.Provide.BloomFPRate by the caller. |
| 213 | // blockCount sizes the bloom filter (pass 0 if unknown). |
| 214 | func ExecuteFastProvideDAG( |
| 215 | cmdCtx context.Context, |
| 216 | nodeCtx context.Context, |
| 217 | roots []cid.Cid, |
| 218 | strategy config.ProvideStrategy, |
| 219 | bs blockstore.Blockstore, |
| 220 | prov node.DHTProvider, |
| 221 | wait bool, |
| 222 | fpRate uint, |
| 223 | blockCount uint, |
| 224 | ) { |
| 225 | if len(roots) == 0 { |
| 226 | return |
| 227 | } |
| 228 | if (strategy&config.ProvideStrategyPinned) == 0 && |
| 229 | (strategy&config.ProvideStrategyMFS) == 0 { |
| 230 | return |
| 231 | } |
| 232 | |
| 233 | do := func(ctx context.Context) { |
| 234 | expectedItems := max(uint(walker.DefaultBloomInitialCapacity), blockCount) |
| 235 | tracker, err := walker.NewBloomTracker(expectedItems, fpRate) |
| 236 | if err != nil { |
| 237 | log.Errorf("fast-provide-dag: bloom tracker: %s", err) |
| 238 | return |
| 239 | } |
| 240 | |
| 241 | ch := make(chan cid.Cid) // unbuffered for backpressure |
| 242 | done := make(chan struct{}) |
| 243 | go func() { |
| 244 | defer close(done) |
| 245 | for c := range ch { |
| 246 | if err := prov.StartProviding(false, c.Hash()); err != nil { |
| 247 | log.Errorf("fast-provide-dag: %s: %s", c, err) |
| 248 | } |
| 249 | } |
| 250 | }() |
| 251 | |
| 252 | emit := func(c cid.Cid) bool { |
| 253 | select { |
| 254 | case ch <- c: |
| 255 | return true |
| 256 | case <-ctx.Done(): |
| 257 | return false |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | opts := []walker.Option{walker.WithVisitedTracker(tracker)} |
| 262 | useEntities := strategy&config.ProvideStrategyEntities != 0 |
| 263 | |
| 264 | if useEntities { |
| 265 | fetch := walker.NodeFetcherFromBlockstore(bs) |
| 266 | for _, root := range roots { |
| 267 | if ctx.Err() != nil { |
| 268 | break |
| 269 | } |
| 270 | _ = walker.WalkEntityRoots(ctx, root, fetch, emit, opts...) |
| 271 | } |
| 272 | } else { |
| 273 | fetch := walker.LinksFetcherFromBlockstore(bs) |
| 274 | for _, root := range roots { |
| 275 | if ctx.Err() != nil { |
| 276 | break |
| 277 | } |
| 278 | _ = walker.WalkDAG(ctx, root, fetch, emit, opts...) |
| 279 | } |
| 280 | } |
| 281 | |
| 282 | close(ch) |
| 283 | <-done |
| 284 | log.Infow("fast-provide-dag: finished", |
| 285 | "providedCIDs", tracker.Count(), |
| 286 | "skippedBranches", tracker.Deduplicated()) |
| 287 | } |
| 288 | |
| 289 | if wait { |
| 290 | do(cmdCtx) |
| 291 | } else { |
| 292 | // Use the node's lifetime context so the walk survives |
| 293 | // the command handler returning (which cancels req.Context) |
| 294 | // but still cancels on daemon shutdown. |
| 295 | go do(nodeCtx) |
| 296 | } |
| 297 | } |