Aggregation of Accepted Pieces to Roots on Chain

[frrist]

Code Path Trace

1. accept.go:82 → s.PDP().Aggregator().AggregatePiece(ctx, pdpPiece)
2. LocalAggregator.AggregatePiece (local.go:66) → Enqueues to pieceQueue with task PieceAggregateTask
3. PieceAggregateTask handler (provider.go:128) → PieceAggregator.AggregatePieces()
4. PieceAggregator.AggregatePieces (steps.go:92-114) → Aggregates pieces in buffer, enqueues to linkQueue with task PieceSubmitTask
5. PieceSubmitTask handler (provider.go:119) → Manager.Submit()
6. Manager.Submit (aggregate_manager.go:160) → Buffers aggregates, triggers doSubmit() when full
7. Manager.doSubmit (aggregate_manager.go:301) → Enqueues to managerQueue with task ManagerTaskName
8. ManagerTaskName handler (aggregate_manager.go:234) → AddRootsTaskHandler.Handle()
9. AddRootsTaskHandler.Handle (aggregate_manager.go:349) → aggregate_manager.go:384 → api.AddRoots()

Mermaid Sequence Diagram

  sequenceDiagram
      participant Accept as blob/Accept Handler
      participant Aggregator as LocalAggregator
      participant PieceQ as Piece Queue
      participant PieceAgg as PieceAggregator
      participant Buffer as InProgress Workspace
      participant LinkQ as Link Queue
      participant Manager as Submission Manager
      participant MgrBuffer as Manager Buffer
      participant MgrQ as Manager Queue
      participant Handler as AddRoots Handler
      participant API as ProofSet API

      Note over Accept,API: State: PIECE_RECEIVED
      Accept->>Aggregator: AggregatePiece(piece)
      Aggregator->>PieceQ: Enqueue(PieceAggregateTask)
      Note over PieceQ: State: PIECE_QUEUED

      PieceQ-->>PieceAgg: Execute Task
      Note over PieceAgg: State: AGGREGATING
      PieceAgg->>Buffer: GetBuffer()
      Buffer-->>PieceAgg: Current buffer
      PieceAgg->>PieceAgg: AggregatePieces()<br/>(BufferingAggregator)

      alt Aggregate Complete (>= 128MB)
          Note over PieceAgg: State: AGGREGATE_CREATED
          PieceAgg->>Buffer: PutBuffer(updated)
          PieceAgg->>PieceAgg: Store aggregate
          PieceAgg->>LinkQ: Enqueue(PieceSubmitTask)
          Note over LinkQ: State: AGGREGATE_QUEUED

          LinkQ-->>Manager: Execute Task: Submit(aggregateLink)
          Note over Manager: State: BUFFERING_SUBMISSION
          Manager->>MgrBuffer: Get buffer.Aggregates()
          MgrBuffer-->>Manager: Current aggregates

          alt Buffer would overflow (> MaxBatchSize)
              Manager->>Manager: doSubmit(current buffer)
              Manager->>MgrQ: Enqueue(ManagerTaskName, aggregates)
              Manager->>MgrBuffer: ClearAggregates()
              Note over MgrQ: State: ROOTS_QUEUED
              Manager->>MgrBuffer: AppendAggregates(new items)
          else Buffer has space
              Manager->>MgrBuffer: AppendAggregates(new items)
          end

          Note over Manager: Also: processLoop polls every 30s
          Manager->>MgrBuffer: Get Aggregates() [periodic]
          alt Buffer not empty
              Manager->>Manager: doSubmit(buffer)
              Manager->>MgrQ: Enqueue(ManagerTaskName)
              Manager->>MgrBuffer: ClearAggregates()
          end

          MgrQ-->>Handler: Execute Task: Handle(links)
          Note over Handler: State: SUBMITTING_ROOTS
          Handler->>Handler: Get ProofSetID
          loop For each aggregate link
              Handler->>Handler: Get aggregate from store
              Handler->>Handler: Build RootAdd (root + subroots)
          end
          Handler->>API: AddRoots(proofSetID, roots) [LINE 384]
          Note over API: State: ADDING_ROOTS
          API-->>Handler: txHash
          Note over Handler: State: ROOTS_ADDED

          Handler-->>LinkQ: Task complete
          LinkQ->>LinkQ: Enqueue(PieceAcceptTask)
      else Continue buffering
          Note over PieceAgg: State: AGGREGATING
          PieceAgg->>Buffer: PutBuffer(updated)
      end

Loading

State Machine

The system can be modeled as a state machine with the following states:

  stateDiagram-v2
      [*] --> PIECE_RECEIVED: blob/accept invoked
      PIECE_RECEIVED --> PIECE_QUEUED: Enqueue to PieceQueue
      PIECE_QUEUED --> AGGREGATING: Worker picks up task

      AGGREGATING --> AGGREGATING: Buffer < 128MB
      AGGREGATING --> AGGREGATE_CREATED: Buffer >= 128MB

      AGGREGATE_CREATED --> AGGREGATE_QUEUED: Enqueue to LinkQueue
      AGGREGATE_QUEUED --> BUFFERING_SUBMISSION: Worker picks up task

      BUFFERING_SUBMISSION --> BUFFERING_SUBMISSION: Manager buffer has space
      BUFFERING_SUBMISSION --> ROOTS_QUEUED: Buffer full OR poll interval

      ROOTS_QUEUED --> SUBMITTING_ROOTS: Worker picks up task
      SUBMITTING_ROOTS --> ADDING_ROOTS: API call to AddRoots
      ADDING_ROOTS --> ROOTS_ADDED: Transaction confirmed

      ROOTS_ADDED --> [*]

      note right of AGGREGATING
          Buffering pieces until
          total >= MinAggregateSize (128MB)
      end note

      note right of BUFFERING_SUBMISSION
          Manager buffers up to
          MaxBatchSize (10) aggregates
          or polls every 30 seconds
      end note

      note right of ADDING_ROOTS
          Blockchain transaction
          at aggregate_manager.go:384
      end note

Loading

Key Observations

1. Three-tier buffering system: Pieces → Aggregates → Batch of roots
2. Four async queues: PieceQueue, LinkQueue (x2 tasks), ManagerQueue
3. Two buffer thresholds:
   - Aggregation: 128MB minimum per aggregate
   - Submission: 10 aggregates maximum per batch (DefaultMaxBatchSizeBytes)
4. Dual submission triggers: Buffer overflow OR 30-second poll interval
5. Serial submission: Manager queue has MaxWorkers=1 to keep AddRoots calls serial

[Peeja]

I'm a little confused about the state machine. The service has a single, global state? Isn't the service always AGGREGATING? I would expect each aggregate to have a state, starting with accepting new pieces, and closing off at some point, causing blob/accept to make a new aggregate to accept new pieces while shipping the closed one on to the next stage.