| 1 | package commands |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "os" |
| 9 | "strings" |
| 10 | "text/tabwriter" |
| 11 | "time" |
| 12 | "unicode/utf8" |
| 13 | |
| 14 | humanize "github.com/dustin/go-humanize" |
| 15 | "github.com/ipfs/boxo/dag/walker" |
| 16 | dag "github.com/ipfs/boxo/ipld/merkledag" |
| 17 | boxoprovider "github.com/ipfs/boxo/provider" |
| 18 | cid "github.com/ipfs/go-cid" |
| 19 | cmds "github.com/ipfs/go-ipfs-cmds" |
| 20 | "github.com/ipfs/kubo/config" |
| 21 | "github.com/ipfs/kubo/core/commands/cmdenv" |
| 22 | "github.com/libp2p/go-libp2p-kad-dht/fullrt" |
| 23 | "github.com/libp2p/go-libp2p-kad-dht/provider" |
| 24 | "github.com/libp2p/go-libp2p-kad-dht/provider/buffered" |
| 25 | "github.com/libp2p/go-libp2p-kad-dht/provider/dual" |
| 26 | "github.com/libp2p/go-libp2p-kad-dht/provider/stats" |
| 27 | routing "github.com/libp2p/go-libp2p/core/routing" |
| 28 | "github.com/probe-lab/go-libdht/kad/key" |
| 29 | "golang.org/x/exp/constraints" |
| 30 | "golang.org/x/term" |
| 31 | ) |
| 32 | |
| 33 | const ( |
| 34 | provideQuietOptionName = "quiet" |
| 35 | provideLanOptionName = "lan" |
| 36 | |
| 37 | provideStatAllOptionName = "all" |
| 38 | provideStatCompactOptionName = "compact" |
| 39 | provideStatNetworkOptionName = "network" |
| 40 | provideStatConnectivityOptionName = "connectivity" |
| 41 | provideStatOperationsOptionName = "operations" |
| 42 | provideStatTimingsOptionName = "timings" |
| 43 | provideStatScheduleOptionName = "schedule" |
| 44 | provideStatQueuesOptionName = "queues" |
| 45 | provideStatWorkersOptionName = "workers" |
| 46 | |
| 47 | // lowWorkerThreshold is the threshold below which worker availability warnings are shown |
| 48 | lowWorkerThreshold = 2 |
| 49 | ) |
| 50 | |
| 51 | var ProvideCmd = &cmds.Command{ |
| 52 | Status: cmds.Experimental, |
| 53 | Helptext: cmds.HelpText{ |
| 54 | Tagline: "Control and monitor content providing", |
| 55 | ShortDescription: ` |
| 56 | Control providing operations. |
| 57 | |
| 58 | OVERVIEW: |
| 59 | |
| 60 | The provide system publishes provider records so other peers can discover |
| 61 | which nodes hold each CID. Content is reprovided periodically (every |
| 62 | Provide.DHT.Interval) according to Provide.Strategy. |
| 63 | |
| 64 | CONFIGURATION: |
| 65 | |
| 66 | Learn more: https://github.com/ipfs/kubo/blob/master/docs/config.md#provide |
| 67 | |
| 68 | SEE ALSO: |
| 69 | |
| 70 | For ad-hoc immediate announcements, see 'ipfs provide once'. |
| 71 | `, |
| 72 | }, |
| 73 | |
| 74 | Subcommands: map[string]*cmds.Command{ |
| 75 | "clear": provideClearCmd, |
| 76 | "once": provideOnceCmd, |
| 77 | "stat": provideStatCmd, |
| 78 | }, |
| 79 | } |
| 80 | |
| 81 | var provideClearCmd = &cmds.Command{ |
| 82 | Status: cmds.Experimental, |
| 83 | Helptext: cmds.HelpText{ |
| 84 | Tagline: "Clear all CIDs from the provide queue.", |
| 85 | ShortDescription: ` |
| 86 | Clears the provide queue: CIDs waiting to be advertised to the DHT for the |
| 87 | first time. Does not affect content that is already being reprovided on |
| 88 | schedule. |
| 89 | |
| 90 | Kubo also clears the queue automatically on restart when it detects a |
| 91 | change of Provide.Strategy. |
| 92 | |
| 93 | See: https://github.com/ipfs/kubo/blob/master/docs/config.md#providestrategy |
| 94 | `, |
| 95 | }, |
| 96 | Options: []cmds.Option{ |
| 97 | cmds.BoolOption(provideQuietOptionName, "q", "Do not write output."), |
| 98 | }, |
| 99 | Run: func(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment) error { |
| 100 | n, err := cmdenv.GetNode(env) |
| 101 | if err != nil { |
| 102 | return err |
| 103 | } |
| 104 | |
| 105 | quiet, _ := req.Options[provideQuietOptionName].(bool) |
| 106 | if n.Provider == nil { |
| 107 | return nil |
| 108 | } |
| 109 | |
| 110 | cleared := n.Provider.Clear() |
| 111 | if quiet { |
| 112 | return nil |
| 113 | } |
| 114 | _ = re.Emit(cleared) |
| 115 | |
| 116 | return nil |
| 117 | }, |
| 118 | Type: int(0), |
| 119 | Encoders: cmds.EncoderMap{ |
| 120 | cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, cleared int) error { |
| 121 | quiet, _ := req.Options[provideQuietOptionName].(bool) |
| 122 | if quiet { |
| 123 | return nil |
| 124 | } |
| 125 | |
| 126 | _, err := fmt.Fprintf(w, "removed %d items from provide queue\n", cleared) |
| 127 | return err |
| 128 | }), |
| 129 | }, |
| 130 | } |
| 131 | |
| 132 | // ProvideOnceEvent is emitted once per CID announced by 'ipfs provide once'. |
| 133 | type ProvideOnceEvent struct { |
| 134 | Queued string |
| 135 | } |
| 136 | |
| 137 | var provideOnceCmd = &cmds.Command{ |
| 138 | Status: cmds.Experimental, |
| 139 | Helptext: cmds.HelpText{ |
| 140 | Tagline: "Announce CIDs to the routing system on demand.", |
| 141 | ShortDescription: ` |
| 142 | Publishes provider records for the given CIDs once. The periodic |
| 143 | reprovide schedule (driven by Provide.Strategy and Provide.DHT.Interval) |
| 144 | is left unchanged: CIDs announced here are NOT added to the schedule. |
| 145 | CIDs can be passed as arguments or streamed from stdin (one per line). |
| 146 | |
| 147 | The default sweep provider (Provide.DHT.SweepEnabled=true) submits the CIDs |
| 148 | to its burst-provide queue and returns as each CID is queued; dedicated |
| 149 | burst workers publish the records to the DHT. Use 'ipfs provide stat' to |
| 150 | monitor progress. |
| 151 | |
| 152 | The legacy provider (Provide.DHT.SweepEnabled=false) queues the CIDs for |
| 153 | its serial worker pool, which publishes one CID at a time and may take |
| 154 | significantly longer to complete. |
| 155 | |
| 156 | Use --recursive to walk the DAG and announce every reachable block. With |
| 157 | the default Provide.Strategy=all, every block is already announced, so -r |
| 158 | is only useful with selective strategies like 'roots' or 'pinned+entities'. |
| 159 | |
| 160 | CIDs must already exist in the local blockstore. |
| 161 | |
| 162 | CIDs are deduplicated across arguments, stdin, and DAG walks. Dedup uses |
| 163 | a bloom filter, so at very large scale a small fraction of CIDs may be |
| 164 | skipped (default rate ~1 in 4.75M). |
| 165 | |
| 166 | OUTPUT: |
| 167 | |
| 168 | Output is streamed as each CID is queued. With --enc=json, one |
| 169 | {"Queued": "<cid>"} object is emitted per line. With the text encoder |
| 170 | (default) on a terminal, a single line shows the running count; on a pipe, |
| 171 | a final count is printed at the end. |
| 172 | `, |
| 173 | }, |
| 174 | Arguments: []cmds.Argument{ |
| 175 | cmds.StringArg("cid", true, true, "The CID(s) to announce.").EnableStdin(), |
| 176 | }, |
| 177 | Options: []cmds.Option{ |
| 178 | cmds.BoolOption(recursiveOptionName, "r", "Recursively announce the entire DAG."), |
| 179 | }, |
| 180 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 181 | nd, err := cmdenv.GetNode(env) |
| 182 | if err != nil { |
| 183 | return err |
| 184 | } |
| 185 | if !nd.IsOnline { |
| 186 | return ErrNotOnline |
| 187 | } |
| 188 | cfg, err := nd.Repo.Config() |
| 189 | if err != nil { |
| 190 | return err |
| 191 | } |
| 192 | if !cfg.Provide.Enabled.WithDefault(config.DefaultProvideEnabled) { |
| 193 | return errors.New("cannot provide: Provide.Enabled is false") |
| 194 | } |
| 195 | if len(nd.PeerHost.Network().Conns()) == 0 && !cfg.HasHTTPProviderConfigured() { |
| 196 | return errors.New("cannot provide: no connected peers") |
| 197 | } |
| 198 | |
| 199 | recursive, _ := req.Options[recursiveOptionName].(bool) |
| 200 | |
| 201 | // seen deduplicates across all roots and recursive walks, so a CID |
| 202 | // shared by multiple roots (or repeated in argv/stdin) is announced |
| 203 | // exactly once per invocation. The bloom autoscales as more CIDs |
| 204 | // arrive, keeping memory bounded for arbitrarily large inputs at |
| 205 | // the cost of a small false-positive rate (default ~1 in 4.75M) |
| 206 | // that may cause an occasional CID to be skipped. |
| 207 | seen, err := walker.NewBloomTracker(walker.MinBloomCapacity, walker.DefaultBloomFPRate) |
| 208 | if err != nil { |
| 209 | return err |
| 210 | } |
| 211 | |
| 212 | // announce queues a single CID into the provide system and emits one |
| 213 | // event for it. Uses ProvideOnce so the CID is published without |
| 214 | // being added to the keystore: the periodic reprovide schedule |
| 215 | // (driven by Provide.Strategy) is unaffected. Errors propagate to |
| 216 | // the caller. |
| 217 | announce := func(c cid.Cid) error { |
| 218 | if err := nd.Provider.ProvideOnce(c.Hash()); err != nil { |
| 219 | return err |
| 220 | } |
| 221 | return res.Emit(&ProvideOnceEvent{Queued: c.String()}) |
| 222 | } |
| 223 | |
| 224 | // processRoot validates a root CID against the local blockstore and |
| 225 | // announces either just that CID or every block reachable from it. |
| 226 | processRoot := func(arg string) error { |
| 227 | c, err := cid.Decode(arg) |
| 228 | if err != nil { |
| 229 | return fmt.Errorf("invalid CID %q: %w", arg, err) |
| 230 | } |
| 231 | has, err := nd.Blockstore.Has(req.Context, c) |
| 232 | if err != nil { |
| 233 | return err |
| 234 | } |
| 235 | if !has { |
| 236 | return fmt.Errorf("block %s not found locally, cannot provide", c) |
| 237 | } |
| 238 | |
| 239 | if !recursive { |
| 240 | if !seen.Visit(c) { |
| 241 | return nil |
| 242 | } |
| 243 | return announce(c) |
| 244 | } |
| 245 | |
| 246 | // Stream per-block: visit emits as it walks. Cancel the walk on |
| 247 | // the first announce error so we don't keep fetching DAG nodes |
| 248 | // after we've already failed. |
| 249 | ctx, cancel := context.WithCancel(req.Context) |
| 250 | defer cancel() |
| 251 | var visitErr error |
| 252 | walkErr := dag.Walk(ctx, dag.GetLinksDirect(nd.DAG), c, func(child cid.Cid) bool { |
| 253 | // Skip subtrees we've already walked from a previous root or |
| 254 | // argument: returning false stops descent into this node. |
| 255 | if !seen.Visit(child) { |
| 256 | return false |
| 257 | } |
| 258 | if err := announce(child); err != nil { |
| 259 | visitErr = err |
| 260 | cancel() |
| 261 | return false |
| 262 | } |
| 263 | return true |
| 264 | }) |
| 265 | if visitErr != nil { |
| 266 | return visitErr |
| 267 | } |
| 268 | return walkErr |
| 269 | } |
| 270 | |
| 271 | args := argumentIterator{req.Arguments, req.BodyArgs()} |
| 272 | for { |
| 273 | arg, ok := args.next() |
| 274 | if !ok { |
| 275 | break |
| 276 | } |
| 277 | if err := processRoot(arg); err != nil { |
| 278 | return err |
| 279 | } |
| 280 | } |
| 281 | return args.err() |
| 282 | }, |
| 283 | PostRun: cmds.PostRunMap{ |
| 284 | cmds.CLI: func(res cmds.Response, re cmds.ResponseEmitter) error { |
| 285 | // In text mode we render the running counter and final summary |
| 286 | // directly to stderr/stdout, bypassing the encoder so the TTY |
| 287 | // redraw works. For other encoders (json, xml) we must let the |
| 288 | // encoder serialize each event, so forward the stream as-is. |
| 289 | if enc, _ := res.Request().Options[cmds.EncLong].(string); enc != "" && enc != cmds.Text { |
| 290 | return cmds.Copy(re, res) |
| 291 | } |
| 292 | |
| 293 | // Text mode: render directly to stderr/stdout below. Do not |
| 294 | // call re.Emit from this branch, or output will race with the |
| 295 | // running counter. |
| 296 | isTTY := term.IsTerminal(int(os.Stderr.Fd())) |
| 297 | var count int |
| 298 | for { |
| 299 | v, err := res.Next() |
| 300 | if err == io.EOF { |
| 301 | break |
| 302 | } |
| 303 | if err != nil { |
| 304 | if isTTY && count > 0 { |
| 305 | fmt.Fprintln(os.Stderr) |
| 306 | } |
| 307 | return err |
| 308 | } |
| 309 | if _, ok := v.(*ProvideOnceEvent); !ok { |
| 310 | log.Errorf("provide once postrun: received unexpected type %T", v) |
| 311 | continue |
| 312 | } |
| 313 | count++ |
| 314 | if isTTY { |
| 315 | fmt.Fprintf(os.Stderr, "\rqueued %d CID(s) for immediate provide", count) |
| 316 | } |
| 317 | } |
| 318 | if isTTY && count > 0 { |
| 319 | fmt.Fprintln(os.Stderr) |
| 320 | } else { |
| 321 | fmt.Fprintf(os.Stdout, "queued %d CID(s) for immediate provide\n", count) |
| 322 | } |
| 323 | return nil |
| 324 | }, |
| 325 | }, |
| 326 | Type: ProvideOnceEvent{}, |
| 327 | Encoders: cmds.EncoderMap{ |
| 328 | // Used when PostRun is not invoked (HTTP API consumers in text mode). |
| 329 | // One CID per line keeps the stream pipe-friendly. |
| 330 | cmds.Text: cmds.MakeTypedEncoder(func(_ *cmds.Request, w io.Writer, e *ProvideOnceEvent) error { |
| 331 | _, err := fmt.Fprintf(w, "%s\n", e.Queued) |
| 332 | return err |
| 333 | }), |
| 334 | }, |
| 335 | } |
| 336 | |
| 337 | type provideStats struct { |
| 338 | Sweep *stats.Stats |
| 339 | Legacy *boxoprovider.ReproviderStats |
| 340 | FullRT bool // only used for legacy stats |
| 341 | } |
| 342 | |
| 343 | // extractSweepingProvider extracts a SweepingProvider from the given provider interface. |
| 344 | // It handles unwrapping buffered and dual providers, selecting LAN or WAN as specified. |
| 345 | // Returns nil if the provider is not a sweeping provider type. |
| 346 | func extractSweepingProvider(prov any, useLAN bool) *provider.SweepingProvider { |
| 347 | switch p := prov.(type) { |
| 348 | case *provider.SweepingProvider: |
| 349 | return p |
| 350 | case *dual.SweepingProvider: |
| 351 | if useLAN { |
| 352 | return p.LAN |
| 353 | } |
| 354 | return p.WAN |
| 355 | case *buffered.SweepingProvider: |
| 356 | // Recursively extract from the inner provider |
| 357 | return extractSweepingProvider(p.Provider, useLAN) |
| 358 | default: |
| 359 | return nil |
| 360 | } |
| 361 | } |
| 362 | |
| 363 | var provideStatCmd = &cmds.Command{ |
| 364 | Status: cmds.Experimental, |
| 365 | Helptext: cmds.HelpText{ |
| 366 | Tagline: "Show statistics about the provide system", |
| 367 | ShortDescription: ` |
| 368 | Returns statistics about the node's provide system. |
| 369 | |
| 370 | OVERVIEW: |
| 371 | |
| 372 | The provide system publishes provider records mapping CIDs to your peer |
| 373 | ID. Records expire after a fixed TTL, so the system reprovides them on a |
| 374 | schedule to keep content discoverable. |
| 375 | |
| 376 | Two provider types exist: |
| 377 | |
| 378 | - Sweep provider (default): divides the DHT keyspace into regions and |
| 379 | sweeps through them over the reprovide interval. Batches CIDs that map |
| 380 | to the same DHT servers, reducing lookups from N (one per CID) to a |
| 381 | small constant based on DHT size (~3k for 10k DHT servers). Spreads work |
| 382 | evenly over time and announces records just before they expire. |
| 383 | |
| 384 | - Legacy provider: announces each CID with a separate DHT lookup. Tries |
| 385 | to reprovide all content as fast as possible at each cycle start. Fine |
| 386 | for small datasets, slow past a few thousand CIDs. |
| 387 | |
| 388 | Learn more: |
| 389 | - Config: https://github.com/ipfs/kubo/blob/master/docs/config.md#provide |
| 390 | - Metrics: https://github.com/ipfs/kubo/blob/master/docs/provide-stats.md |
| 391 | |
| 392 | DEFAULT OUTPUT: |
| 393 | |
| 394 | Shows a brief summary including queue sizes, scheduled items, average record |
| 395 | holders, ongoing/total provides, and worker warnings. |
| 396 | |
| 397 | DETAILED OUTPUT: |
| 398 | |
| 399 | Use --all for detailed statistics with these sections: connectivity, queues, |
| 400 | schedule, timings, network, operations, and workers. Individual sections can |
| 401 | be displayed with their flags (e.g., --network, --operations). Multiple flags |
| 402 | can be combined. |
| 403 | |
| 404 | Use --compact for monitoring-friendly 2-column output (requires --all). |
| 405 | |
| 406 | EXAMPLES: |
| 407 | |
| 408 | Monitor provider statistics in real-time with 2-column layout: |
| 409 | |
| 410 | watch ipfs provide stat --all --compact |
| 411 | |
| 412 | Get statistics in JSON format for programmatic processing: |
| 413 | |
| 414 | ipfs provide stat --enc=json | jq |
| 415 | |
| 416 | NOTES: |
| 417 | |
| 418 | - This interface is experimental and may change between releases |
| 419 | - Legacy provider shows basic stats only (no flags supported) |
| 420 | - "Regions" are keyspace divisions for spreading reprovide work |
| 421 | - For Dual DHT: use --lan for LAN provider stats (default is WAN) |
| 422 | `, |
| 423 | }, |
| 424 | Arguments: []cmds.Argument{}, |
| 425 | Options: []cmds.Option{ |
| 426 | cmds.BoolOption(provideLanOptionName, "Show stats for LAN DHT only (for Sweep+Dual DHT only)"), |
| 427 | cmds.BoolOption(provideStatAllOptionName, "a", "Display all provide sweep stats"), |
| 428 | cmds.BoolOption(provideStatCompactOptionName, "Display stats in 2-column layout (requires --all)"), |
| 429 | cmds.BoolOption(provideStatConnectivityOptionName, "Display DHT connectivity status"), |
| 430 | cmds.BoolOption(provideStatNetworkOptionName, "Display network stats (peers, reachability, region size)"), |
| 431 | cmds.BoolOption(provideStatScheduleOptionName, "Display reprovide schedule (CIDs/regions scheduled, next reprovide time)"), |
| 432 | cmds.BoolOption(provideStatTimingsOptionName, "Display timing information (uptime, cycle start, reprovide interval)"), |
| 433 | cmds.BoolOption(provideStatWorkersOptionName, "Display worker pool stats (active/available/queued workers)"), |
| 434 | cmds.BoolOption(provideStatOperationsOptionName, "Display operation stats (ongoing/past provides, rates, errors)"), |
| 435 | cmds.BoolOption(provideStatQueuesOptionName, "Display provide and reprovide queue sizes"), |
| 436 | }, |
| 437 | Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 438 | nd, err := cmdenv.GetNode(env) |
| 439 | if err != nil { |
| 440 | return err |
| 441 | } |
| 442 | |
| 443 | if !nd.IsOnline { |
| 444 | return ErrNotOnline |
| 445 | } |
| 446 | |
| 447 | lanStats, _ := req.Options[provideLanOptionName].(bool) |
| 448 | |
| 449 | // Handle legacy provider |
| 450 | if legacySys, ok := nd.Provider.(boxoprovider.System); ok { |
| 451 | if lanStats { |
| 452 | return errors.New("LAN stats only available for Sweep provider with Dual DHT") |
| 453 | } |
| 454 | stats, err := legacySys.Stat() |
| 455 | if err != nil { |
| 456 | return err |
| 457 | } |
| 458 | _, fullRT := nd.DHTClient.(*fullrt.FullRT) |
| 459 | return res.Emit(provideStats{Legacy: &stats, FullRT: fullRT}) |
| 460 | } |
| 461 | |
| 462 | // Extract sweeping provider (handles buffered and dual unwrapping) |
| 463 | sweepingProvider := extractSweepingProvider(nd.Provider, lanStats) |
| 464 | if sweepingProvider == nil { |
| 465 | if lanStats { |
| 466 | return errors.New("LAN stats only available for Sweep provider with Dual DHT") |
| 467 | } |
| 468 | return fmt.Errorf("stats not available with current routing system %T", nd.Provider) |
| 469 | } |
| 470 | |
| 471 | s, err := sweepingProvider.Stats(req.Context) |
| 472 | if err != nil { |
| 473 | return err |
| 474 | } |
| 475 | return res.Emit(provideStats{Sweep: &s}) |
| 476 | }, |
| 477 | Encoders: cmds.EncoderMap{ |
| 478 | cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, s provideStats) error { |
| 479 | wtr := tabwriter.NewWriter(w, 1, 2, 1, ' ', 0) |
| 480 | defer wtr.Flush() |
| 481 | |
| 482 | all, _ := req.Options[provideStatAllOptionName].(bool) |
| 483 | compact, _ := req.Options[provideStatCompactOptionName].(bool) |
| 484 | connectivity, _ := req.Options[provideStatConnectivityOptionName].(bool) |
| 485 | queues, _ := req.Options[provideStatQueuesOptionName].(bool) |
| 486 | schedule, _ := req.Options[provideStatScheduleOptionName].(bool) |
| 487 | network, _ := req.Options[provideStatNetworkOptionName].(bool) |
| 488 | timings, _ := req.Options[provideStatTimingsOptionName].(bool) |
| 489 | operations, _ := req.Options[provideStatOperationsOptionName].(bool) |
| 490 | workers, _ := req.Options[provideStatWorkersOptionName].(bool) |
| 491 | |
| 492 | flagCount := 0 |
| 493 | for _, enabled := range []bool{all, connectivity, queues, schedule, network, timings, operations, workers} { |
| 494 | if enabled { |
| 495 | flagCount++ |
| 496 | } |
| 497 | } |
| 498 | |
| 499 | if s.Legacy != nil { |
| 500 | if flagCount > 0 { |
| 501 | return errors.New("cannot use flags with legacy provide stats") |
| 502 | } |
| 503 | fmt.Fprintf(wtr, "TotalReprovides:\t%s\n", humanNumber(s.Legacy.TotalReprovides)) |
| 504 | fmt.Fprintf(wtr, "AvgReprovideDuration:\t%s\n", humanDuration(s.Legacy.AvgReprovideDuration)) |
| 505 | fmt.Fprintf(wtr, "LastReprovideDuration:\t%s\n", humanDuration(s.Legacy.LastReprovideDuration)) |
| 506 | if !s.Legacy.LastRun.IsZero() { |
| 507 | fmt.Fprintf(wtr, "LastReprovide:\t%s\n", humanTime(s.Legacy.LastRun)) |
| 508 | if s.FullRT { |
| 509 | fmt.Fprintf(wtr, "NextReprovide:\t%s\n", humanTime(s.Legacy.LastRun.Add(s.Legacy.ReprovideInterval))) |
| 510 | } |
| 511 | } |
| 512 | return nil |
| 513 | } |
| 514 | |
| 515 | if s.Sweep == nil { |
| 516 | return errors.New("no provide stats available") |
| 517 | } |
| 518 | |
| 519 | // Sweep provider stats |
| 520 | if s.Sweep.Closed { |
| 521 | fmt.Fprintf(wtr, "Provider is closed\n") |
| 522 | return nil |
| 523 | } |
| 524 | |
| 525 | if compact && !all { |
| 526 | return errors.New("--compact requires --all flag") |
| 527 | } |
| 528 | |
| 529 | brief := flagCount == 0 |
| 530 | showHeadings := flagCount > 1 || all |
| 531 | |
| 532 | compactMode := all && compact |
| 533 | var cols [2][]string |
| 534 | col0MaxWidth := 0 |
| 535 | // formatLine handles both normal and compact output modes: |
| 536 | // - Normal mode: all lines go to cols[0], col parameter is ignored |
| 537 | // - Compact mode: col 0 for left column, col 1 for right column |
| 538 | formatLine := func(col int, format string, a ...any) { |
| 539 | if compactMode { |
| 540 | s := fmt.Sprintf(format, a...) |
| 541 | cols[col] = append(cols[col], s) |
| 542 | if col == 0 { |
| 543 | col0MaxWidth = max(col0MaxWidth, utf8.RuneCountInString(s)) |
| 544 | } |
| 545 | return |
| 546 | } |
| 547 | format = strings.Replace(format, ": ", ":\t", 1) |
| 548 | format = strings.Replace(format, ", ", ",\t", 1) |
| 549 | cols[0] = append(cols[0], fmt.Sprintf(format, a...)) |
| 550 | } |
| 551 | addBlankLine := func(col int) { |
| 552 | if !brief { |
| 553 | formatLine(col, "") |
| 554 | } |
| 555 | } |
| 556 | sectionTitle := func(col int, title string) { |
| 557 | if !brief && showHeadings { |
| 558 | formatLine(col, "%s:", title) |
| 559 | } |
| 560 | } |
| 561 | |
| 562 | indent := " " |
| 563 | if brief || !showHeadings { |
| 564 | indent = "" |
| 565 | } |
| 566 | |
| 567 | // Connectivity |
| 568 | if all || connectivity || brief && s.Sweep.Connectivity.Status != "online" { |
| 569 | sectionTitle(1, "Connectivity") |
| 570 | since := s.Sweep.Connectivity.Since |
| 571 | if since.IsZero() { |
| 572 | formatLine(1, "%sStatus: %s", indent, s.Sweep.Connectivity.Status) |
| 573 | } else { |
| 574 | formatLine(1, "%sStatus: %s (%s)", indent, s.Sweep.Connectivity.Status, humanTime(since)) |
| 575 | } |
| 576 | addBlankLine(1) |
| 577 | } |
| 578 | |
| 579 | // Queues |
| 580 | if all || queues || brief { |
| 581 | sectionTitle(1, "Queues") |
| 582 | formatLine(1, "%sProvide queue: %s CIDs, %s regions", indent, humanSI(s.Sweep.Queues.PendingKeyProvides, 1), humanSI(s.Sweep.Queues.PendingRegionProvides, 1)) |
| 583 | formatLine(1, "%sReprovide queue: %s regions", indent, humanSI(s.Sweep.Queues.PendingRegionReprovides, 1)) |
| 584 | addBlankLine(1) |
| 585 | } |
| 586 | |
| 587 | // Schedule |
| 588 | if all || schedule || brief { |
| 589 | sectionTitle(0, "Schedule") |
| 590 | formatLine(0, "%sCIDs scheduled: %s", indent, humanNumber(s.Sweep.Schedule.Keys)) |
| 591 | formatLine(0, "%sRegions scheduled: %s", indent, humanNumberOrNA(s.Sweep.Schedule.Regions)) |
| 592 | if !brief { |
| 593 | formatLine(0, "%sAvg prefix length: %s", indent, humanFloatOrNA(s.Sweep.Schedule.AvgPrefixLength)) |
| 594 | nextPrefix := key.BitString(s.Sweep.Schedule.NextReprovidePrefix) |
| 595 | if nextPrefix == "" { |
| 596 | nextPrefix = "N/A" |
| 597 | } |
| 598 | formatLine(0, "%sNext region prefix: %s", indent, nextPrefix) |
| 599 | nextReprovideAt := s.Sweep.Schedule.NextReprovideAt.Format("15:04:05") |
| 600 | if s.Sweep.Schedule.NextReprovideAt.IsZero() { |
| 601 | nextReprovideAt = "N/A" |
| 602 | } |
| 603 | formatLine(0, "%sNext region reprovide: %s", indent, nextReprovideAt) |
| 604 | } |
| 605 | addBlankLine(0) |
| 606 | } |
| 607 | |
| 608 | // Timings |
| 609 | if all || timings { |
| 610 | sectionTitle(1, "Timings") |
| 611 | formatLine(1, "%sUptime: %s (%s)", indent, humanDuration(s.Sweep.Timing.Uptime), humanTime(time.Now().Add(-s.Sweep.Timing.Uptime))) |
| 612 | formatLine(1, "%sCurrent time offset: %s", indent, humanDuration(s.Sweep.Timing.CurrentTimeOffset)) |
| 613 | formatLine(1, "%sCycle started: %s", indent, humanTime(s.Sweep.Timing.CycleStart)) |
| 614 | formatLine(1, "%sReprovide interval: %s", indent, humanDuration(s.Sweep.Timing.ReprovidesInterval)) |
| 615 | addBlankLine(1) |
| 616 | } |
| 617 | |
| 618 | // Network |
| 619 | if all || network || brief { |
| 620 | sectionTitle(0, "Network") |
| 621 | formatLine(0, "%sAvg record holders: %s", indent, humanFloatOrNA(s.Sweep.Network.AvgHolders)) |
| 622 | if !brief { |
| 623 | formatLine(0, "%sPeers swept: %s", indent, humanInt(s.Sweep.Network.Peers)) |
| 624 | formatLine(0, "%sFull keyspace coverage: %t", indent, s.Sweep.Network.CompleteKeyspaceCoverage) |
| 625 | if s.Sweep.Network.Peers > 0 { |
| 626 | formatLine(0, "%sReachable peers: %s (%s%%)", indent, humanInt(s.Sweep.Network.Reachable), humanNumber(100*s.Sweep.Network.Reachable/s.Sweep.Network.Peers)) |
| 627 | } else { |
| 628 | formatLine(0, "%sReachable peers: %s", indent, humanInt(s.Sweep.Network.Reachable)) |
| 629 | } |
| 630 | formatLine(0, "%sAvg region size: %s", indent, humanFloatOrNA(s.Sweep.Network.AvgRegionSize)) |
| 631 | formatLine(0, "%sReplication factor: %s", indent, humanNumber(s.Sweep.Network.ReplicationFactor)) |
| 632 | addBlankLine(0) |
| 633 | } |
| 634 | } |
| 635 | |
| 636 | // Operations |
| 637 | if all || operations || brief { |
| 638 | sectionTitle(1, "Operations") |
| 639 | // Ongoing operations |
| 640 | formatLine(1, "%sOngoing provides: %s CIDs, %s regions", indent, humanSI(s.Sweep.Operations.Ongoing.KeyProvides, 1), humanSI(s.Sweep.Operations.Ongoing.RegionProvides, 1)) |
| 641 | formatLine(1, "%sOngoing reprovides: %s CIDs, %s regions", indent, humanSI(s.Sweep.Operations.Ongoing.KeyReprovides, 1), humanSI(s.Sweep.Operations.Ongoing.RegionReprovides, 1)) |
| 642 | // Past operations summary |
| 643 | formatLine(1, "%sTotal CIDs provided: %s", indent, humanNumber(s.Sweep.Operations.Past.KeysProvided)) |
| 644 | if !brief { |
| 645 | formatLine(1, "%sTotal records provided: %s", indent, humanNumber(s.Sweep.Operations.Past.RecordsProvided)) |
| 646 | formatLine(1, "%sTotal provide errors: %s", indent, humanNumber(s.Sweep.Operations.Past.KeysFailed)) |
| 647 | formatLine(1, "%sCIDs provided/min/worker: %s", indent, humanFloatOrNA(s.Sweep.Operations.Past.KeysProvidedPerMinute)) |
| 648 | formatLine(1, "%sCIDs reprovided/min/worker: %s", indent, humanFloatOrNA(s.Sweep.Operations.Past.KeysReprovidedPerMinute)) |
| 649 | formatLine(1, "%sRegion reprovide duration: %s", indent, humanDurationOrNA(s.Sweep.Operations.Past.RegionReprovideDuration)) |
| 650 | formatLine(1, "%sAvg CIDs/reprovide: %s", indent, humanFloatOrNA(s.Sweep.Operations.Past.AvgKeysPerReprovide)) |
| 651 | formatLine(1, "%sRegions reprovided (last cycle): %s", indent, humanNumber(s.Sweep.Operations.Past.RegionReprovidedLastCycle)) |
| 652 | addBlankLine(1) |
| 653 | } |
| 654 | } |
| 655 | |
| 656 | // Workers |
| 657 | displayWorkers := all || workers |
| 658 | if displayWorkers || brief { |
| 659 | availableReservedBurst := max(0, s.Sweep.Workers.DedicatedBurst-s.Sweep.Workers.ActiveBurst) |
| 660 | availableReservedPeriodic := max(0, s.Sweep.Workers.DedicatedPeriodic-s.Sweep.Workers.ActivePeriodic) |
| 661 | availableFreeWorkers := max(0, s.Sweep.Workers.Max-max(s.Sweep.Workers.DedicatedBurst, s.Sweep.Workers.ActiveBurst)-max(s.Sweep.Workers.DedicatedPeriodic, s.Sweep.Workers.ActivePeriodic)) |
| 662 | availableBurst := availableFreeWorkers + availableReservedBurst |
| 663 | availablePeriodic := availableFreeWorkers + availableReservedPeriodic |
| 664 | |
| 665 | if displayWorkers || availableBurst <= lowWorkerThreshold || availablePeriodic <= lowWorkerThreshold { |
| 666 | // Either we want to display workers information, or we are low on |
| 667 | // available workers and want to warn the user. |
| 668 | sectionTitle(0, "Workers") |
| 669 | specifyWorkers := " workers" |
| 670 | if compactMode { |
| 671 | specifyWorkers = "" |
| 672 | } |
| 673 | formatLine(0, "%sActive%s: %s / %s (max)", indent, specifyWorkers, humanInt(s.Sweep.Workers.Active), humanInt(s.Sweep.Workers.Max)) |
| 674 | if brief { |
| 675 | // Brief mode - show condensed worker info |
| 676 | formatLine(0, "%sPeriodic%s: %s active, %s available, %s queued", indent, specifyWorkers, |
| 677 | humanInt(s.Sweep.Workers.ActivePeriodic), humanInt(availablePeriodic), humanInt(s.Sweep.Workers.QueuedPeriodic)) |
| 678 | formatLine(0, "%sBurst%s: %s active, %s available, %s queued\n", indent, specifyWorkers, |
| 679 | humanInt(s.Sweep.Workers.ActiveBurst), humanInt(availableBurst), humanInt(s.Sweep.Workers.QueuedBurst)) |
| 680 | } else { |
| 681 | formatLine(0, "%sFree%s: %s", indent, specifyWorkers, humanInt(availableFreeWorkers)) |
| 682 | formatLine(0, "%s %-14s %-9s %s", indent, "Workers stats:", "Periodic", "Burst") |
| 683 | formatLine(0, "%s %-14s %-9s %s", indent, "Active:", humanInt(s.Sweep.Workers.ActivePeriodic), humanInt(s.Sweep.Workers.ActiveBurst)) |
| 684 | formatLine(0, "%s %-14s %-9s %s", indent, "Dedicated:", humanInt(s.Sweep.Workers.DedicatedPeriodic), humanInt(s.Sweep.Workers.DedicatedBurst)) |
| 685 | formatLine(0, "%s %-14s %-9s %s", indent, "Available:", humanInt(availablePeriodic), humanInt(availableBurst)) |
| 686 | formatLine(0, "%s %-14s %-9s %s", indent, "Queued:", humanInt(s.Sweep.Workers.QueuedPeriodic), humanInt(s.Sweep.Workers.QueuedBurst)) |
| 687 | formatLine(0, "%sMax connections/worker: %s", indent, humanInt(s.Sweep.Workers.MaxProvideConnsPerWorker)) |
| 688 | addBlankLine(0) |
| 689 | } |
| 690 | } |
| 691 | } |
| 692 | if compactMode { |
| 693 | col0Width := col0MaxWidth + 2 |
| 694 | // Print both columns side by side |
| 695 | maxRows := max(len(cols[0]), len(cols[1])) |
| 696 | if maxRows == 0 { |
| 697 | return nil |
| 698 | } |
| 699 | for i := range maxRows - 1 { // last line is empty |
| 700 | var left, right string |
| 701 | if i < len(cols[0]) { |
| 702 | left = cols[0][i] |
| 703 | } |
| 704 | if i < len(cols[1]) { |
| 705 | right = cols[1][i] |
| 706 | } |
| 707 | fmt.Fprintf(wtr, "%-*s %s\n", col0Width, left, right) |
| 708 | } |
| 709 | } else { |
| 710 | if !brief { |
| 711 | cols[0] = cols[0][:len(cols[0])-1] // remove last blank line |
| 712 | } |
| 713 | for _, line := range cols[0] { |
| 714 | fmt.Fprintln(wtr, line) |
| 715 | } |
| 716 | } |
| 717 | return nil |
| 718 | }), |
| 719 | }, |
| 720 | Type: provideStats{}, |
| 721 | } |
| 722 | |
| 723 | func humanDuration(val time.Duration) string { |
| 724 | if val > time.Second { |
| 725 | return val.Truncate(100 * time.Millisecond).String() |
| 726 | } |
| 727 | return val.Truncate(time.Microsecond).String() |
| 728 | } |
| 729 | |
| 730 | func humanDurationOrNA(val time.Duration) string { |
| 731 | if val <= 0 { |
| 732 | return "N/A" |
| 733 | } |
| 734 | return humanDuration(val) |
| 735 | } |
| 736 | |
| 737 | func humanTime(val time.Time) string { |
| 738 | if val.IsZero() { |
| 739 | return "N/A" |
| 740 | } |
| 741 | return val.Format("2006-01-02 15:04:05") |
| 742 | } |
| 743 | |
| 744 | func humanNumber[T constraints.Float | constraints.Integer](n T) string { |
| 745 | nf := float64(n) |
| 746 | str := humanSI(nf, 0) |
| 747 | fullStr := humanFull(nf, 0) |
| 748 | if str != fullStr { |
| 749 | return fmt.Sprintf("%s\t(%s)", str, fullStr) |
| 750 | } |
| 751 | return str |
| 752 | } |
| 753 | |
| 754 | // humanNumberOrNA is like humanNumber but returns "N/A" for non-positive values. |
| 755 | func humanNumberOrNA[T constraints.Float | constraints.Integer](n T) string { |
| 756 | if n <= 0 { |
| 757 | return "N/A" |
| 758 | } |
| 759 | return humanNumber(n) |
| 760 | } |
| 761 | |
| 762 | // humanFloatOrNA formats a float with 1 decimal place, returning "N/A" for non-positive values. |
| 763 | // This is separate from humanNumberOrNA because it provides simple decimal formatting for |
| 764 | // continuous metrics (averages, rates) rather than SI unit formatting used for discrete counts. |
| 765 | func humanFloatOrNA(val float64) string { |
| 766 | if val <= 0 { |
| 767 | return "N/A" |
| 768 | } |
| 769 | return humanFull(val, 1) |
| 770 | } |
| 771 | |
| 772 | func humanSI[T constraints.Float | constraints.Integer](val T, decimals int) string { |
| 773 | v, unit := humanize.ComputeSI(float64(val)) |
| 774 | return fmt.Sprintf("%s%s", humanFull(v, decimals), unit) |
| 775 | } |
| 776 | |
| 777 | func humanInt[T constraints.Integer](val T) string { |
| 778 | return humanFull(float64(val), 0) |
| 779 | } |
| 780 | |
| 781 | func humanFull(val float64, decimals int) string { |
| 782 | return humanize.CommafWithDigits(val, decimals) |
| 783 | } |
| 784 | |
| 785 | // provideCIDSync performs a synchronous/blocking provide operation to announce |
| 786 | // the given CID to the DHT. |
| 787 | // |
| 788 | // - If the accelerated DHT client is used, a DHT lookup isn't needed, we |
| 789 | // directly allocate provider records to closest peers. |
| 790 | // - If Provide.DHT.SweepEnabled=true or OptimisticProvide=true, we make an |
| 791 | // optimistic provide call. |
| 792 | // - Else we make a standard provide call (much slower). |
| 793 | // |
| 794 | // IMPORTANT: The caller MUST verify DHT availability using HasActiveDHTClient() |
| 795 | // before calling this function. Calling with a nil or invalid router will cause |
| 796 | // a panic - this is the caller's responsibility to prevent. |
| 797 | func provideCIDSync(ctx context.Context, router routing.Routing, c cid.Cid) error { |
| 798 | return router.Provide(ctx, c, true) |
| 799 | } |