feat(provide): +unique and +entities strategy modifiers

config/import.go

@@ -23,6 +23,7 @@ const (
 	DefaultHashFunction    = "sha2-256"
 	DefaultFastProvideRoot = true
 	DefaultFastProvideWait = false
+	DefaultFastProvideDAG  = false
 
 	DefaultUnixFSHAMTDirectorySizeThreshold = 262144 // 256KiB - https://github.com/ipfs/boxo/blob/6c5a07602aed248acc86598f30ab61923a54a83e/ipld/unixfs/io/directory.go#L26
 
@@ -71,6 +72,7 @@ type Import struct {
 	BatchMaxNodes                     OptionalInteger
 	BatchMaxSize                      OptionalInteger
 	FastProvideRoot                   Flag
+	FastProvideDAG                    Flag
 	FastProvideWait                   Flag
 }
 

config/provide.go

@@ -12,6 +12,22 @@ const (
 	DefaultProvideEnabled  = true
 	DefaultProvideStrategy = "all"
 
+	// DefaultProvideBloomFPRate is the target false positive rate for the
+	// bloom filter used by +unique and +entities reprovide cycles and
+	// fast-provide-dag walks. Expressed as 1/N (one false positive per N
+	// lookups). At ~1 in 4.75M (~0.00002%) each CID costs ~4 bytes before
+	// ipfs/bbloom's power-of-two rounding.
+	//
+	// Kubo owns this default independently of boxo/dag/walker; the two
+	// values may diverge over time without coordination.
+	DefaultProvideBloomFPRate = 4_750_000
+
+	// MinProvideBloomFPRate is the smallest accepted Provide.BloomFPRate.
+	// Below 1 in 1M the bloom filter becomes lossy enough to drop a
+	// meaningful fraction of CIDs from each reprovide cycle (e.g. at
+	// rate=10_000 a 100M-CID repo skips ~10K CIDs per cycle).
+	MinProvideBloomFPRate = 1_000_000
+
 	// DHT provider defaults
 	DefaultProvideDHTInterval                 = 22 * time.Hour // https://github.com/ipfs/kubo/pull/9326
 	DefaultProvideDHTMaxWorkers               = 16             // Unified default for both sweep and legacy providers
@@ -36,6 +52,8 @@ const (
 	ProvideStrategyPinned
 	ProvideStrategyRoots
 	ProvideStrategyMFS
+	ProvideStrategyUnique   // bloom filter cross-DAG deduplication
+	ProvideStrategyEntities // entity-aware traversal (implies Unique)
 )
 
 // Provide configures both immediate CID announcements (provide operations) for new content
@@ -50,6 +68,16 @@ type Provide struct {
 	// Default: DefaultProvideStrategy
 	Strategy *OptionalString `json:",omitempty"`
 
+	// BloomFPRate sets the target false positive rate of the bloom filter
+	// used by Provide.Strategy modifiers +unique and +entities (and the
+	// matching fast-provide-dag walk). Expressed as 1/N (one false
+	// positive per N lookups), so higher N means lower FP rate but more
+	// memory per CID. Only takes effect when Provide.Strategy includes
+	// +unique or +entities.
+	//
+	// Default: DefaultProvideBloomFPRate
+	BloomFPRate *OptionalInteger `json:",omitempty"`
+
 	// DHT configures DHT-specific provide and reprovide settings.
 	DHT ProvideDHT
 }
@@ -100,25 +128,78 @@ type ProvideDHT struct {
 	ResumeEnabled Flag `json:",omitempty"`
 }
 
-func ParseProvideStrategy(s string) ProvideStrategy {
+func ParseProvideStrategy(s string) (ProvideStrategy, error) {
 	var strategy ProvideStrategy
 	for part := range strings.SplitSeq(s, "+") {
 		switch part {
-		case "all", "flat", "": // special case, does not mix with others ("flat" is deprecated, maps to "all")
-			return ProvideStrategyAll
+		case "all", "flat":
+			strategy |= ProvideStrategyAll
+		case "":
+			// empty string (default config) maps to "all",
+			// but empty tokens from splitting (e.g. "pinned+") are invalid
+			if s == "" {
+				strategy |= ProvideStrategyAll
+			} else {
+				return 0, fmt.Errorf("invalid provide strategy: empty token in %q", s)
+			}
 		case "pinned":
 			strategy |= ProvideStrategyPinned
 		case "roots":
 			strategy |= ProvideStrategyRoots
 		case "mfs":
 			strategy |= ProvideStrategyMFS
+		case "unique":
+			strategy |= ProvideStrategyUnique
+		case "entities":
+			strategy |= ProvideStrategyEntities | ProvideStrategyUnique
+		default:
+			return 0, fmt.Errorf("unknown provide strategy token: %q in %q", part, s)
+		}
+	}
+	// "all" provides every block and cannot be combined with selective strategies
+	if strategy&ProvideStrategyAll != 0 && strategy != ProvideStrategyAll {
+		return 0, fmt.Errorf("\"all\" strategy cannot be combined with other strategies in %q", s)
+	}
+	// +unique/+entities require a base strategy that walks DAGs (pinned and/or mfs)
+	wantsDedup := strategy&(ProvideStrategyUnique|ProvideStrategyEntities) != 0
+	if wantsDedup {
+		walksDAGs := strategy&(ProvideStrategyPinned|ProvideStrategyMFS) != 0
+		if !walksDAGs {
+			return 0, fmt.Errorf("+unique/+entities must combine with pinned and/or mfs in %q", s)
+		}
+		if strategy&ProvideStrategyRoots != 0 {
+			return 0, fmt.Errorf("+unique/+entities is incompatible with roots in %q", s)
 		}
 	}
+	return strategy, nil
+}
+
+// MustParseProvideStrategy is like ParseProvideStrategy but panics on error.
+// Use with strategy strings that have already been validated at startup.
+func MustParseProvideStrategy(s string) ProvideStrategy {
+	strategy, err := ParseProvideStrategy(s)
+	if err != nil {
+		panic(err)
+	}
 	return strategy
 }
 
 // ValidateProvideConfig validates the Provide configuration according to DHT requirements.
 func ValidateProvideConfig(cfg *Provide) error {
+	// Validate Provide.Strategy
+	strategy := cfg.Strategy.WithDefault(DefaultProvideStrategy)
+	if _, err := ParseProvideStrategy(strategy); err != nil {
+		return fmt.Errorf("Provide.Strategy: %w", err)
+	}
+
+	// Validate Provide.BloomFPRate
+	if !cfg.BloomFPRate.IsDefault() {
+		rate := cfg.BloomFPRate.WithDefault(DefaultProvideBloomFPRate)
+		if rate < MinProvideBloomFPRate {
+			return fmt.Errorf("Provide.BloomFPRate must be >= %d (1 in 1M), got %d", MinProvideBloomFPRate, rate)
+		}
+	}
+
 	// Validate Provide.DHT.Interval
 	if !cfg.DHT.Interval.IsDefault() {
 		interval := cfg.DHT.Interval.WithDefault(DefaultProvideDHTInterval)
@@ -184,7 +265,7 @@ func ValidateProvideConfig(cfg *Provide) error {
 // ShouldProvideForStrategy determines if content should be provided based on the provide strategy
 // and content characteristics (pinned status, root status, MFS status).
 func ShouldProvideForStrategy(strategy ProvideStrategy, isPinned bool, isPinnedRoot bool, isMFS bool) bool {
-	if strategy == ProvideStrategyAll {
+	if strategy&ProvideStrategyAll != 0 {
 		// 'all' strategy: always provide
 		return true
 	}

config/provide_test.go

@@ -9,27 +9,146 @@ import (
 )
 
 func TestParseProvideStrategy(t *testing.T) {
-	tests := []struct {
-		input  string
-		expect ProvideStrategy
-	}{
-		{"all", ProvideStrategyAll},
-		{"pinned", ProvideStrategyPinned},
-		{"mfs", ProvideStrategyMFS},
-		{"pinned+mfs", ProvideStrategyPinned | ProvideStrategyMFS},
-		{"invalid", 0},
-		{"all+invalid", ProvideStrategyAll},
-		{"", ProvideStrategyAll},
-		{"flat", ProvideStrategyAll}, // deprecated, maps to "all"
-		{"flat+all", ProvideStrategyAll},
-	}
+	t.Run("valid strategies", func(t *testing.T) {
+		tests := []struct {
+			input  string
+			expect ProvideStrategy
+		}{
+			{"all", ProvideStrategyAll},
+			{"pinned", ProvideStrategyPinned},
+			{"roots", ProvideStrategyRoots},
+			{"mfs", ProvideStrategyMFS},
+			{"pinned+mfs", ProvideStrategyPinned | ProvideStrategyMFS},
+			{"pinned+roots", ProvideStrategyPinned | ProvideStrategyRoots},
+			{"pinned+mfs+roots", ProvideStrategyPinned | ProvideStrategyMFS | ProvideStrategyRoots},
+			{"", ProvideStrategyAll},                                   // empty string = default = all
+			{"flat", ProvideStrategyAll},                               // deprecated, maps to "all"
+			{"flat+all", ProvideStrategyAll},                           // redundant but valid
+			{"all+all", ProvideStrategyAll},                            // redundant but valid
+			{"mfs+pinned", ProvideStrategyMFS | ProvideStrategyPinned}, // order doesn't matter
+			// +unique and +entities modifiers
+			{"pinned+unique", ProvideStrategyPinned | ProvideStrategyUnique},
+			{"pinned+entities", ProvideStrategyPinned | ProvideStrategyEntities | ProvideStrategyUnique},
+			{"pinned+unique+entities", ProvideStrategyPinned | ProvideStrategyUnique | ProvideStrategyEntities},
+			{"mfs+unique", ProvideStrategyMFS | ProvideStrategyUnique},
+			{"mfs+entities", ProvideStrategyMFS | ProvideStrategyEntities | ProvideStrategyUnique},
+			{"pinned+mfs+unique", ProvideStrategyPinned | ProvideStrategyMFS | ProvideStrategyUnique},
+			{"pinned+mfs+entities", ProvideStrategyPinned | ProvideStrategyMFS | ProvideStrategyEntities | ProvideStrategyUnique},
+		}
 
-	for _, tt := range tests {
-		result := ParseProvideStrategy(tt.input)
-		if result != tt.expect {
-			t.Errorf("ParseProvideStrategy(%q) = %d, want %d", tt.input, result, tt.expect)
+		for _, tt := range tests {
+			result, err := ParseProvideStrategy(tt.input)
+			require.NoError(t, err, "ParseProvideStrategy(%q)", tt.input)
+			assert.Equal(t, tt.expect, result, "ParseProvideStrategy(%q)", tt.input)
 		}
-	}
+	})
+
+	t.Run("unknown token (including typos)", func(t *testing.T) {
+		tests := []struct {
+			input string
+			err   string
+		}{
+			{"invalid", `unknown provide strategy token: "invalid"`},
+			{"uniuqe", `unknown provide strategy token: "uniuqe"`},        // typo of "unique"
+			{"entites", `unknown provide strategy token: "entites"`},      // cspell:disable-line -- intentional typo of "entities"
+			{"pinned+uniuqe", `unknown provide strategy token: "uniuqe"`}, // typo in combo
+		}
+
+		for _, tt := range tests {
+			_, err := ParseProvideStrategy(tt.input)
+			require.Error(t, err, "ParseProvideStrategy(%q) should fail", tt.input)
+			assert.Contains(t, err.Error(), tt.err)
+		}
+	})
+
+	t.Run("empty token from delimiter", func(t *testing.T) {
+		tests := []string{
+			"pinned+",     // trailing +
+			"+pinned",     // leading +
+			"pinned++mfs", // double +
+		}
+
+		for _, input := range tests {
+			_, err := ParseProvideStrategy(input)
+			require.Error(t, err, "ParseProvideStrategy(%q) should fail", input)
+			assert.Contains(t, err.Error(), "empty token")
+		}
+	})
+
+	t.Run("all cannot be combined with other strategies", func(t *testing.T) {
+		tests := []string{
+			"all+pinned",
+			"all+mfs",
+			"all+roots",
+			"flat+pinned",
+			"all+pinned+mfs",
+		}
+
+		for _, input := range tests {
+			_, err := ParseProvideStrategy(input)
+			require.Error(t, err, "ParseProvideStrategy(%q) should fail", input)
+			assert.Contains(t, err.Error(), "cannot be combined")
+		}
+	})
+
+	t.Run("+unique/+entities require base strategy", func(t *testing.T) {
+		tests := []string{
+			"unique",              // modifier alone
+			"entities",            // modifier alone
+			"unique+entities",     // modifiers without base
+			"roots+unique",        // roots is incompatible
+			"roots+entities",      // roots is incompatible
+			"roots+pinned+unique", // roots mixed with pinned+unique
+		}
+
+		for _, input := range tests {
+			_, err := ParseProvideStrategy(input)
+			require.Error(t, err, "ParseProvideStrategy(%q) should fail", input)
+		}
+	})
+}
+
+func TestMustParseProvideStrategy(t *testing.T) {
+	t.Run("valid input returns strategy", func(t *testing.T) {
+		assert.Equal(t, ProvideStrategyAll, MustParseProvideStrategy("all"))
+		assert.Equal(t, ProvideStrategyPinned|ProvideStrategyMFS, MustParseProvideStrategy("pinned+mfs"))
+	})
+
+	t.Run("invalid input panics", func(t *testing.T) {
+		assert.Panics(t, func() { MustParseProvideStrategy("bogus") })
+		assert.Panics(t, func() { MustParseProvideStrategy("all+pinned") })
+	})
+}
+
+func TestValidateProvideConfig_Strategy(t *testing.T) {
+	t.Run("valid strategies", func(t *testing.T) {
+		for _, s := range []string{
+			"all", "pinned", "roots", "mfs", "pinned+mfs",
+			"pinned+unique", "pinned+entities", "pinned+mfs+entities",
+		} {
+			cfg := &Provide{Strategy: NewOptionalString(s)}
+			require.NoError(t, ValidateProvideConfig(cfg), "strategy=%q", s)
+		}
+	})
+
+	t.Run("default (nil) strategy is valid", func(t *testing.T) {
+		cfg := &Provide{}
+		require.NoError(t, ValidateProvideConfig(cfg))
+	})
+
+	t.Run("invalid strategy", func(t *testing.T) {
+		cfg := &Provide{Strategy: NewOptionalString("bogus")}
+		err := ValidateProvideConfig(cfg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "Provide.Strategy")
+	})
+
+	t.Run("all combined with others", func(t *testing.T) {
+		cfg := &Provide{Strategy: NewOptionalString("all+pinned")}
+		err := ValidateProvideConfig(cfg)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "cannot be combined")
+	})
 }
 
 func TestValidateProvideConfig_Interval(t *testing.T) {
@@ -70,6 +189,49 @@ func TestValidateProvideConfig_Interval(t *testing.T) {
 	}
 }
 
+func TestValidateProvideConfig_BloomFPRate(t *testing.T) {
+	tests := []struct {
+		name    string
+		fpRate  int64
+		wantErr bool
+		errMsg  string
+	}{
+		{"valid default value", DefaultProvideBloomFPRate, false, ""},
+		{"valid minimum (1M)", MinProvideBloomFPRate, false, ""},
+		{"valid high (10M)", 10_000_000, false, ""},
+		{"valid very high (100M)", 100_000_000, false, ""},
+		{"invalid below minimum (999_999)", 999_999, true, "must be >="},
+		{"invalid small (10_000)", 10_000, true, "must be >="},
+		{"invalid one", 1, true, "must be >="},
+		{"invalid zero", 0, true, "must be >="},
+		{"invalid negative", -1, true, "must be >="},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := &Provide{
+				BloomFPRate: NewOptionalInteger(tt.fpRate),
+			}
+
+			err := ValidateProvideConfig(cfg)
+
+			if tt.wantErr {
+				require.Error(t, err, "expected error for fpRate=%d", tt.fpRate)
+				if tt.errMsg != "" {
+					assert.Contains(t, err.Error(), tt.errMsg, "error message mismatch")
+				}
+			} else {
+				require.NoError(t, err, "unexpected error for fpRate=%d", tt.fpRate)
+			}
+		})
+	}
+
+	t.Run("default (nil) BloomFPRate is valid", func(t *testing.T) {
+		cfg := &Provide{}
+		require.NoError(t, ValidateProvideConfig(cfg))
+	})
+}
+
 func TestValidateProvideConfig_MaxWorkers(t *testing.T) {
 	tests := []struct {
 		name       string