feat: add experimental on-demand pinning

Add a background checker that automatically pins content when DHT provider counts fall below a configurable replication target and unpins once enough providers exist again after a grace period.

Gated behind Experimental.OnDemandPinningEnabled.
New CLI commands: ipfs pin ondemand {add,rm,ls}

Safety measures:
- storage budget check (respects StorageMax/GCWatermark)
- idle timeout on recursive DAG fetches (2 min without progress)
- pin partitioning via pin name to distinguish on-demand pins from persitent pins.

Author: ihlec

config/config.go

@@ -35,15 +35,16 @@ type Config struct {
 	Migration Migration
 	AutoConf  AutoConf
 
-	Provide       Provide    // Merged Provider and Reprovider configuration
-	Provider      Provider   // Deprecated: use Provide. Will be removed in a future release.
-	Reprovider    Reprovider // Deprecated: use Provide. Will be removed in a future release.
-	HTTPRetrieval HTTPRetrieval
-	Experimental  Experiments
-	Plugins       Plugins
-	Pinning       Pinning
-	Import        Import
-	Version       Version
+	Provide         Provide    // Merged Provider and Reprovider configuration
+	Provider        Provider   // Deprecated: use Provide. Will be removed in a future release.
+	Reprovider      Reprovider // Deprecated: use Provide. Will be removed in a future release.
+	HTTPRetrieval   HTTPRetrieval
+	Experimental    Experiments
+	Plugins         Plugins
+	Pinning         Pinning
+	OnDemandPinning OnDemandPinning
+	Import          Import
+	Version         Version
 
 	Internal Internal // experimental/unstable options
 

config/experiments.go

@@ -10,6 +10,7 @@ type Experiments struct {
 	OptimisticProvide             bool
 	OptimisticProvideJobsPoolSize int
 	GatewayOverLibp2p             bool `json:",omitempty"`
+	OnDemandPinningEnabled        bool
 
 	GraphsyncEnabled     graphsyncEnabled                 `json:",omitempty"`
 	AcceleratedDHTClient experimentalAcceleratedDHTClient `json:",omitempty"`

config/ondemandpin.go

@@ -0,0 +1,20 @@
+package config
+
+import "time"
+
+const (
+	DefaultOnDemandPinReplicationTarget = 5
+	DefaultOnDemandPinCheckInterval     = 10 * time.Minute
+	DefaultOnDemandPinUnpinGracePeriod  = 24 * time.Hour
+)
+
+type OnDemandPinning struct {
+	// Minimum providers desired in the DHT (excluding self).
+	ReplicationTarget OptionalInteger
+
+	// How often the checker evaluates all registered CIDs.
+	CheckInterval OptionalDuration
+
+	// How long replication must stay above target before unpinning.
+	UnpinGracePeriod OptionalDuration
+}

core/commands/commands_test.go

@@ -167,6 +167,10 @@ func TestCommands(t *testing.T) {
 		"/pin/remote/service/add",
 		"/pin/remote/service/ls",
 		"/pin/remote/service/rm",
+		"/pin/ondemand",
+		"/pin/ondemand/add",
+		"/pin/ondemand/rm",
+		"/pin/ondemand/ls",
 		"/pin/rm",
 		"/pin/update",
 		"/pin/verify",

core/commands/pin/ondemandpin.go

@@ -0,0 +1,280 @@
+package pin
+
+import (
+	"fmt"
+	"io"
+	"time"
+
+	cmds "github.com/ipfs/go-ipfs-cmds"
+	"github.com/ipfs/kubo/config"
+	cmdenv "github.com/ipfs/kubo/core/commands/cmdenv"
+	"github.com/ipfs/kubo/core/commands/cmdutils"
+	"github.com/ipfs/kubo/ondemandpin"
+)
+
+const onDemandLiveOptionName = "live"
+
+var onDemandPinCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Manage on-demand pins.",
+		ShortDescription: `
+			On-demand pinning automatically pins content when few target providers exist in the DHT, 
+			and unpins when replication has been above target for a grace period.
+			Requires Experimental.OnDemandPinningEnabled = true.
+			`,
+	},
+	Subcommands: map[string]*cmds.Command{
+		"add": addOnDemandPinCmd,
+		"rm":  rmOnDemandPinCmd,
+		"ls":  listOnDemandPinCmd,
+	},
+}
+
+type OnDemandPinOutput struct {
+	Cid string `json:"Cid"`
+}
+
+var addOnDemandPinCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline:          "Register CIDs for on-demand pinning.",
+		ShortDescription: `Adds the given CID(s) to the on-demand pin registry. The background checker will evaluate replication and pin if needed.`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("cid", true, true, "CID(s) to register."),
+	},
+	Type: OnDemandPinOutput{},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		n, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		cfg, err := n.Repo.Config()
+		if err != nil {
+			return err
+		}
+		if !cfg.Experimental.OnDemandPinningEnabled {
+			return fmt.Errorf("on-demand pinning is not enabled; set Experimental.OnDemandPinningEnabled = true in config")
+		}
+
+		store := n.OnDemandPinStore
+
+		api, err := cmdenv.GetApi(env, req)
+		if err != nil {
+			return err
+		}
+
+		for _, arg := range req.Arguments {
+			p, err := cmdutils.PathOrCidPath(arg)
+			if err != nil {
+				return fmt.Errorf("invalid CID or path %q: %w", arg, err)
+			}
+
+			rp, _, err := api.ResolvePath(req.Context, p)
+			if err != nil {
+				return fmt.Errorf("resolving %q: %w", arg, err)
+			}
+			c := rp.RootCid()
+
+			if err := store.Add(req.Context, c); err != nil {
+				return err
+			}
+
+			if checker := n.OnDemandPinChecker; checker != nil {
+				checker.Enqueue(c)
+			}
+
+			if err := res.Emit(&OnDemandPinOutput{
+				Cid: c.String(),
+			}); err != nil {
+				return err
+			}
+		}
+		return nil
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *OnDemandPinOutput) error {
+			fmt.Fprintf(w, "registered %s for on-demand pinning\n", out.Cid)
+			return nil
+		}),
+	},
+}
+
+var rmOnDemandPinCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Remove CIDs from on-demand pinning.",
+		ShortDescription: `
+			Removes the given CID(s) from the on-demand pin registry. 
+			If the content was pinned by the checker, it is also unpinned.
+			Works even when on-demand pinning is disabled, to clean up previously registered CIDs.
+			`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("cid", true, true, "CID(s) to remove."),
+	},
+	Type: OnDemandPinOutput{},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		n, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		store := n.OnDemandPinStore
+
+		api, err := cmdenv.GetApi(env, req)
+		if err != nil {
+			return err
+		}
+
+		for _, arg := range req.Arguments {
+			p, err := cmdutils.PathOrCidPath(arg)
+			if err != nil {
+				return fmt.Errorf("invalid CID or path %q: %w", arg, err)
+			}
+
+			rp, _, err := api.ResolvePath(req.Context, p)
+			if err != nil {
+				return fmt.Errorf("resolving %q: %w", arg, err)
+			}
+			c := rp.RootCid()
+
+			isOurs, err := ondemandpin.PinHasName(req.Context, n.Pinning, c, ondemandpin.OnDemandPinName)
+			if err != nil {
+				return fmt.Errorf("checking pin state for %s: %w", c, err)
+			}
+			if isOurs {
+				if err := api.Pin().Rm(req.Context, rp); err != nil {
+					return fmt.Errorf("unpinning %s: %w", c, err)
+				}
+			}
+
+			if err := store.Remove(req.Context, c); err != nil {
+				return err
+			}
+
+			if err := res.Emit(&OnDemandPinOutput{
+				Cid: c.String(),
+			}); err != nil {
+				return err
+			}
+		}
+		return nil
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *OnDemandPinOutput) error {
+			fmt.Fprintf(w, "removed %s from on-demand pinning\n", out.Cid)
+			return nil
+		}),
+	},
+}
+
+type OnDemandLsOutput struct {
+	Cid             string `json:"Cid"`
+	PinnedByUs      bool   `json:"PinnedByUs"`
+	Providers       *int   `json:"Providers,omitempty"`
+	LastAboveTarget string `json:"LastAboveTarget,omitempty"`
+	CreatedAt       string `json:"CreatedAt"`
+}
+
+var listOnDemandPinCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "List on-demand pins.",
+		ShortDescription: `
+			Lists CIDs registered for on-demand pinning with their current state.
+			Use --live to include real-time provider counts from the DHT.
+			`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("cid", false, true, "Optional CID(s) to filter."),
+	},
+	Options: []cmds.Option{
+		cmds.BoolOption(onDemandLiveOptionName, "l", "Perform live provider lookup."),
+	},
+	Type: OnDemandLsOutput{},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		n, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		store := n.OnDemandPinStore
+
+		live, _ := req.Options[onDemandLiveOptionName].(bool)
+
+		var globalTarget int
+		if live {
+			cfg, err := n.Repo.Config()
+			if err != nil {
+				return err
+			}
+			globalTarget = int(cfg.OnDemandPinning.ReplicationTarget.WithDefault(config.DefaultOnDemandPinReplicationTarget))
+		}
+
+		var records []ondemandpin.Record
+		if len(req.Arguments) > 0 {
+			api, err := cmdenv.GetApi(env, req)
+			if err != nil {
+				return err
+			}
+			for _, arg := range req.Arguments {
+				p, err := cmdutils.PathOrCidPath(arg)
+				if err != nil {
+					return fmt.Errorf("invalid CID or path %q: %w", arg, err)
+				}
+				rp, _, err := api.ResolvePath(req.Context, p)
+				if err != nil {
+					return fmt.Errorf("resolving %q: %w", arg, err)
+				}
+				rec, err := store.Get(req.Context, rp.RootCid())
+				if err != nil {
+					return err
+				}
+				records = append(records, *rec)
+			}
+		} else {
+			records, err = store.List(req.Context)
+			if err != nil {
+				return err
+			}
+		}
+
+		for _, rec := range records {
+			out := OnDemandLsOutput{
+				Cid:        rec.Cid.String(),
+				PinnedByUs: rec.PinnedByUs,
+				CreatedAt:  rec.CreatedAt.Format(time.RFC3339),
+			}
+			if !rec.LastAboveTarget.IsZero() {
+				out.LastAboveTarget = rec.LastAboveTarget.Format(time.RFC3339)
+			}
+
+			if live && n.Routing != nil {
+				count := ondemandpin.CountProviders(req.Context, n.Routing, n.Identity, rec.Cid, globalTarget)
+				out.Providers = &count
+			}
+
+			if err := res.Emit(&out); err != nil {
+				return err
+			}
+		}
+		return nil
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *OnDemandLsOutput) error {
+			pinState := "not-pinned"
+			if out.PinnedByUs {
+				pinState = "pinned"
+			}
+			fmt.Fprintf(w, "%s", out.Cid)
+			if out.Providers != nil {
+				fmt.Fprintf(w, "  providers=%d", *out.Providers)
+			}
+			fmt.Fprintf(w, "  %s  created=%s", pinState, out.CreatedAt)
+			if out.LastAboveTarget != "" {
+				fmt.Fprintf(w, "  above-target-since=%s", out.LastAboveTarget)
+			}
+			fmt.Fprintln(w)
+			return nil
+		}),
+	},
+}

core/commands/pin/pin.go

@@ -32,12 +32,13 @@ var PinCmd = &cmds.Command{
 	},
 
 	Subcommands: map[string]*cmds.Command{
-		"add":    addPinCmd,
-		"rm":     rmPinCmd,
-		"ls":     listPinCmd,
-		"verify": verifyPinCmd,
-		"update": updatePinCmd,
-		"remote": remotePinCmd,
+		"add":      addPinCmd,
+		"rm":       rmPinCmd,
+		"ls":       listPinCmd,
+		"verify":   verifyPinCmd,
+		"update":   updatePinCmd,
+		"remote":   remotePinCmd,
+		"ondemand": onDemandPinCmd,
 	},
 }
 

core/core.go

@@ -56,6 +56,7 @@ import (
 	"github.com/ipfs/kubo/core/node"
 	"github.com/ipfs/kubo/core/node/libp2p"
 	"github.com/ipfs/kubo/fuse/mount"
+	"github.com/ipfs/kubo/ondemandpin"
 	"github.com/ipfs/kubo/p2p"
 	"github.com/ipfs/kubo/repo"
 	irouting "github.com/ipfs/kubo/routing"
@@ -71,10 +72,12 @@ type IpfsNode struct {
 	Repo repo.Repo
 
 	// Local node
-	Pinning         pin.Pinner             // the pinning manager
-	Mounts          Mounts                 `optional:"true"` // current mount state, if any.
-	PrivateKey      ic.PrivKey             `optional:"true"` // the local node's private Key
-	PNetFingerprint libp2p.PNetFingerprint `optional:"true"` // fingerprint of private network
+	Pinning            pin.Pinner
+	OnDemandPinStore   *ondemandpin.Store
+	OnDemandPinChecker *ondemandpin.Checker   `optional:"true"`
+	Mounts             Mounts                 `optional:"true"` // current mount state, if any.
+	PrivateKey         ic.PrivKey             `optional:"true"` // the local node's private Key
+	PNetFingerprint    libp2p.PNetFingerprint `optional:"true"` // fingerprint of private network
 
 	// Services
 	Peerstore                   pstore.Peerstore          `optional:"true"` // storage for other Peer instances