Chunk Data Model supports per-chunk service event mapping

engine/common/rpc/convert/execution_results_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
+// TODO: fails with input non-nil ChunkBody.ServiceEventCount
 func TestConvertExecutionResult(t *testing.T) {
 	t.Parallel()
 
@@ -25,6 +26,7 @@ func TestConvertExecutionResult(t *testing.T) {
 	assert.Equal(t, er, converted)
 }
 
+// TODO: fails with input non-nil ChunkBody.ServiceEventCount
 func TestConvertExecutionResults(t *testing.T) {
 	t.Parallel()
 
@@ -43,6 +45,7 @@ func TestConvertExecutionResults(t *testing.T) {
 	assert.Equal(t, results, converted)
 }
 
+// TODO: fails with input non-nil ChunkBody.ServiceEventCount
 func TestConvertExecutionResultMetaList(t *testing.T) {
 	t.Parallel()
 

engine/execution/block_result.go

@@ -49,6 +49,12 @@ func (er *BlockExecutionResult) AllEvents() flow.EventsList {
 	return res
 }
 
+// ServiceEventCountForChunk returns the number of service events emitted in the given chunk.
+func (er *BlockExecutionResult) ServiceEventCountForChunk(chunkIndex int) uint16 {
+	serviceEventCount := len(er.collectionExecutionResults[chunkIndex].serviceEvents)
+	return uint16(serviceEventCount)
+}
+
 func (er *BlockExecutionResult) AllServiceEvents() flow.EventsList {
 	res := make(flow.EventsList, 0)
 	for _, ce := range er.collectionExecutionResults {
@@ -199,6 +205,7 @@ func (ar *BlockAttestationResult) ChunkAt(index int) *flow.Chunk {
 		attestRes.startStateCommit,
 		len(execRes.TransactionResults()),
 		attestRes.eventCommit,
+		ar.ServiceEventCountForChunk(index),
 		attestRes.endStateCommit,
 		execRes.executionSnapshot.TotalComputationUsed(),
 	)

engine/execution/block_result_test.go

@@ -0,0 +1,81 @@
+package execution
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/onflow/flow-go/utils/slices"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// makeBlockExecutionResultFixture makes a BlockExecutionResult fixture
+// with the specified allocation of service events to chunks.
+func makeBlockExecutionResultFixture(serviceEventsPerChunk []int) *BlockExecutionResult {
+	fixture := new(BlockExecutionResult)
+	for _, nServiceEvents := range serviceEventsPerChunk {
+		fixture.collectionExecutionResults = append(fixture.collectionExecutionResults,
+			CollectionExecutionResult{
+				serviceEvents:          unittest.EventsFixture(nServiceEvents),
+				convertedServiceEvents: unittest.ServiceEventsFixture(nServiceEvents),
+			})
+	}
+	return fixture
+}
+
+// Tests that ServiceEventCountForChunk method works as expected under various circumstances:
+func TestBlockExecutionResult_ServiceEventCountForChunk(t *testing.T) {
+	t.Run("no service events", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 1
+		blockResult := makeBlockExecutionResultFixture(make([]int, nChunks))
+		// all chunks should have 0 service event count
+		for chunkIndex := 0; chunkIndex < nChunks; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(0), count)
+		}
+	})
+	t.Run("service events only in system chunk", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 2 // at least 2 chunks
+		// add between 1 and 10 service events, all in the system chunk
+		serviceEventAllocation := make([]int, nChunks)
+		nServiceEvents := rand.Intn(10) + 1
+		serviceEventAllocation[nChunks-1] = nServiceEvents
+
+		blockResult := makeBlockExecutionResultFixture(serviceEventAllocation)
+		// all non-system chunks should have zero service event count
+		for chunkIndex := 0; chunkIndex < nChunks-1; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(0), count)
+		}
+		// the system chunk should contain all service events
+		assert.Equal(t, uint16(nServiceEvents), blockResult.ServiceEventCountForChunk(nChunks-1))
+	})
+	t.Run("service events only outside system chunk", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 2 // at least 2 chunks
+		// add 1 service event to all non-system chunks
+		serviceEventAllocation := slices.Fill(1, nChunks)
+		serviceEventAllocation[nChunks-1] = 0
+
+		blockResult := makeBlockExecutionResultFixture(serviceEventAllocation)
+		// all non-system chunks should have 1 service event
+		for chunkIndex := 0; chunkIndex < nChunks-1; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(1), count)
+		}
+		// the system chunk service event count should include all service events
+		assert.Equal(t, uint16(0), blockResult.ServiceEventCountForChunk(nChunks-1))
+	})
+	t.Run("service events in both system chunk and other chunks", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 2 // at least 2 chunks
+		// add 1 service event to all chunks (including system chunk)
+		serviceEventAllocation := slices.Fill(1, nChunks)
+
+		blockResult := makeBlockExecutionResultFixture(serviceEventAllocation)
+		// all chunks should have service event count of 1
+		for chunkIndex := 0; chunkIndex < nChunks; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(1), count)
+		}
+	})
+}

model/flow/chunk.go

@@ -2,9 +2,11 @@ package flow
 
 import (
 	"fmt"
+	"io"
 	"log"
 
 	"github.com/ipfs/go-cid"
+	"github.com/onflow/go-ethereum/rlp"
 	"github.com/vmihailenco/msgpack/v4"
 )
 
@@ -20,19 +22,112 @@ func init() {
 	}
 }
 
+// ChunkBodyV0 is the prior version of ChunkBody, used for computing backward-compatible IDs and tests.
+// Compared to ChunkBody, ChunkBodyV0 does not have the ServiceEventCount field.
+type ChunkBodyV0 struct {
+	CollectionIndex      uint
+	StartState           StateCommitment
+	EventCollection      Identifier
+	BlockID              Identifier
+	TotalComputationUsed uint64
+	NumberOfTransactions uint64
+}
+
 type ChunkBody struct {
 	CollectionIndex uint
 
 	// execution info
 	StartState      StateCommitment // start state when starting executing this chunk
 	EventCollection Identifier      // Events generated by executing results
-	BlockID         Identifier      // Block id of the execution result this chunk belongs to
+	// ServiceEventCount defines how many service events were emitted in this chunk.
+	// By reading this field in prior chunks in the same ExecutionResult, we can
+	// compute exactly what service events were emitted in this chunk.
+	//
+	// Let C be this chunk, K be the set of chunks in the ExecutionResult containing C.
+	// Then the service event indices for C are given by:
+	//    StartIndex = ∑Ci.ServiceEventCount : Ci ∈ K, Ci.Index < C.Index
+	//    EndIndex = StartIndex + C.ServiceEventCount
+	// The service events for C are given by:
+	//    ExecutionResult.ServiceEvents[StartIndex:EndIndex]
+	//
+	// BACKWARD COMPATIBILITY:
+	// (1) If ServiceEventCount is nil, this indicates that this chunk was created by an older software version
+	//     which did support specifying a mapping between chunks and service events.
+	//     In this case, all service events are assumed to have been emitted in the system chunk (last chunk).
+	//     This was the implicit behaviour prior to the introduction of this field.
+	// (2) Otherwise, ServiceEventCount must be non-nil.
+	// Within an ExecutionResult, all chunks must use either representation (1) or (2), not both.
+	ServiceEventCount *uint16
+	BlockID           Identifier // Block id of the execution result this chunk belongs to
 
 	// Computation consumption info
 	TotalComputationUsed uint64 // total amount of computation used by running all txs in this chunk
 	NumberOfTransactions uint64 // number of transactions inside the collection
 }
 
+// EncodeRLP defines custom encoding logic for the Chunk type.
+// This method exists only so that the embedded ChunkBody's EncodeRLP method is
+// not interpreted as the RLP encoding for the entire Chunk.
+func (ch Chunk) EncodeRLP(w io.Writer) error {
+	return rlp.Encode(w, struct {
+		ChunkBody ChunkBody
+		Index     uint64
+		EndState  StateCommitment
+	}{
+		ChunkBody: ch.ChunkBody,
+		Index:     ch.Index,
+		EndState:  ch.EndState,
+	})
+}
+
+// EncodeRLP defines custom encoding logic for the ChunkBody type.
+// The encoding is defined for backward compatibility with prior data model version (ChunkBodyV0):
+//   - All new ChunkBody instances must have non-nil ServiceEventCount field
+//   - A nil ServiceEventCount field indicates a v0 version of ChunkBody
+//   - when computing the ID of such a ChunkBody, the ServiceEventCount field is omitted from the fingerprint
+func (ch ChunkBody) EncodeRLP(w io.Writer) error {
+	var err error
+	if ch.ServiceEventCount == nil {
+		err = rlp.Encode(w, struct {
+			CollectionIndex      uint
+			StartState           StateCommitment
+			EventCollection      Identifier
+			BlockID              Identifier
+			TotalComputationUsed uint64
+			NumberOfTransactions uint64
+		}{
+			CollectionIndex:      ch.CollectionIndex,
+			StartState:           ch.StartState,
+			EventCollection:      ch.EventCollection,
+			BlockID:              ch.BlockID,
+			TotalComputationUsed: ch.TotalComputationUsed,
+			NumberOfTransactions: ch.NumberOfTransactions,
+		})
+	} else {
+		err = rlp.Encode(w, struct {
+			CollectionIndex      uint
+			StartState           StateCommitment
+			EventCollection      Identifier
+			ServiceEventCount    *uint16
+			BlockID              Identifier
+			TotalComputationUsed uint64
+			NumberOfTransactions uint64
+		}{
+			CollectionIndex:      ch.CollectionIndex,
+			StartState:           ch.StartState,
+			EventCollection:      ch.EventCollection,
+			ServiceEventCount:    ch.ServiceEventCount,
+			BlockID:              ch.BlockID,
+			TotalComputationUsed: ch.TotalComputationUsed,
+			NumberOfTransactions: ch.NumberOfTransactions,
+		})
+	}
+	if err != nil {
+		return fmt.Errorf("failed to rlp encode ChunkBody: %w", err)
+	}
+	return nil
+}
+
 type Chunk struct {
 	ChunkBody
 
@@ -47,6 +142,7 @@ func NewChunk(
 	startState StateCommitment,
 	numberOfTransactions int,
 	eventCollection Identifier,
+	serviceEventCount uint16,
 	endState StateCommitment,
 	totalComputationUsed uint64,
 ) *Chunk {
@@ -57,6 +153,7 @@ func NewChunk(
 			StartState:           startState,
 			NumberOfTransactions: uint64(numberOfTransactions),
 			EventCollection:      eventCollection,
+			ServiceEventCount:    &serviceEventCount,
 			TotalComputationUsed: totalComputationUsed,
 		},
 		Index:    uint64(collectionIndex),