pdp: integrate Curio's PDP pipeline (harmonydb), remove GORM, adopt FSRegister/SettleTask by frrist · Pull Request #22 · fil-forge/piri

frrist · 2026-06-26T21:22:48Z

Summary

Piri's PDP core is now Curio's, consumed as a Go module: pkg/curiopdp wires harmonydb (Postgres-only), harmonytask, and the pdpv0 tasks (init/next proving period, prove, settle, watchers) into the fx graph. Piri's bespoke scheduler/chainsched/sender tasks and the GORM half of the dual-database model are removed; registration goes through Curio's contract.FSRegister and settlement through tasks/pay. The UCAN layer is unchanged, and the binary builds CGO-free (skiff tags).

Validation (smelt devnet, end-to-end)

Boot, on-chain provider registration, proof-set creation, upload/retrieve.
Full proving loop: aggregation → addPieces confirmed → proving-period init → provePossession accepted by the PDPVerifier (receipt status 1).
Ungraceful restart mid-pipeline recovers through to a verified proof.

Dependencies

filecoin-project/curio#1317 has merged; go.mod now pins the mainline merge commit filecoin-project/curio@38ca280c (v1.28.3-0.20260717015646-38ca280c43a2). No remaining blockers.

alanshaw

Submitting what I have so far...

alanshaw · 2026-07-20T15:36:09Z


 func init() {
 	Cmd.AddCommand(log.Cmd)
-	Cmd.AddCommand(payment.Cmd)


Why were these CLI commands removed?

Two reasons this went away:

We now import Curio's settlement task (pay.SettleTask + SettleWatcher), which triggers settlement autonomously. In the general case every 7 days.

This isn't just convenience automation. Without it, an operator who never settles (or waits a very long time) hits an out-of-gas failure: From the smart contract code settleRail(railId, untilEpoch) runs _settleWithRateChanges, whose while (processedEpoch < targetEpoch) loop walks the rate-change segments across the entire unsettled window. The wider that window, the more iterations and the more gas, until a single settleRail exceeds the block gas limit and can never land. We saw exactly this near the end of the Storacha Forge network. Curio's task defends against it two ways: the 7-day cadence keeps the window bounded, and calculateSettleUpTo binary-searches untilEpoch downward until the gas estimate fits, settling partially rather than reverting. We could modify our old code to behave this way, but borrowing from Curio seemed like the right choice here.

The problem this presents is the loss of a withdraw command/capability. Previously the task did both settle and withdraw, now we just have settle with withdraw left to the operator to figure out - which is not great....

Two random thoughts here:

The forever work in progress pay.filecoin.cloud dashboard ought to address this, but we've been asking for this feature for nearly 2 years and its never arived 😠

The offical Filecoin PDP docs have zero content on how a PDP operator in FOC land get paid: https://docs.filecoin.io/provide-storage/pdp/install-and-run-pdp

So net-net here: I think we'll either need to get clarification from the FOC team and/or reintroduce a withdraw command to Piri, similar to the previous one, but constratined to just call FilecoinPayV1.withdraw(token, amount)

ACK, ok makes sense. Worth opening an issue for re-introducing the withdraw feature and explain the situation?

alanshaw · 2026-07-20T15:47:51Z

Can we consume Curio's existing SQL migrations? How do we consume a future migration from Curio?

The SQL schema is the weakest layer of this integration. We can't consume Curio's migrations wholesale as you noticed in the second file, their migration set is Curio's entire schema (sector pipeline, market, alerting, etc.), so importing it drags 100+ tables unrelated to PDP into every Piri deployment.

So what's checked in here is a pg_dump of a Curio DB that I hand-distilled down to the PDP keep-set (pdp_*, harmony_*, parked_*, message_*eth*, eth_keys). filecoin_payment_transactions (00000001) is a follow-on because it doesn't match that prefix set but tasks/pay needs it.

I don't have a great answer short of a schema library shared by both projects. Open to alternatives — realistically this might need more coordination with the Curio folks, and now that the curio-piri library-ization has merged upstream, asking them to expose a PDP-scoped embeddable migration FS is probably the cleanest path. Happy to file an issue to track that?

In the meantime, one stopgap could be a CI job that applies Curio's full migration set, filters to the keep-set, and diffs against our snapshot. If there is a diff we fail CI, wdyt?

I don't have a great answer short of a schema library shared by both projects. Open to alternatives — realistically this might need more coordination with the Curio folks, and now that the curio-piri library-ization has merged upstream, asking them to expose a PDP-scoped embeddable migration FS is probably the cleanest path. Happy to file an issue to track that?

Yes agree.

In the meantime, one stopgap could be a CI job that applies Curio's full migration set, filters to the keep-set, and diffs against our snapshot. If there is a diff we fail CI, wdyt?

I would wait and see what happens with the exposed PDP-scoped embeddable migration FS proposal...

alanshaw · 2026-07-20T15:49:07Z

+-- (adopted by Piri). It was dropped from the PDP closure (00000000) because the
+-- closure-extraction keep-set only matched pdp_* / harmony_* / parked_* /
+-- message_*eth* / eth_keys, and this table doesn't match that prefix set.
+-- Applied as a follow-on migration (harmonyquery runs it after the closure).


Oh I see their migrations include stuff that is not related to PDP...

alanshaw · 2026-07-20T16:52:08Z

+const taskGroup = `group:"curio_harmonytasks"`
+
+// harmonyMachineID identifies this node in harmony_machines.
+// TODO(e2e): wire to Piri's real listen address; must be unique per node.


Can it not be a IP/port? Seems brittle if a user needs to change the port for some reason....could it just be the node DID?

Yes, it can be — inside the embedded harmonytask engine host_and_port is just an opaque identity string (machine registration, cordon checks, task-history attribution); nothing dials or parses it, and the Curio components that do treat it as an address (cluster UI, inter-node RPC) aren't part of what piri embeds. A DID is also the sturdier uniqueness story. Tracked in #39, which resolves the TODO(e2e) on that const.

alanshaw · 2026-07-20T16:53:00Z

+// pdpServiceLabel must match PDPService.name — pdp_data_set_creates.service and
+// pdp_piece_uploads.service FK to pdp_services(service_label), so a row with this
+// label must exist before CreateProofSet / piece upload.
+const pdpServiceLabel = "storacha"


alanshaw

Submitting further feedback. Still not finished reviewing.

What happens in SQLite mode now? Does it even work?

alanshaw · 2026-07-21T15:04:35Z

+		emptyBuffer := Aggregation{
+			Roots: []cid.Cid{},
+		}
+		if err := sw.store.Put(ctx, aggBufferKey{}, emptyBuffer); err != nil {


Should this be in a transaction then?

There's no transaction to reach for here — the buffer sits on a KV blob store, and the init path is single-threaded at startup, so the check-then-put has no concurrent writer. The real atomicity problem is the one the comment alludes to (enqueue+clear spanning the buffer and the queue), and the actual fix is #30: it deletes this buffer entirely and moves the aggregation pipeline into harmonydb, where every stage transition is transactional. So: yes in spirit, and #30 is that answer.

alanshaw · 2026-07-21T15:05:41Z

 	return out, nil
 }

-func (c *Client) RepairProofSet(ctx context.Context, proofSetID uint64) (*types.RepairResult, error) {


Why does repair go away?

Its been removed under the assumption that Curio core obviates the need for it, and since re-implementing it on the curio-based schema would be a large diff here. Tracked via #36

alanshaw · 2026-07-21T15:13:16Z

+	if _, err := p.db.Exec(ctx,
+		`INSERT INTO pdp_piece_uploads (id, service, notify_url, check_hash_codec, check_hash, check_size)
+		 VALUES ($1, $2, $3, $4, $5, $6)`,
+		uploadUUID.String(), "storacha", notifyURL, allocation.Piece.Name, allocation.Piece.Hash, allocation.Piece.Size); err != nil {


Lets define/use a const instead of magic string.

alanshaw · 2026-07-21T15:19:38Z

+		// 3. Create a reference in pdp_piecerefs.
+		if _, err := tx.Exec(
+			`INSERT INTO pdp_piecerefs (service, piece_cid, piece_ref) VALUES ($1, $2, $3)`,
+			"storacha", params.PieceCID.String(), refID); err != nil {


Same "storacha" as const please!

alanshaw · 2026-07-21T15:23:53Z

-				}).Error; err != nil {
-				return types.WrapError(types.KindInternal, "failed to create pieceMH to commp", err)
+			if _, err := tx.Exec(
+				`INSERT INTO pdp_piece_mh_to_commp (mhash, size, commp, commp_v1) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING`,


I'm not sure what the value of storing both of these is, since we have both and can always convert from 2 to 1. Is commp_v1 even used?

Yes — pkg/pdp/piece/resolver.go:95 queries WHERE commp_v1 = $1. Curio's pdpv0 tables store v1-form piece CIDs, so when the prove task's piece reader hands us a piece to fetch, the lookup key arrives in v1 form and we resolve it back to the blob multihash. We store both forms because you can't index your way from one to the other at query time: v1→v2 needs the raw size, and a v2-only column would make every v1-keyed lookup a per-row conversion instead of an indexed hit (there's a btree index on commp_v1 in the closure schema for exactly this).

alanshaw · 2026-07-21T15:25:50Z

+		n, err := tx.Exec(`
+            INSERT INTO message_waits_eth (signed_tx_hash, tx_status)
+            VALUES ($1, $2)
+        `, txHashLower, "pending")


Might be nice to have a status enum?

summarized in with some other coordination bits: #40

alanshaw · 2026-07-21T15:33:05Z

+		ProvingPeriod             *int64  `db:"proving_period"`
+		ChallengeWindow           *int64  `db:"challenge_window"`
+		ChallengeRequestMsgHash   *string `db:"challenge_request_msg_hash"`
+	}


It strikes me that the database types we're defining here could/should be exported from Curio for us to use.

summarized in with some other coordination bits: #40

alanshaw

I have some concerns in roots_add.go that need to be resolved but I think otherwise all the rest of my feedback is non-blocking.

alanshaw · 2026-07-22T13:36:20Z

 	rootCIDs := make([]string, len(request))
 	for i, req := range request {
-		rootCIDs[i] = req.Root.String()
+		v1, err := asPieceCIDv1(req.Root.String())


Don't we have a library for this? We always seem to have a CID when we call this - why are we encoding to a string and then immediately decoding?

The library is go-fil-commcid — the helper is just a thin wrapper over commcid.PieceCidV1FromV2 that adds a pass-through for CIDs already in v1 form (commcid errors on v1 input). The string round-trip was pointless though. I've changed the signature to take cid.Cid directly and dropped the decode at all call sites.

On why we convert to v1 at all here (cause we didn't previously!): this is required by Curio's task engine, not a style choice. Pre-Curio, piri was v2 end-to-end — we stored raw v2 CIDs in the root-add tables and ran our own prove task that decoded them with PieceCidV2ToDataCommitment (old pkg/pdp/tasks/prove_pdp.go). That code is ofc gone; we now run Curio's stock pdpv0 prove task, which reads the piece/sub_piece columns and converts them with commcid.CIDToPieceCommitmentV1 (curio tasks/pdpv0/task_prove.go:782,919) — that function requires the fil-commitment-unsealed (v1) codec and returns ErrIncorrectCodec on a v2 CID. Curio's own writer does the same v2→v1 conversion at the API boundary before insert (pdp/handlers_add.go → insertPieceAdds). So the API stays v2, but anything hitting these tables has to be v1 or proving breaks.

alanshaw · 2026-07-22T14:40:42Z

+			"proofset_id", id, "dead_txs", hashes)
+		if _, err := p.db.Exec(ctx, `
+			DELETE FROM pdp_data_set_piece_adds
+			WHERE data_set = $1 AND add_message_hash = ANY($2)


add_message_hash is the hash of the message to add roots? Should this be AND root = ANY($2) where $2 is rootCIDs? i.e. we are currently cleaning more than we need to here - is that necessary/intended?

Agreed it looks obtuse, but it's intentional — I've added a comment in the code expanding on it. Two parts to the answer:

Why by tx hash and not by piece: a failed/reverted tx is dead for every piece it staged, not just the ones in this request. The watcher only promotes piece_adds → pieces on confirmed success, so rows under a dead tx can never progress — they're garbage regardless of what this call asked for. Scoping the delete to piece = ANY($2) would just leave a half-cleared dead tx whose remaining rows linger until (unless) a later call happens to include those pieces. Clearing the whole tx loses nothing: any caller retrying the other pieces finds no staged rows and resubmits them as new. A piece staged under both a dead tx and a live one keeps its live rows and stays classified in-flight.

Why delete at all: stock Curio actually never deletes these rows — a DB trigger marks them add_message_ok = FALSE on tx failure and every Curio consumer (addpiece watcher, piece GC, dataset verify) filters them out as permanent tombstones; retries are the client's problem. Piri's AddRoots is that retrying client, and this guard is the only reader that would ever act on those rows — deleting them keeps the idempotency partition simple and stops repeated AddRoots calls from re-fetching an ever-growing dead row-set. (WaitForConfirmation only detects the failure and errors the call; it does no cleanup.)

alanshaw · 2026-07-22T14:57:47Z

+		return true, nil
+	}, harmonydb.OptionRetry())
+	if err != nil {
+		log.Errorw("Failed to insert into database", "error", err, "txHash", txHashLower, "subroots", subrootInfoMap)


This seems bad - we've sent a blockchain transaction to add roots but subsequently failed to record the roots or the transaction. Shouldn't this be the other way around - only send the transaction if we can record that there is a transaction to wait on and these are the roots added in it. If we fail to send the blockchain transaction then we rollback/set failed status?

Agreed the window is real, but it's a pre-existing gap (documented in the TODO block at the top of this file) rather than something this PR introduces, and it's also exactly how Curio orders it — handleAddPieceToDataSet does sender.Send then the message_waits_eth/piece_adds insert (curio pdp/handlers_add.go:421 → :435). We're deliberately mirroring Curio's write-shape here since its watcher/prove tasks consume these tables.

The naive inversion doesn't work with Curio's sender: the rows are keyed on the signed tx hash, which doesn't exist until Curio's send task signs the tx (nonce assignment and gas-estimation rebuild happen inside SenderETH, which enqueues a harmony task and polls). It's also not a silent black hole — Send durably records the tx in message_sends_eth (unsigned before signing, signed hash before broadcast), so a crash in the window leaves a reconcilable trace; the exposure is a duplicate addPieces submission on retry, in a narrow crash window. The fix we have sketched (TODO at top of file) is placeholder rows keyed by a deterministic request ID written before Send, swapped to the real hash after — works without touching Curio. Given this PR already narrows the recovery behavior (dead-tx clearing means retries no longer wedge on stale rows), I'd rather land that as a follow-up — will open an issue. #34

alanshaw

PTAL at the other comments - would be good to get at least some of it resolved but otherwise this LGTM (since we are tracking the edge case bug).

…el implementation (#22) Replace Piri's homegrown PDP machinery (GORM models, sender, task engine, chain scheduler, prove task) with Curio's exported pdpv0 pipeline, now that the curio-piri library-ization has merged upstream (pinned to mainline curio @ 38ca280c43a2). - Run Curio's harmonytask scheduler and pdpv0 tasks (prove, proving period init/next, watchers) against harmonydb (Postgres); Piri wires them via fx (pkg/curiopdp). - Ship a hand-distilled PDP-scoped schema closure (pdp-closure.sql): pdp_*, harmony_*, parked_*, message_*eth*, eth_keys, plus filecoin_payment_transactions for tasks/pay. (#40 tracks upstreaming a PDP-scoped migration FS.) - Remove GORM throughout; adopt Curio's FSRegister provisioning and automatic payment SettleTask (withdraw gap tracked in #37; removed repair command breadcrumbs in #36). - Store piece CIDs in Curio's v1 CommP form (its prove task requires fil-commitment-unsealed); keep the v2 form at the API boundary. - Make AddRoots idempotent: partition requests per root into completed / in-flight / dead-tx / new; clear dead-tx piece-add rows so retries resubmit instead of wedging (send-then-record crash window tracked in #34). - Don't reset the aggregation manager submission buffer on startup — preserves buffered-but-unflushed roots across restarts. - Build CGO-free via Curio's 'skiff' FFI-free variants; Go 1.26 builder. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

…ng one Four findings, none in the test's behaviour. The code is unchanged; all four were claims made about it. THE PANIC NEEDS -count=5 AND A FAILING CONDITION, and three revisions of that paragraph have now each supplied a different half. Round two said -count=5; the rewrite dropped it and substituted "a deadline short enough to fail" -- the wrong half, in the very sentence meant to supply the missing precondition. Measured: 0 panics in 10 at -count=1, 10 in 10 at -count=5. TestPeriodicRotator is the last test in this package, so at -count=1 the process exits before the stray log lands. "UNDRAINED IMPLIES STILL PRODUCING" DOES NOT FOLLOW, and it is false in exactly the case the drain half exists for: with an entry the rotator can never reach, the rotator is stuck rather than producing, require aborts, and nothing panics -- 0 in 10 at -count=5, against 10 in 10 for a rotator still running. THE LOCK'S STATED REASON CANNOT HAPPEN. Stop()'s ctx.Err() branch does return without joining, but require.NoError on the line above Goexits first, so execution never reaches the lock with the goroutine alive. The lock is redundant and kept for consistency, which is what the comment now says. Also, the PR body's table claimed the previous head did not detect an unreachable fixture entry. It detects it 20 times in 20. What the last push changed there is the added "Condition never satisfied" line, not detection where there was none -- a "before" column written from expectation rather than from a run. Comments only: every changed line is a `//` line, so nothing a check runs can behave differently. Rounds stop here under rule 10 as amended by forge #22, which skips a round when that is all that changed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CGAGAib517Ae1kg8SCdcEt

frrist requested a review from alanshaw as a code owner June 26, 2026 21:22

frrist self-assigned this Jun 26, 2026

frrist marked this pull request as draft June 26, 2026 21:24

frrist force-pushed the frrist/hello-again-curio branch from c3a66fc to 504dd83 Compare July 16, 2026 19:46

frrist marked this pull request as ready for review July 16, 2026 21:47

alanshaw reviewed Jul 20, 2026

View reviewed changes

alanshaw reviewed Jul 21, 2026

View reviewed changes

alanshaw requested changes Jul 22, 2026

View reviewed changes

frrist mentioned this pull request Jul 22, 2026

AddRoots: crash window between tx send and DB record can cause duplicate addPieces submissions #34

Open

alanshaw approved these changes Jul 22, 2026

View reviewed changes

frrist force-pushed the frrist/hello-again-curio branch from 1eb81a4 to cb62fcf Compare July 22, 2026 20:24

frrist force-pushed the frrist/hello-again-curio branch from cb62fcf to 7de588e Compare July 22, 2026 21:20

frrist merged commit 939de00 into main Jul 22, 2026
8 checks passed

frrist deleted the frrist/hello-again-curio branch July 22, 2026 22:56

Conversation

frrist commented Jun 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Validation (smelt devnet, end-to-end)

Dependencies

Uh oh!

alanshaw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alanshaw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alanshaw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

frrist commented Jun 26, 2026 •

edited

Loading