feat: introduce trickle sp tier

Author: silent-cipher

apps/backend/.env.example

@@ -22,10 +22,23 @@
 # Chain-specific variables never inherit the unprefixed slot and MUST be
 # prefixed: WALLET_ADDRESS, WALLET_PRIVATE_KEY, SESSION_KEY_PRIVATE_KEY, RPC_URL,
 # CLICKHOUSE_URL, PDP_SUBGRAPH_ENDPOINT, SUBGRAPH_ENDPOINT, DEALBOT_DATASET_VERSION,
-# BLOCKED_SP_IDS, BLOCKED_SP_ADDRESSES, DATASET_LIFECYCLE_CHECK_ENABLED
+# BLOCKED_SP_IDS, BLOCKED_SP_ADDRESSES, EXPECTED_APPROVED_SP_IDS,
+# EXPECTED_APPROVED_SP_ADDRESSES, DATASET_LIFECYCLE_CHECK_ENABLED
 # (network-dependent default: off on mainnet, so a shared =true must not enable
 # the canary there).
 #
+# SPs get full-rate testing (DEALS_PER_SP_PER_HOUR, DATASET_CREATIONS_PER_SP_PER_HOUR,
+# MIN_NUM_DATASETS_FOR_CHECKS, and the DATASET_LIFECYCLE_CHECK_ENABLED canary) only
+# if they're on-chain approved OR listed in
+# EXPECTED_APPROVED_SP_IDS/EXPECTED_APPROVED_SP_ADDRESSES (either one is enough —
+# they're checked independently, same as BLOCKED_SP_IDS/BLOCKED_SP_ADDRESSES).
+# Every other active, unblocked SP (new/unknown/not-yet-vetted) is throttled to a
+# fixed trickle tier (1 deal + 1 data-set-creation attempt every 4 hours, 1
+# data-set target; see `trickleTierRates` in src/config/constants.ts) to bound
+# wallet spend on SPs dealbot hasn't vetted yet. The lifecycle-check canary is
+# never scheduled at all for trickle-tier SPs (it creates+terminates a real
+# throwaway dataset every run). See #681.
+#
 # Process globals (database, HTTP ports, ClickHouse batching, pg-boss) have no
 # per-network form. Variables for INACTIVE networks are ignored; only active
 # networks are validated at startup.
@@ -140,6 +153,11 @@ CALIBRATION_PULL_PIECE_CLEANUP_INTERVAL_SECONDS=604800
 # CALIBRATION_BLOCKED_SP_IDS=1234,5678
 # CALIBRATION_BLOCKED_SP_ADDRESSES=0xAbCd...,0x1234...
 
+# SPs eligible for full-rate testing beyond those already isApproved on-chain
+# (see #681) — candidates being considered for approval.
+# CALIBRATION_EXPECTED_APPROVED_SP_IDS=1234,5678
+# CALIBRATION_EXPECTED_APPROVED_SP_ADDRESSES=0xAbCd...,0x1234...
+
 # -----------------------------------------------------------------------------
 # Per-network configuration — MAINNET  (uncomment when adding to NETWORKS)
 # -----------------------------------------------------------------------------
@@ -174,6 +192,12 @@ MAINNET_MAINTENANCE_WINDOWS_UTC=07:00,22:00
 MAINNET_MAINTENANCE_WINDOW_MINUTES=20
 # MAINNET_BLOCKED_SP_IDS=1234,5678
 # MAINNET_BLOCKED_SP_ADDRESSES=0xAbCd...,0x1234...
+
+# SPs eligible for full-rate testing beyond those already isApproved on-chain
+# (see #681) — candidates being considered for approval.
+# MAINNET_EXPECTED_APPROVED_SP_IDS=1234,5678
+# MAINNET_EXPECTED_APPROVED_SP_ADDRESSES=0xAbCd...,0x1234...
+
 MAINNET_MAX_DATASET_STORAGE_SIZE_BYTES=25769803776
 MAINNET_TARGET_DATASET_STORAGE_SIZE_BYTES=21474836480
 MAINNET_MAX_PIECE_CLEANUP_RUNTIME_SECONDS=3000

apps/backend/src/common/sp-tier.ts

@@ -0,0 +1,20 @@
+import { INetworkConfig } from "src/config/types.js";
+
+/**
+ * Returns true if the provider qualifies for the full-rate testing tier:
+ * already `isApproved` on-chain (approved SPs must stay fully monitored
+ * regardless of manual-list staleness), or manually curated as an
+ * expected-approval candidate. Every other SP (new, unknown, or not yet
+ * vetted) defaults to the trickle tier — see #681.
+ */
+export function isFullRateTier(
+  cfg: Pick<INetworkConfig, "expectedApprovedSpAddresses" | "expectedApprovedSpIds">,
+  address: string,
+  isApproved: boolean,
+  id?: bigint | null,
+): boolean {
+  if (isApproved) return true;
+  if (cfg.expectedApprovedSpAddresses.has(address.toLowerCase())) return true;
+  if (id != null && cfg.expectedApprovedSpIds.has(String(id))) return true;
+  return false;
+}

apps/backend/src/config/constants.ts

@@ -36,6 +36,20 @@ export const networkDefaults = {
   pullPieceCleanupIntervalSeconds: 7 * 24 * 3600, // 7 days
 } satisfies NetworkDefaults;
 
+/**
+ * Rate/target ceiling applied to SPs that are not on the full-rate tier (see
+ * `isFullRateTier` in `common/sp-tier.ts`) — i.e. not `isApproved` on-chain
+ * and not in `EXPECTED_APPROVED_SP_IDS`/`EXPECTED_APPROVED_SP_ADDRESSES`.
+ *
+ * Deliberately NOT env-configurable per network: this tier exists to bound
+ * worst-case wallet spend on new/unknown SPs (#681).
+ */
+export const trickleTierRates = {
+  dealsPerSpPerHour: 1 / 4, // 1 data-storage check every 4 hours
+  dataSetCreationsPerSpPerHour: 1 / 4, // 1 data-set-creation tick every 4 hours
+  minNumDataSetsForChecks: 1,
+} as const;
+
 /**
  * Uppercase env-var prefixes for every supported network, e.g.
  * `["CALIBRATION", "MAINNET"]`

apps/backend/src/config/env.schema.ts

@@ -189,6 +189,8 @@ const perNetworkFieldRules = (): Record<PerNetworkVar, Joi.Schema> => ({
   MAINTENANCE_WINDOW_MINUTES: Joi.number().min(20).max(360).optional(),
   BLOCKED_SP_IDS: Joi.string().optional().allow(""),
   BLOCKED_SP_ADDRESSES: Joi.string().optional().allow(""),
+  EXPECTED_APPROVED_SP_IDS: Joi.string().optional().allow(""),
+  EXPECTED_APPROVED_SP_ADDRESSES: Joi.string().optional().allow(""),
   PIECE_CLEANUP_PER_SP_PER_HOUR: Joi.number().min(0.001).max(20).optional(),
   MAX_PIECE_CLEANUP_RUNTIME_SECONDS: Joi.number().min(60).optional(),
   MAX_DATASET_STORAGE_SIZE_BYTES: Joi.number().integer().min(1).optional(),

apps/backend/src/config/loader.ts

@@ -234,6 +234,9 @@ function loadNetworkEnvPrefix(
     blockedSpIds: parseIdList(resolve("BLOCKED_SP_IDS")),
     blockedSpAddresses: parseAddressList(resolve("BLOCKED_SP_ADDRESSES")),
 
+    expectedApprovedSpIds: parseIdList(resolve("EXPECTED_APPROVED_SP_IDS")),
+    expectedApprovedSpAddresses: parseAddressList(resolve("EXPECTED_APPROVED_SP_ADDRESSES")),
+
     pullChecksPerSpPerHour: coerceFloat(resolve("PULL_CHECKS_PER_SP_PER_HOUR"), networkDefaults.pullChecksPerSpPerHour),
     pullCheckJobTimeoutSeconds: coerceNumber(
       resolve("PULL_CHECK_JOB_TIMEOUT_SECONDS"),

apps/backend/src/config/network-fields.ts

@@ -47,6 +47,8 @@ export const PER_NETWORK_VARS = [
   "MAINTENANCE_WINDOW_MINUTES",
   "BLOCKED_SP_IDS",
   "BLOCKED_SP_ADDRESSES",
+  "EXPECTED_APPROVED_SP_IDS",
+  "EXPECTED_APPROVED_SP_ADDRESSES",
   "PIECE_CLEANUP_PER_SP_PER_HOUR",
   "MAX_PIECE_CLEANUP_RUNTIME_SECONDS",
   "MAX_DATASET_STORAGE_SIZE_BYTES",
@@ -76,6 +78,8 @@ export const CHAIN_SPECIFIC_NETWORK_VARS = new Set<PerNetworkVar>([
   "DEALBOT_DATASET_VERSION",
   "BLOCKED_SP_IDS",
   "BLOCKED_SP_ADDRESSES",
+  "EXPECTED_APPROVED_SP_IDS",
+  "EXPECTED_APPROVED_SP_ADDRESSES",
   // Kept chain-specific so a shared `=true` can't accidentally enable it on mainnet.
   "DATASET_LIFECYCLE_CHECK_ENABLED",
 ]);

apps/backend/src/config/types.ts

@@ -141,6 +141,17 @@ export type BaseNetworkConfig = {
   blockedSpIds: Set<string>;
   blockedSpAddresses: Set<string>;
 
+  /**
+   * Manually curated SPs that should receive full-rate testing even though
+   * they aren't (yet) `isApproved` on-chain — e.g. candidates being
+   * considered for approval. Combined with `isApproved` to gate the
+   * full-rate tier (see `isFullRateTier` in `common/sp-tier.ts`); every
+   * other active, unblocked SP defaults to the trickle tier
+   * (`trickleTierRates` in `config/constants.ts`). See #681.
+   */
+  expectedApprovedSpIds: Set<string>;
+  expectedApprovedSpAddresses: Set<string>;
+
   /** Piece Cleanup Config */
   maxDatasetStorageSizeBytes: number;
   targetDatasetStorageSizeBytes: number;

apps/backend/src/deal/deal.service.spec.ts

@@ -1365,7 +1365,7 @@ describe("DealService", () => {
     it("returns undefined when minDataSets=1 and baseline is live", async () => {
       probeSpy.mockResolvedValueOnce({ status: "live", dataSetId: 1n });
 
-      const result = await service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK);
+      const result = await service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK, 1);
       expect(result).toBeUndefined();
       expect(probeSpy).toHaveBeenCalledTimes(1);
       const [, metadata] = probeSpy.mock.calls[0] ?? [];
@@ -1375,7 +1375,7 @@ describe("DealService", () => {
     it("throws DealJobTerminatedDataSetError when baseline is terminated", async () => {
       probeSpy.mockResolvedValueOnce({ status: "terminated", dataSetId: 42n });
 
-      await expect(service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK)).rejects.toBeInstanceOf(
+      await expect(service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK, 1)).rejects.toBeInstanceOf(
         DealJobTerminatedDataSetError,
       );
     });
@@ -1390,7 +1390,7 @@ describe("DealService", () => {
       vi.spyOn(Math, "random").mockReturnValue(0.5); // → index 1
       probeSpy.mockResolvedValueOnce({ status: "live", dataSetId: 7n });
 
-      const result = await service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK);
+      const result = await service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK, 3);
       expect(result).toEqual({ dealbotDS: "1" });
       expect(probeSpy).toHaveBeenCalledTimes(1);
       const [, metadata] = probeSpy.mock.calls[0] ?? [];
@@ -1412,7 +1412,7 @@ describe("DealService", () => {
       setupIndexedProbe();
       probeSpy.mockResolvedValueOnce({ status: "live", dataSetId: 1n });
 
-      const result = await service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK);
+      const result = await service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK, 3);
       expect(result).toBeUndefined();
     });
 
@@ -1431,7 +1431,7 @@ describe("DealService", () => {
       });
 
       await expect(
-        service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK, controller.signal),
+        service.resolveDataSetMetadataForDeal("0xprovider", DEFAULT_NETWORK, 3, controller.signal),
       ).rejects.toThrow();
       expect(probeSpy).toHaveBeenCalledTimes(1);
     });

apps/backend/src/deal/deal.service.ts

@@ -17,8 +17,10 @@ import {
   redactSensitiveText,
   toStructuredError,
 } from "../common/logging.js";
+import { isFullRateTier } from "../common/sp-tier.js";
 import { createSynapseFromConfig } from "../common/synapse-factory.js";
 import type { DataFile, Hex, Network } from "../common/types.js";
+import { trickleTierRates } from "../config/constants.js";
 import type { IConfig, INetworkConfig } from "../config/types.js";
 import { Deal } from "../database/entities/deal.entity.js";
 import { DealStatus, IpniStatus, ServiceType } from "../database/types.js";
@@ -84,9 +86,14 @@ export class DealService {
   ): Promise<Deal> {
     options.signal?.throwIfAborted();
 
+    const networkCfg = this.getNetworkConfig(options.network);
+    const isFullTier = isFullRateTier(networkCfg, pdpProvider.serviceProvider, pdpProvider.isApproved, pdpProvider.id);
+    const minDataSets = isFullTier ? networkCfg.minNumDataSetsForChecks : trickleTierRates.minNumDataSetsForChecks;
+
     const extraDataSetMetadata = await this.resolveDataSetMetadataForDeal(
       pdpProvider.serviceProvider,
       options.network,
+      minDataSets,
       options.signal,
       options.logContext,
     );
@@ -117,9 +124,10 @@ export class DealService {
    * Pick which data-set slot this deal will target.
    *
    * Policy:
-   *   - If `minNumDataSetsForChecks > 1` and a random index > 0 is selected,
-   *     probe that slot first. If live, use it. If missing or terminated,
-   *     fall through to baseline (data_set_creation owns repair/provisioning).
+   *   - If `minDataSets > 1` (the caller's per-SP tier target — see
+   *     `isFullRateTier`) and a random index > 0 is selected, probe that
+   *     slot first. If live, use it. If missing or terminated, fall through
+   *     to baseline (data_set_creation owns repair/provisioning).
    *   - Probe baseline. If terminated, throw `DealJobTerminatedDataSetError`
    *     (baseline is the fallback target; nothing else to try).
    *   - Live or missing baseline → return `undefined` (use baseline slot).
@@ -131,6 +139,7 @@ export class DealService {
   async resolveDataSetMetadataForDeal(
     providerAddress: string,
     network: Network,
+    minDataSets: number,
     signal?: AbortSignal,
     logContext?: ProviderJobContext,
   ): Promise<Record<string, string> | undefined> {
@@ -141,6 +150,7 @@ export class DealService {
       providerAddress,
       baseDataSetMetadata,
       network,
+      minDataSets,
       signal,
       logContext,
     );
@@ -174,10 +184,10 @@ export class DealService {
     providerAddress: string,
     baseDataSetMetadata: Record<string, string>,
     network: Network,
+    minDataSets: number,
     signal: AbortSignal | undefined,
     logContext: ProviderJobContext | undefined,
   ): Promise<Record<string, string> | undefined> {
-    const minDataSets = this.getNetworkConfig(network).minNumDataSetsForChecks;
     if (minDataSets <= 1) return undefined;
     const dsIndex = Math.floor(Math.random() * minDataSets);
     if (dsIndex === 0) return undefined;

apps/backend/src/jobs/jobs.service.spec.ts

@@ -24,6 +24,7 @@ describe("JobsService schedule rows", () => {
     upsertSchedule: ReturnType<typeof vi.fn>;
     deleteSchedulesForInactiveProviders: ReturnType<typeof vi.fn>;
     deleteSchedulesByJobType: ReturnType<typeof vi.fn>;
+    deleteSchedulesForAddresses: ReturnType<typeof vi.fn>;
     countPausedSchedules: ReturnType<typeof vi.fn>;
     findDueSchedulesWithManager: ReturnType<typeof vi.fn>;
     runTransaction: ReturnType<typeof vi.fn>;
@@ -90,6 +91,7 @@ describe("JobsService schedule rows", () => {
       upsertSchedule: vi.fn(),
       deleteSchedulesForInactiveProviders: vi.fn(async () => []),
       deleteSchedulesByJobType: vi.fn(async () => 0),
+      deleteSchedulesForAddresses: vi.fn(async () => []),
       countPausedSchedules: vi.fn(async () => []),
       findDueSchedulesWithManager: vi.fn(),
       runTransaction: vi.fn(async (callback: (manager: unknown) => Promise<void>) => {
@@ -151,6 +153,8 @@ describe("JobsService schedule rows", () => {
       maintenanceWindowMinutes: 20,
       blockedSpIds: new Set(),
       blockedSpAddresses: new Set(),
+      expectedApprovedSpIds: new Set(),
+      expectedApprovedSpAddresses: new Set(),
       pieceCleanupPerSpPerHour: 1,
       maxPieceCleanupRuntimeSeconds: 300,
       maxDatasetStorageSizeBytes: 24 * 1024 * 1024 * 1024,
@@ -1337,7 +1341,7 @@ describe("JobsService schedule rows", () => {
     };
 
     const providerRegistryRepository = {
-      findByAddress: vi.fn(() => ({ id: 1, name: "test-provider" })),
+      findByAddress: vi.fn(() => ({ id: 1, name: "test-provider", isApproved: true })),
     };
 
     service = buildService({
@@ -1474,7 +1478,9 @@ describe("JobsService schedule rows", () => {
     } as unknown as JobsServiceDeps[0];
 
     const dataSetLifecycleService = { runLifecycleCheck: vi.fn(async () => undefined) };
-    const providerRegistryRepository = { findByAddress: vi.fn(() => ({ id: 1, name: "test-provider" })) };
+    const providerRegistryRepository = {
+      findByAddress: vi.fn(() => ({ id: 1, name: "test-provider", isApproved: true })),
+    };
 
     service = buildService({
       configService,
@@ -1519,7 +1525,7 @@ describe("JobsService schedule rows", () => {
     } as unknown as JobsServiceDeps[0];
     service = buildService({ configService });
 
-    providerRegistryRepositoryMock.findActiveAddresses.mockResolvedValueOnce([{ address: "0xaaa" }]);
+    providerRegistryRepositoryMock.findActiveAddresses.mockResolvedValueOnce([{ address: "0xaaa", isApproved: true }]);
 
     await callPrivate(service, "ensureScheduleRows", DEFAULT_NETWORK);
 
@@ -1529,11 +1535,12 @@ describe("JobsService schedule rows", () => {
     expect(lifecycleUpserts).toHaveLength(1);
     expect(lifecycleUpserts[0][1]).toBe("0xaaa");
     expect(jobScheduleRepositoryMock.deleteSchedulesByJobType).not.toHaveBeenCalled();
+    expect(jobScheduleRepositoryMock.deleteSchedulesForAddresses).not.toHaveBeenCalled();
   });
 
   it("removes data_set_lifecycle_check schedules when disabled", async () => {
     // base config has dataSetLifecycleCheckEnabled=false
-    providerRegistryRepositoryMock.findActiveAddresses.mockResolvedValueOnce([{ address: "0xaaa" }]);
+    providerRegistryRepositoryMock.findActiveAddresses.mockResolvedValueOnce([{ address: "0xaaa", isApproved: true }]);
 
     await callPrivate(service, "ensureScheduleRows", DEFAULT_NETWORK);
 
@@ -1547,6 +1554,35 @@ describe("JobsService schedule rows", () => {
     );
   });
 
+  it("does not schedule data_set_lifecycle_check for trickle-tier providers, and clears stale rows", async () => {
+    baseConfigValues = {
+      ...baseConfigValues,
+      networks: {
+        calibration: { ...(baseConfigValues.networks as any).calibration, dataSetLifecycleCheckEnabled: true },
+      } as unknown as IConfig["networks"],
+    };
+    configService = {
+      get: vi.fn((key: keyof IConfig) => baseConfigValues[key]),
+    } as unknown as JobsServiceDeps[0];
+    service = buildService({ configService });
+
+    // Neither approved nor on the expected-approved list -> trickle tier.
+    providerRegistryRepositoryMock.findActiveAddresses.mockResolvedValueOnce([{ address: "0xaaa", isApproved: false }]);
+
+    await callPrivate(service, "ensureScheduleRows", DEFAULT_NETWORK);
+
+    const lifecycleUpserts = jobScheduleRepositoryMock.upsertSchedule.mock.calls.filter(
+      (call) => call[0] === "data_set_lifecycle_check",
+    );
+    expect(lifecycleUpserts).toHaveLength(0);
+    expect(jobScheduleRepositoryMock.deleteSchedulesByJobType).not.toHaveBeenCalled();
+    expect(jobScheduleRepositoryMock.deleteSchedulesForAddresses).toHaveBeenCalledWith(
+      "data_set_lifecycle_check",
+      ["0xaaa"],
+      DEFAULT_NETWORK,
+    );
+  });
+
   it("sets active, inactive, and tested provider gauge values after refresh", async () => {
     providerRegistryRepositoryMock.countByNetwork.mockResolvedValueOnce(10);
     providerRegistryRepositoryMock.countActiveByNetwork.mockResolvedValueOnce(7);