Conversation
frrist
left a comment
frrist
left a comment
There was a problem hiding this comment.
Good stuff!
Main thing here I'd like to discuss more is the Consolidate routine marking records as processed when a failure occurs due to a transient failure while processing.
| // TODO: do more validation here. | ||
| // At the very least that the invocation is a retrieval invocation and the audience is the node |
There was a problem hiding this comment.
Just thinking about storacha/RFC#68 here, I wonder how challenging it would be to perform filtering such that retrievals resulting from replication, or indexing could be excluded. Alternatively this responsibility could be pushed down to piri.
There was a problem hiding this comment.
mmm, I think the billing service is the right place for that business logic, where rules about what we want or don't want to pay for could live.
But yeah, I'm not sure how easy it will be to differentiate them, because they are all retrieval requests in the end. An idea that comes to mind is keeping a white-list of actors that are not charged for egress/do not produce billable egress. Something like "if it was the indexing-service that requested the retrieval, then this receipt doesn't count. And if it was a storage node, it doesn't count either". That would require a registry of nodes in the network, but I think this is something we will need for other use case anyway.
|
|
||
| func (d *DynamoEgressTable) GetUnprocessed(ctx context.Context, limit int) ([]EgressRecord, error) { | ||
| // Scan all shards for the current date for unprocessed records | ||
| today := time.Now().UTC().Format("2006-01-02") |
There was a problem hiding this comment.
anything significant about this date?
There was a problem hiding this comment.
surprisingly enough, that's a standard format string in Go. The way you tell the formatter where you want the year to be in the final string is by using the "2006" or "06" sequences (if you want the long or the short format), "01" is the month, "02" the day and so on. Take a look at the time.Format example in the docs to see how absolutely weird it is.
| today := time.Now().UTC().Format("2006-01-02") | ||
| var allRecords []EgressRecord | ||
|
|
||
| for shard := range 10 { |
There was a problem hiding this comment.
because the AI said so. All of that will change, the current implementation comes from a moment when the spec was still WIP and access patterns were not clear. Please ignore that for now.
| } | ||
| } | ||
|
|
||
| func (s *Server) getReceiptsHandler() http.HandlerFunc { |
There was a problem hiding this comment.
Does https://github.com/storacha/piri/blob/main/pkg/service/egresstracker/service.go#L247 call this? Just curious to make sure I understand the flow of things here.
There was a problem hiding this comment.
yes, that's the endpoint were nodes will be able to fetch consolidate receipts from, exactly in that call you linked
| // Mark records as processed | ||
| if err := c.egressTable.MarkAsProcessed(ctx, records); err != nil { | ||
| return fmt.Errorf("marking records as processed: %w", err) | ||
| } |
There was a problem hiding this comment.
Are all records really processed at the end of this method? e.g. say a Piri node is offline (due to an update or whatever) when fetchReceipts is called, the method fails, and we mark the record as processed anyways. I think we'll need to me more granular here.
There was a problem hiding this comment.
yeah, proper error handling is not implemented yet
|
thanks for the input @frrist. The main goal of this PR is to implement the bare minimum functionality to allow the interaction with the nodes to "work", which means the receipts endpoint and the basic structure is there. Consolidation needs more work, as does validation in the nodes. My aim is to get the basic flow working so that remaining changes are isolated to their own components. So please review this as "MVP, almost WIP" code 🙂 |
|
My approval on this is based on the work being an MVP/WIP. Let's not forget to file an issue for the point raised here: #5 (comment) |
2 participants
Ref: storacha/piri#174
Implement egress record consolidation.
A periodic task:
A new
/receiptsendpoint is offered for nodes to fetchspace/egress/consolidatereceipts so that they can check the results of consolidation against their own records.Known limitations
space/egress/concludeinvocations are supposed to be invoked on a consolidator service as per the spec. The consolidator service has been implemented as a sub-component of the etracker service, so the invocation is issued and executed by the etracker service itself. In this implementation, instead of setting up a new service method and do the invocation, we just create the invocation and generate the receipt when processing the records./receiptsendpoint always returns 404. I'll add a new dynamoDB table to store receipts keyed by the corresponding invocation CID.