feat: validate release cause

cmd/cli/setup/register.go

@@ -832,6 +832,8 @@ func registerWithDelegator(ctx context.Context, cmd *cobra.Command, cfg *appcfg.
 		cmds := []ucan.Command{
 			blob.Allocate.Command,
 			blob.Accept.Command,
+			blob.Release.Command,
+			blob.Reject.Command,
 			pdp.Info.Command,
 			replicacmds.Allocate.Command,
 		}

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/fil-forge/delegator v0.0.0-20260619085531-c96197bec34c
 	github.com/fil-forge/filecoin-services/go v0.0.0-20260507172456-36ebe4467390
 	github.com/fil-forge/go-ipni-tools v0.0.0-20260519194815-545b9421aec0
-	github.com/fil-forge/libforge v0.0.0-20260630210927-2b55dbcf944f
+	github.com/fil-forge/libforge v0.0.0-20260730113809-b13386bc395a
 	github.com/fil-forge/piri-signing-service v0.0.0-20260619151844-6c3d40a4bf54
 	github.com/fil-forge/ucantone v0.0.0-20260727203046-ccb77059de44
 	github.com/filecoin-project/curio v1.28.3-0.20260717015646-38ca280c43a2
@@ -50,7 +50,6 @@ require (
 	github.com/multiformats/go-multihash v0.2.3
 	github.com/ncruces/go-sqlite3 v0.24.1
 	github.com/pelletier/go-toml/v2 v2.2.4
-	github.com/raulk/clock v1.1.0
 	github.com/samber/lo v1.53.0
 	github.com/schollz/progressbar/v3 v3.19.0
 	github.com/shirou/gopsutil/v4 v4.25.6
@@ -266,6 +265,7 @@ require (
 	github.com/prometheus/procfs v0.20.1 // indirect
 	github.com/prometheus/statsd_exporter v0.29.0 // indirect
 	github.com/puzpuzpuz/xsync/v2 v2.5.1 // indirect
+	github.com/raulk/clock v1.1.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect

go.sum

@@ -551,8 +551,8 @@ github.com/fil-forge/go-ipni-tools v0.0.0-20260519194815-545b9421aec0 h1:HAfXUPv
 github.com/fil-forge/go-ipni-tools v0.0.0-20260519194815-545b9421aec0/go.mod h1:3NRV/7wc4/0uzzrGdI7NoN/yeF1UvqKRwMyjBqGc5s0=
 github.com/fil-forge/go-ucanto v0.0.0-20260507172450-5cb5d073f8ab h1:2J2cDThqTKP6/0k3SfdlSxfyPa3aLqjTYnmvbEcryfg=
 github.com/fil-forge/go-ucanto v0.0.0-20260507172450-5cb5d073f8ab/go.mod h1:lZF3UXZ2hGLKYmXdquG50JqI9pRlUrV6lubGtgOYfwc=
-github.com/fil-forge/libforge v0.0.0-20260630210927-2b55dbcf944f h1:R73fE6WIAJF0tA7YGxJx7rxslPbrGxGnqkh8HOlWyc0=
-github.com/fil-forge/libforge v0.0.0-20260630210927-2b55dbcf944f/go.mod h1:0kXihIQ4L2uZ00nR5XrZ/Y8Db7Ht/qQNuiWslwMJ95M=
+github.com/fil-forge/libforge v0.0.0-20260730113809-b13386bc395a h1:voLQ76lKOQNRg7F+Aah2y0rN67jpDjFsiFMSpxE1Yf4=
+github.com/fil-forge/libforge v0.0.0-20260730113809-b13386bc395a/go.mod h1:0kXihIQ4L2uZ00nR5XrZ/Y8Db7Ht/qQNuiWslwMJ95M=
 github.com/fil-forge/piri-signing-service v0.0.0-20260619151844-6c3d40a4bf54 h1:KzQzCcEBBF6AfhZOwbQIPh5W2HwCgLS8VFTmch0oMgo=
 github.com/fil-forge/piri-signing-service v0.0.0-20260619151844-6c3d40a4bf54/go.mod h1:O3GqUy+G7hb+oiGW6Wp77W2lRmZs7bewM9s/Vy4D6qo=
 github.com/fil-forge/ucantone v0.0.0-20260727203046-ccb77059de44 h1:ofvb2Qq7++VPRelGsLbtnd1ZMKVT4n4QGoa79BhJ6VQ=

pkg/curiopdp/db.go

@@ -1,52 +1,21 @@
 package curiopdp
 
 import (
-	"embed"
 	"fmt"
-	"strings"
 
 	"github.com/filecoin-project/curio/harmony/harmonydb"
 
 	"github.com/fil-forge/piri/pkg/config/app"
+	"github.com/fil-forge/piri/pkg/curiopdp/schema"
 )
 
-// curatedSchemaFS holds ONLY the consolidated PDP closure schema (pdp_* + the
-// harmonytask / eth-message / parked-piece closure), not Curio's full migration
-// set. harmonyquery reads migrations from the "sql/" directory of this FS.
-//
-//go:embed sql
-var curatedSchemaFS embed.FS
-
-// HarmonyDBSchema is the Postgres schema the PDP pipeline tables live in (kept
-// separate from Piri's own "scheduler"/gorm schema).
-const HarmonyDBSchema = "curio"
-
 // ProvideHarmonyDB builds a harmonydb.DB (harmonyquery on Postgres) that applies
-// only the curated PDP closure schema. Curio's pdpv0 tasks + harmonytask run on
-// this DB. Requires Postgres — harmonytask's claim uses FOR UPDATE SKIP LOCKED.
+// only the curated PDP closure schema (see pkg/curiopdp/schema). Curio's pdpv0
+// tasks + harmonytask run on this DB. Requires Postgres — harmonytask's claim
+// uses FOR UPDATE SKIP LOCKED.
 func ProvideHarmonyDB(cfg app.StorageConfig) (*harmonydb.DB, error) {
 	if !cfg.Database.IsPostgres() {
 		return nil, fmt.Errorf("curio PDP pipeline requires Postgres (set database type to postgres)")
 	}
-	u := cfg.Database.Postgres.URL
-	port := u.Port()
-	if port == "" {
-		port = "5432"
-	}
-	pass, _ := u.User.Password()
-	sslmode := u.Query().Get("sslmode")
-	if sslmode == "" {
-		sslmode = "disable"
-	}
-	embedFS := curatedSchemaFS
-	return harmonydb.NewFromConfig(harmonydb.Config{
-		Hosts:      []string{u.Hostname()},
-		Port:       port,
-		Username:   u.User.Username(),
-		Password:   pass,
-		Database:   strings.TrimPrefix(u.Path, "/"),
-		SSLMode:    sslmode,
-		Schema:     HarmonyDBSchema,
-		SqlEmbedFS: &embedFS,
-	})
+	return schema.NewDB(cfg.Database.Postgres.URL)
 }

pkg/curiopdp/schema/schema.go

@@ -0,0 +1,51 @@
+// Package schema holds Piri's curated PDP closure schema (pdp_* plus the
+// harmonytask / eth-message / parked-piece closure — NOT Curio's full
+// migration set) and builds harmonydb handles that apply it. It is a leaf
+// package so tests anywhere in the tree (including pkg/pdp/service, which
+// pkg/curiopdp imports) can spin a schema'd harmonydb without an import
+// cycle.
+package schema
+
+import (
+	"embed"
+	"net/url"
+	"strings"
+
+	"github.com/filecoin-project/curio/harmony/harmonydb"
+)
+
+// FS holds the curated schema; harmonyquery reads migrations from its "sql/"
+// directory.
+//
+//go:embed sql
+var FS embed.FS
+
+// HarmonyDBSchema is the Postgres schema the PDP pipeline tables live in
+// (kept separate from Piri's own "scheduler"/gorm schema).
+const HarmonyDBSchema = "curio"
+
+// NewDB builds a harmonydb.DB at the given Postgres URL with the curated
+// schema applied. Requires Postgres — harmonytask's claim uses FOR UPDATE
+// SKIP LOCKED.
+func NewDB(u url.URL) (*harmonydb.DB, error) {
+	port := u.Port()
+	if port == "" {
+		port = "5432"
+	}
+	pass, _ := u.User.Password()
+	sslmode := u.Query().Get("sslmode")
+	if sslmode == "" {
+		sslmode = "disable"
+	}
+	embedFS := FS
+	return harmonydb.NewFromConfig(harmonydb.Config{
+		Hosts:      []string{u.Hostname()},
+		Port:       port,
+		Username:   u.User.Username(),
+		Password:   pass,
+		Database:   strings.TrimPrefix(u.Path, "/"),
+		SSLMode:    sslmode,
+		Schema:     HarmonyDBSchema,
+		SqlEmbedFS: &embedFS,
+	})
+}

pkg/curiopdp/schema/sql/00000002-pending-piece-removals.sql

@@ -0,0 +1,12 @@
+-- NB(piri): blob byte-release requests, written by /blob/remove and
+-- /blob/reject when the last space claim on a digest is gone. RemovePiece
+-- never deletes inline — the removal machinery advances these rows
+-- asynchronously, re-verifying claims (acceptance/allocation stores) and
+-- pipeline state before releasing bytes and, for proven pieces, scheduling
+-- on-chain deletion (schedulePieceDeletions → rm_message_hash on
+-- pdp_data_set_pieces, mirroring Curio's handleDeleteDataSetPiece).
+CREATE TABLE pdp_pending_piece_removals (
+    digest bytea NOT NULL, -- blob multihash, raw bytes
+    requested_at timestamptz NOT NULL DEFAULT now(),
+    CONSTRAINT pdp_pending_piece_removals_pk PRIMARY KEY (digest)
+);

pkg/curiopdp/schema/sql/00000003-blob-pipeline.sql

@@ -0,0 +1,41 @@
+-- NB(piri): harmonytask-native aggregation pipeline state (replaces the
+-- commp/aggregator/manager jobqueues that ran on the separate piri-local
+-- aggregator_db). One pdp_blob_pipeline row per accepted blob, from
+-- /blob/accept until its aggregate's pieces are staged on-chain; the removal
+-- machinery reads it transactionally to answer "is this digest still in
+-- flight?".
+--
+-- Column typing follows the curio tables this pipeline joins against
+-- (pdp_piece_mh_to_commp: mhash bytea, commp text): multihash digests are
+-- raw bytes, piece CIDs are text strings.
+
+CREATE TABLE pdp_blob_pipeline (
+    digest bytea NOT NULL, -- blob multihash, raw bytes
+    -- stage 1: commp. The task id is provenance for the aggregation task's
+    -- Follows crash net and is never cleared; commp is the v2 piece CID
+    -- string, set when the stage completes.
+    commp_task_id bigint,
+    commp text,
+    commp_hashed_at timestamptz, -- stamped with commp; per-stage age for stuck-row triage
+    -- stage 2: aggregation. agg_task_id records the per-piece aggregation
+    -- task spawned for this blob (spawn dedup); the fold itself operates on
+    -- every unaggregated row under the task's Max=1 serialization.
+    agg_task_id bigint,
+    aggregate_root text, -- aggregate root piece CID (v2), set when folded in
+    aggregated_at timestamptz, -- stamped with aggregate_root
+    created_at timestamptz NOT NULL DEFAULT now(),
+    CONSTRAINT pdp_blob_pipeline_pk PRIMARY KEY (digest)
+);
+CREATE INDEX pdp_blob_pipeline_unaggregated_idx
+    ON pdp_blob_pipeline (created_at) WHERE aggregate_root IS NULL;
+
+-- stage 3: root submission buffer (the manager). One row per aggregate
+-- root awaiting an addPieces submission; rows are claimed by a PDPAddRoots
+-- batch task and deleted (with their pipeline rows) once the batch is
+-- submitted on-chain.
+CREATE TABLE pdp_root_submissions (
+    root text NOT NULL, -- aggregate root piece CID (v2)
+    add_task_id bigint,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    CONSTRAINT pdp_root_submissions_pk PRIMARY KEY (root)
+);

pkg/fx/app/pdp.go

@@ -6,8 +6,8 @@ import (
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/ethclient"
-	"github.com/fil-forge/piri/pkg/pdp/aggregation"
 	"github.com/fil-forge/piri/pkg/pdp/piece"
+	"github.com/fil-forge/piri/pkg/pdp/pipeline"
 	"github.com/fil-forge/piri/pkg/pdp/smartcontracts"
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/client"
@@ -36,7 +36,7 @@ var PDPModule = fx.Module("pdp",
 		),
 	),
 	smartcontracts.Module,
-	aggregation.Module,
+	pipeline.Module, // aggregation pipeline (commp/aggregate/add-roots) + removal sweep as harmonytasks
 	curiopdp.Module, // Curio pdpv0 pipeline (harmonytask + prove/proving-period) on harmonydb
 	pdp.Module,
 	piece.Module,

pkg/fx/database/provider.go

@@ -19,7 +19,6 @@ import (
 // PostgreSQL schema names for each logical database
 const (
 	SchemaReplicator    = "replicator"
-	SchemaAggregator    = "aggregator"
 	SchemaEgressTracker = "egress_tracker"
 )
 
@@ -29,10 +28,6 @@ var Module = fx.Module("database",
 			ProvideReplicatorDB,
 			fx.ResultTags(`name:"replicator_db"`),
 		),
-		fx.Annotate(
-			ProvideAggregatorDB,
-			fx.ResultTags(`name:"aggregator_db"`),
-		),
 		fx.Annotate(
 			ProvideEgressTrackerDB,
 			fx.ResultTags(`name:"egress_tracker_db"`),
@@ -91,57 +86,6 @@ func ProvideReplicatorDB(lc fx.Lifecycle, cfg app.StorageConfig) (*sql.DB, error
 	return db, nil
 }
 
-// ProvideAggregatorDB provides the database for the aggregator job queue.
-// Supports both SQLite (default) and PostgreSQL backends.
-func ProvideAggregatorDB(lc fx.Lifecycle, cfg app.StorageConfig) (*sql.DB, error) {
-	var db *sql.DB
-	var err error
-
-	if cfg.Database.IsPostgres() {
-		// Use PostgreSQL with separate schema
-		opts := postgresdb.OptionsFromConfig(cfg.Database.Postgres)
-		db, err = postgresdb.New(cfg.Database.Postgres.URL.String(), SchemaAggregator, opts...)
-		if err != nil {
-			return nil, fmt.Errorf("creating postgres aggregator database: %w", err)
-		}
-	} else {
-		// Use SQLite (default) - derive path from DataDir
-		dbPath := sqliteDBPath(cfg.DataDir, "aggregator", "jobqueue", "jobqueue.db")
-		if dbPath == "" {
-			db, err = sqlitedb.NewMemory()
-			if err != nil {
-				return nil, fmt.Errorf("creating in-memory aggregator database: %w", err)
-			}
-		} else {
-			// Ensure directory exists for file-based database
-			if err := ensureSQLiteDir(dbPath); err != nil {
-				return nil, fmt.Errorf("creating aggregator database directory: %w", err)
-			}
-
-			db, err = sqlitedb.New(dbPath,
-				database.WithJournalMode(database.JournalModeWAL),
-				database.WithTimeout(5*time.Second),
-				database.WithSyncMode(database.SyncModeNORMAL),
-			)
-			if err != nil {
-				return nil, fmt.Errorf("creating aggregator database: %w", err)
-			}
-			configureSQLiteConnection(db)
-		}
-	}
-
-	lc.Append(fx.Hook{
-		OnStart: func(ctx context.Context) error {
-			return db.PingContext(ctx)
-		},
-		OnStop: func(ctx context.Context) error {
-			return db.Close()
-		},
-	})
-
-	return db, nil
-}
-
 // ProvideEgressTrackerDB provides the database for the egress tracker job queue.
 // Supports both SQLite (default) and PostgreSQL backends.
 func ProvideEgressTrackerDB(lc fx.Lifecycle, cfg app.StorageConfig) (*sql.DB, error) {

pkg/fx/pdp/provider.go

@@ -12,7 +12,6 @@ import (
 
 	// curio infra
 	"github.com/filecoin-project/curio/harmony/harmonydb"
-	"github.com/filecoin-project/curio/harmony/harmonytask"
 	"github.com/filecoin-project/curio/lib/chainsched"
 	"github.com/filecoin-project/curio/lib/ethchain"
 	"github.com/filecoin-project/curio/tasks/message"
@@ -26,6 +25,7 @@ import (
 	"github.com/fil-forge/piri/pkg/service/proofs"
 	"github.com/fil-forge/piri/pkg/service/signer"
 	"github.com/fil-forge/piri/pkg/store/acceptancestore"
+	"github.com/fil-forge/piri/pkg/store/allocationstore"
 	"github.com/fil-forge/piri/pkg/store/blobstore"
 	"github.com/fil-forge/piri/pkg/store/receiptstore"
 )
@@ -49,6 +49,7 @@ var Module = fx.Module("pdp-service",
 			// concrete *PDPService receivers.
 			fx.As(new(types.PieceWriterAPI)),
 			fx.As(new(types.PieceCommPAPI)),
+			fx.As(new(types.PieceRemoverAPI)),
 		),
 		ProvideProofSetIDProvider,
 		fx.Annotate(
@@ -68,11 +69,11 @@ type Params struct {
 	Config           app.PDPServiceConfig
 	BlobStore        blobstore.Blobstore
 	AcceptanceStore  acceptancestore.AcceptanceStore
+	AllocationStore  allocationstore.AllocationStore
 	ReceiptStore     receiptstore.ReceiptStore
 	Resolver         types.PieceResolverAPI
 	Reader           types.PieceReaderAPI
 	Sender           *message.SenderETH
-	Engine           *harmonytask.TaskEngine
 	ChainScheduler   *chainsched.CurioChainSched
 	ChainClient      service.ChainClient
 	EthClient        ethchain.EthClient // raw eth client — contract.FSRegister signs/sends the register tx and reads balance
@@ -91,11 +92,11 @@ func ProvidePDPService(params Params) (*service.PDPService, error) {
 		params.DB,
 		params.BlobStore,
 		params.AcceptanceStore,
+		params.AllocationStore,
 		params.ReceiptStore,
 		params.Resolver,
 		params.Reader,
 		params.Sender,
-		params.Engine,
 		params.ChainScheduler,
 		params.ChainClient,
 		params.EthClient,