Repository navigation
implement blob/release and blob/reject; migrate jobqueue into harmony task engine #30
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,52 +1,21 @@ | ||
| package curiopdp | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/filecoin-project/curio/harmony/harmonydb" | ||
|
|
||
| "github.com/fil-forge/piri/pkg/config/app" | ||
| "github.com/fil-forge/piri/pkg/curiopdp/schema" | ||
| ) | ||
|
|
||
| // ProvideHarmonyDB builds a harmonydb.DB (harmonyquery on Postgres) that applies | ||
| // 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)") | ||
| } | ||
| return schema.NewDB(cfg.Database.Postgres.URL) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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, | ||
| }) | ||
| } |
File renamed without changes.
File renamed without changes.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
| ); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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(), | ||
|
Member
There was a problem hiding this comment. IDK is it a good idea to have timestamps for lifecycle events, like
Member
Author
There was a problem hiding this comment. Took the idea in f6ea205 — added |
||
| 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) | ||
| ); | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's kinda odd for
commpandaggregate_rootto betextandblobto bebyteano?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It mirrors the curio tables this pipeline joins against:
pdp_piece_mh_to_commpis exactlymhash bytea+commp text, and piece CIDs are text throughout harmonydb (pdp_piecerefs.piece_cid,pdp_data_set_pieces.sub_piece, ...). So the convention is: multihash digests as raw bytes, piece CIDs as text strings. I've added a comment to the schema documenting this, and renamed the column todigest(per your other comment) so it no longer reads as holding the blob's bytes — 12b1710.