feat(subgraph): add PiecesAddedV2 event handler (#685)

feat: add PiecesAddedV2 event handler

Author: silent-cipher

apps/subgraph/scripts/sync-abis.mjs

@@ -13,8 +13,36 @@ import { fwss, pdp } from "@filoz/synapse-core/abis";
 const here = dirname(fileURLToPath(import.meta.url));
 const abisDir = join(here, "..", "abis");
 
+// TODO(https://github.com/FilOzone/dealbot/issues/683): PDPVerifier PR
+// https://github.com/FilOzone/pdp/pull/300 adds a compact PiecesAddedV2 event
+// but hasn't been merged/deployed/picked up by synapse-core yet, so it's
+// still missing from the fetched ABI. Inject it by hand so codegen produces
+// the handlePiecesAddedV2 bindings; drop this once synapse-core ships it
+// natively (`pdp.some(e => e.name === "PiecesAddedV2")` will start passing).
+const PIECES_ADDED_V2_EVENT = {
+  type: "event",
+  name: "PiecesAddedV2",
+  inputs: [
+    { name: "setId", type: "uint256", indexed: true, internalType: "uint256" },
+    { name: "firstPieceId", type: "uint256", indexed: false, internalType: "uint256" },
+    {
+      name: "pieceCids",
+      type: "tuple[]",
+      indexed: false,
+      internalType: "struct Cids.PackedCid[]",
+      components: [
+        { name: "header", type: "bytes32", internalType: "bytes32" },
+        { name: "root", type: "bytes32", internalType: "bytes32" },
+      ],
+    },
+  ],
+  anonymous: false,
+};
+
+const pdpPatched = pdp.some((entry) => entry.name === "PiecesAddedV2") ? pdp : [...pdp, PIECES_ADDED_V2_EVENT];
+
 const targets = [
-  { file: "PDPVerifier.json", abi: pdp },
+  { file: "PDPVerifier.json", abi: pdpPatched },
   { file: "FilecoinWarmStorageService.json", abi: fwss },
 ];
 

apps/subgraph/src/helpers.ts

@@ -160,6 +160,22 @@ export function validateCommPv2(cidData: Bytes): CommPv2ValidationResult {
   return new CommPv2ValidationResult(true, padding, height, offset);
 }
 
+// Reconstructs a full CID byte string from a Cids.PackedCid (header, root), as emitted by
+// PDPVerifier.PiecesAddedV2. `header` is right-aligned in its 32-byte word with zero padding on
+// the left; the CID is recovered by dropping only those leading zero bytes and appending `root`.
+export function reconstructCidFromPackedCid(header: Bytes, root: Bytes): Bytes {
+  let firstNonZero = header.length;
+  for (let i = 0; i < header.length; i++) {
+    if (header[i] != 0) {
+      firstNonZero = i;
+      break;
+    }
+  }
+
+  const headerBytes = Bytes.fromUint8Array(header.slice(firstNonZero));
+  return headerBytes.concat(root);
+}
+
 export function unpaddedSize(padding: BigInt, height: u8): BigInt {
   if (height > 58) {
     return BigInt.zero();

apps/subgraph/src/pdp-verifier.ts

@@ -1,10 +1,11 @@
-import { BigInt, log } from "@graphprotocol/graph-ts";
+import { BigInt, Bytes, log } from "@graphprotocol/graph-ts";
 import {
   DataSetCreated as DataSetCreatedEvent,
   DataSetDeleted as DataSetDeletedEvent,
   DataSetEmpty as DataSetEmptyEvent,
   NextProvingPeriod as NextProvingPeriodEvent,
   PiecesAdded as PiecesAddedEvent,
+  PiecesAddedV2 as PiecesAddedV2Event,
   PiecesRemoved as PiecesRemovedEvent,
   PossessionProven as PossessionProvenEvent,
   StorageProviderChanged as StorageProviderChangedEvent,
@@ -15,6 +16,7 @@ import {
   getRootEntityId,
   getRootSampleKey,
   maxProvingPeriodFor,
+  reconstructCidFromPackedCid,
   unpaddedSize,
   validateCommPv2,
 } from "./helpers";
@@ -163,12 +165,50 @@ export function handleNextProvingPeriod(event: NextProvingPeriodEvent): void {
   }
 }
 
+// Shared by handlePiecesAdded and handlePiecesAddedV2: creates the Root entity for one added
+// piece. Returns false (and logs) if the Root already exists, so the caller can skip counting it
+// toward the EMPTY -> READY transition.
+function createRootFromPieceCid(
+  setId: BigInt,
+  rootId: BigInt,
+  pieceCidBytes: Bytes,
+  proofSetEntityId: Bytes,
+  blockTimestamp: BigInt,
+): boolean {
+  const commPData = validateCommPv2(pieceCidBytes);
+  const rawSize = commPData.isValid ? unpaddedSize(commPData.padding, commPData.height) : BigInt.zero();
+
+  const rootEntityId = getRootEntityId(setId, rootId);
+  if (Root.load(rootEntityId) != null) {
+    log.warning("createRootFromPieceCid: Root {} for Set {} already exists; skipping", [
+      rootId.toString(),
+      setId.toString(),
+    ]);
+    return false;
+  }
+
+  const root = new Root(rootEntityId);
+  root.setId = setId;
+  root.rootId = rootId;
+  root.rawSize = rawSize;
+  root.cid = pieceCidBytes;
+  root.removed = false;
+  root.createdAt = blockTimestamp;
+  root.proofSet = proofSetEntityId;
+  root.sampleKey = getRootSampleKey(rootEntityId);
+  // ipfsRootCID: patched in FWSS handler if applicable.
+  root.save();
+
+  return true;
+}
+
 export function handlePiecesAdded(event: PiecesAddedEvent): void {
   const setId = event.params.setId;
   const rootIdsFromEvent = event.params.pieceIds;
   const pieceCidsFromEvent = event.params.pieceCids;
 
-  const proofSet = DataSet.load(getProofSetEntityId(setId));
+  const proofSetEntityId = getProofSetEntityId(setId);
+  const proofSet = DataSet.load(proofSetEntityId);
   if (proofSet == null) {
     log.warning("handlePiecesAdded: DataSet {} not found", [setId.toString()]);
     return;
@@ -177,35 +217,48 @@ export function handlePiecesAdded(event: PiecesAddedEvent): void {
   let addedAny = false;
 
   for (let i = 0; i < rootIdsFromEvent.length; i++) {
-    const rootId = rootIdsFromEvent[i];
-    const pieceCid = pieceCidsFromEvent[i];
-
-    const pieceBytes = pieceCid.data;
-    const commPData = validateCommPv2(pieceBytes);
-    const rawSize = commPData.isValid ? unpaddedSize(commPData.padding, commPData.height) : BigInt.zero();
-
-    const rootEntityId = getRootEntityId(setId, rootId);
-    if (Root.load(rootEntityId) != null) {
-      log.warning("handlePiecesAdded: Root {} for Set {} already exists; skipping", [
-        rootId.toString(),
-        setId.toString(),
-      ]);
-      continue;
-    }
+    const created = createRootFromPieceCid(
+      setId,
+      rootIdsFromEvent[i],
+      pieceCidsFromEvent[i].data,
+      proofSetEntityId,
+      event.block.timestamp,
+    );
+    if (created) addedAny = true;
+  }
 
-    const root = new Root(rootEntityId);
-    root.setId = setId;
-    root.rootId = rootId;
-    root.rawSize = rawSize;
-    root.cid = pieceBytes;
-    root.removed = false;
-    root.createdAt = event.block.timestamp;
-    root.proofSet = getProofSetEntityId(setId);
-    root.sampleKey = getRootSampleKey(rootEntityId);
-    // ipfsRootCID: patched in FWSS handler if applicable.
-    root.save();
+  // First non-empty add transitions the DataSet to READY. NextProvingPeriod
+  // will then promote it to PROVING.
+  if (addedAny && proofSet.status == DataSetStatus.EMPTY) {
+    proofSet.status = DataSetStatus.READY;
+    proofSet.save();
+  }
+}
+
+// Handles the compact PiecesAddedV2 event (PDPVerifier PR #300), which replaces the legacy
+// PiecesAdded for all new piece additions (PiecesAdded is kept only for historical logs). Piece
+// IDs are contiguous from firstPieceId, and each CID is packed as (header, root) instead of raw
+// bytes; a single addPieces call can emit several of these events once pieces are batched.
+export function handlePiecesAddedV2(event: PiecesAddedV2Event): void {
+  const setId = event.params.setId;
+  const firstPieceId = event.params.firstPieceId;
+  const pieceCidsFromEvent = event.params.pieceCids;
+
+  const proofSetEntityId = getProofSetEntityId(setId);
+  const proofSet = DataSet.load(proofSetEntityId);
+  if (proofSet == null) {
+    log.warning("handlePiecesAddedV2: DataSet {} not found", [setId.toString()]);
+    return;
+  }
+
+  let addedAny = false;
+
+  for (let i = 0; i < pieceCidsFromEvent.length; i++) {
+    const rootId = firstPieceId.plus(BigInt.fromI32(i));
+    const pieceCidBytes = reconstructCidFromPackedCid(pieceCidsFromEvent[i].header, pieceCidsFromEvent[i].root);
 
-    addedAny = true;
+    const created = createRootFromPieceCid(setId, rootId, pieceCidBytes, proofSetEntityId, event.block.timestamp);
+    if (created) addedAny = true;
   }
 
   // First non-empty add transitions the DataSet to READY. NextProvingPeriod

apps/subgraph/subgraph.yaml

@@ -32,6 +32,8 @@ dataSources:
           handler: handleDataSetDeleted
         - event: PiecesAdded(indexed uint256,uint256[],(bytes)[])
           handler: handlePiecesAdded
+        - event: PiecesAddedV2(indexed uint256,uint256,(bytes32,bytes32)[])
+          handler: handlePiecesAddedV2
         - event: PiecesRemoved(indexed uint256,uint256[])
           handler: handlePiecesRemoved
         - event: DataSetEmpty(indexed uint256)

apps/subgraph/tests/pdp-verifier-utils.ts

@@ -6,6 +6,7 @@ import {
   DataSetEmpty,
   NextProvingPeriod,
   PiecesAdded,
+  PiecesAddedV2,
   PossessionProven,
 } from "../generated/PDPVerifier/PDPVerifier";
 
@@ -94,6 +95,55 @@ export function createRootsAddedEvent(
   return rootsAddedEvent;
 }
 
+// header/root packing of the same CID used by createRootsAddedEvent
+// (0x01559120258ff7f7021387dcea7164b7d1c4a98bd6f8d3c187e3114795efa391df307c8aa9d5d5cbac03):
+// the header is right-aligned/zero-padded to 32 bytes, the trailing 32 bytes are the digest ("root").
+export const PACKED_CID_HEADER = "0x0000000000000000000000000000000000000000000001559120258ff7f70213";
+export const PACKED_CID_ROOT = "0x87dcea7164b7d1c4a98bd6f8d3c187e3114795efa391df307c8aa9d5d5cbac03";
+
+export function createPiecesAddedV2Event(
+  setId: BigInt,
+  firstPieceId: BigInt,
+  pieceCount: i32,
+  sender: Address,
+  contractAddress: Address,
+  blockNumber: BigInt = BigInt.fromI32(1),
+  timestamp: BigInt = BigInt.fromI32(1),
+  txHash: Bytes = generateTxHash(6),
+  logIndex: BigInt = BigInt.fromI32(0),
+): PiecesAddedV2 {
+  const piecesAddedV2Event = changetype<PiecesAddedV2>(newMockEvent());
+
+  piecesAddedV2Event.parameters = [];
+  piecesAddedV2Event.address = contractAddress;
+  piecesAddedV2Event.transaction.from = sender;
+  piecesAddedV2Event.transaction.to = contractAddress;
+
+  const setIdParam = new ethereum.EventParam("setId", ethereum.Value.fromUnsignedBigInt(setId));
+  const firstPieceIdParam = new ethereum.EventParam("firstPieceId", ethereum.Value.fromUnsignedBigInt(firstPieceId));
+
+  const pieceCids: Array<ethereum.Tuple> = [];
+  for (let i = 0; i < pieceCount; i++) {
+    const packedCidTuple = new ethereum.Tuple();
+    packedCidTuple.push(ethereum.Value.fromFixedBytes(Bytes.fromHexString(PACKED_CID_HEADER)));
+    packedCidTuple.push(ethereum.Value.fromFixedBytes(Bytes.fromHexString(PACKED_CID_ROOT)));
+    pieceCids.push(packedCidTuple);
+  }
+
+  const pieceCidsParam = new ethereum.EventParam("pieceCids", ethereum.Value.fromTupleArray(pieceCids));
+
+  piecesAddedV2Event.parameters.push(setIdParam);
+  piecesAddedV2Event.parameters.push(firstPieceIdParam);
+  piecesAddedV2Event.parameters.push(pieceCidsParam);
+
+  piecesAddedV2Event.block.number = blockNumber;
+  piecesAddedV2Event.block.timestamp = timestamp;
+  piecesAddedV2Event.transaction.hash = txHash;
+  piecesAddedV2Event.logIndex = logIndex;
+
+  return piecesAddedV2Event;
+}
+
 export function createNextProvingPeriodEvent(
   setId: BigInt,
   challengeEpoch: BigInt,