fix: weave in lock/unlock semantics to liveness provider

node/consensus/global/consensus_liveness_provider.go

@@ -96,9 +96,31 @@ func (p *GlobalLivenessProvider) Collect(
 			continue
 		}
 
+		err = p.engine.executionManager.Lock(
+			frameNumber,
+			message.Address,
+			message.Payload,
+		)
+		if err != nil {
+			p.engine.logger.Debug(
+				"message failed lock",
+				zap.Int("message_index", i),
+				zap.Error(err),
+			)
+			continue
+		}
+
 		acceptedMessages = append(acceptedMessages, message)
 	}
 
+	err := p.engine.executionManager.Unlock()
+	if err != nil {
+		p.engine.logger.Error(
+			"unable to unlock",
+			zap.Error(err),
+		)
+	}
+
 	commitments := make([]*tries.VectorCommitmentTree, 256)
 	for i := range 256 {
 		commitments[i] = &tries.VectorCommitmentTree{}

node/execution/engines/compute_execution_engine.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"math/big"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -285,21 +286,48 @@ func (e *ComputeExecutionEngine) Lock(
 		return nil
 	}
 
+	if len(message) > 4 &&
+		binary.BigEndian.Uint32(message[:4]) == protobufs.MessageBundleType {
+		bundle := &protobufs.MessageBundle{}
+		err = bundle.FromCanonicalBytes(message)
+		if err != nil {
+			return errors.Wrap(err, "lock")
+		}
+
+		for _, r := range bundle.Requests {
+			req, err := r.ToCanonicalBytes()
+			if err != nil {
+				return errors.Wrap(err, "lock")
+			}
+
+			if err = intrinsic.Lock(frameNumber, req[8:]); err != nil {
+				return err
+			}
+		}
+	}
+
 	return errors.Wrap(intrinsic.Lock(frameNumber, message), "lock")
 }
 
-func (e *ComputeExecutionEngine) Unlock(
-	frameNumber uint64,
-	address []byte,
-	message []byte,
-) error {
-	intrinsic, err := e.tryGetIntrinsic(address)
-	if err != nil {
-		// non-applicable
-		return nil
+func (e *ComputeExecutionEngine) Unlock() error {
+	e.intrinsicsMutex.RLock()
+	errs := []string{}
+	for _, intrinsic := range e.intrinsics {
+		err := intrinsic.Unlock()
+		if err != nil {
+			errs = append(errs, err.Error())
+		}
+	}
+	e.intrinsicsMutex.RUnlock()
+
+	if len(errs) != 0 {
+		return errors.Wrap(
+			errors.Errorf("multiple errors: %s", strings.Join(errs, ", ")),
+			"unlock",
+		)
 	}
 
-	return errors.Wrap(intrinsic.Unlock(), "unlock")
+	return nil
 }
 
 func (e *ComputeExecutionEngine) GetCost(message []byte) (*big.Int, error) {

node/execution/engines/global_execution_engine.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"math/big"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -474,21 +475,48 @@ func (e *GlobalExecutionEngine) Lock(
 		return nil
 	}
 
-	return errors.Wrap(intrinsic.Lock(frameNumber, message), "lock")
-}
+	if len(message) > 4 &&
+		binary.BigEndian.Uint32(message[:4]) == protobufs.MessageBundleType {
+		bundle := &protobufs.MessageBundle{}
+		err = bundle.FromCanonicalBytes(message)
+		if err != nil {
+			return errors.Wrap(err, "lock")
+		}
 
-func (e *GlobalExecutionEngine) Unlock(
-	frameNumber uint64,
-	address []byte,
-	message []byte,
-) error {
-	intrinsic, err := e.tryGetIntrinsic(address)
-	if err != nil {
-		// non-applicable
-		return nil
+		for _, r := range bundle.Requests {
+			req, err := r.ToCanonicalBytes()
+			if err != nil {
+				return errors.Wrap(err, "lock")
+			}
+
+			if err = intrinsic.Lock(frameNumber, req[8:]); err != nil {
+				return err
+			}
+		}
 	}
 
-	return errors.Wrap(intrinsic.Unlock(), "unlock")
+	return intrinsic.Lock(frameNumber, message)
+}
+
+func (e *GlobalExecutionEngine) Unlock() error {
+	e.intrinsicsMutex.RLock()
+	errs := []string{}
+	for _, intrinsic := range e.intrinsics {
+		err := intrinsic.Unlock()
+		if err != nil {
+			errs = append(errs, err.Error())
+		}
+	}
+	e.intrinsicsMutex.RUnlock()
+
+	if len(errs) != 0 {
+		return errors.Wrap(
+			errors.Errorf("multiple errors: %s", strings.Join(errs, ", ")),
+			"unlock",
+		)
+	}
+
+	return nil
 }
 
 func (e *GlobalExecutionEngine) tryGetIntrinsic(address []byte) (

node/execution/engines/global_execution_engine_test.go

@@ -110,7 +110,7 @@ func TestGlobalExecutionEngine_ProcessMessage(t *testing.T) {
 			},
 			address:     []byte("invalid_address"),
 			wantErr:     true,
-			errContains: "invalid address for global execution engine",
+			errContains: "invalid shard",
 		},
 	}
 

node/execution/engines/hypergraph_execution_engine.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"math/big"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -477,21 +478,48 @@ func (e *HypergraphExecutionEngine) Lock(
 		return nil
 	}
 
+	if len(message) > 4 &&
+		binary.BigEndian.Uint32(message[:4]) == protobufs.MessageBundleType {
+		bundle := &protobufs.MessageBundle{}
+		err = bundle.FromCanonicalBytes(message)
+		if err != nil {
+			return errors.Wrap(err, "lock")
+		}
+
+		for _, r := range bundle.Requests {
+			req, err := r.ToCanonicalBytes()
+			if err != nil {
+				return errors.Wrap(err, "lock")
+			}
+
+			if err = intrinsic.Lock(frameNumber, req[8:]); err != nil {
+				return err
+			}
+		}
+	}
+
 	return errors.Wrap(intrinsic.Lock(frameNumber, message), "lock")
 }
 
-func (e *HypergraphExecutionEngine) Unlock(
-	frameNumber uint64,
-	address []byte,
-	message []byte,
-) error {
-	intrinsic, err := e.tryGetIntrinsic(address)
-	if err != nil {
-		// non-applicable
-		return nil
+func (e *HypergraphExecutionEngine) Unlock() error {
+	e.intrinsicsMutex.RLock()
+	errs := []string{}
+	for _, intrinsic := range e.intrinsics {
+		err := intrinsic.Unlock()
+		if err != nil {
+			errs = append(errs, err.Error())
+		}
+	}
+	e.intrinsicsMutex.RUnlock()
+
+	if len(errs) != 0 {
+		return errors.Wrap(
+			errors.Errorf("multiple errors: %s", strings.Join(errs, ", ")),
+			"unlock",
+		)
 	}
 
-	return errors.Wrap(intrinsic.Unlock(), "unlock")
+	return nil
 }
 
 func (e *HypergraphExecutionEngine) handleBundle(

node/execution/engines/token_execution_engine.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"math/big"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -537,21 +538,48 @@ func (e *TokenExecutionEngine) Lock(
 		return nil
 	}
 
+	if len(message) > 4 &&
+		binary.BigEndian.Uint32(message[:4]) == protobufs.MessageBundleType {
+		bundle := &protobufs.MessageBundle{}
+		err = bundle.FromCanonicalBytes(message)
+		if err != nil {
+			return errors.Wrap(err, "lock")
+		}
+
+		for _, r := range bundle.Requests {
+			req, err := r.ToCanonicalBytes()
+			if err != nil {
+				return errors.Wrap(err, "lock")
+			}
+
+			if err = intrinsic.Lock(frameNumber, req[8:]); err != nil {
+				return err
+			}
+		}
+	}
+
 	return errors.Wrap(intrinsic.Lock(frameNumber, message), "lock")
 }
 
-func (e *TokenExecutionEngine) Unlock(
-	frameNumber uint64,
-	address []byte,
-	message []byte,
-) error {
-	intrinsic, err := e.tryGetIntrinsic(address)
-	if err != nil {
-		// non-applicable
-		return nil
+func (e *TokenExecutionEngine) Unlock() error {
+	e.intrinsicsMutex.RLock()
+	errs := []string{}
+	for _, intrinsic := range e.intrinsics {
+		err := intrinsic.Unlock()
+		if err != nil {
+			errs = append(errs, err.Error())
+		}
+	}
+	e.intrinsicsMutex.RUnlock()
+
+	if len(errs) != 0 {
+		return errors.Wrap(
+			errors.Errorf("multiple errors: %s", strings.Join(errs, ", ")),
+			"unlock",
+		)
 	}
 
-	return errors.Wrap(intrinsic.Unlock(), "unlock")
+	return nil
 }
 
 func (e *TokenExecutionEngine) handleBundle(
@@ -713,7 +741,7 @@ func (e *TokenExecutionEngine) processIndividualMessage(
 	}
 
 	// Otherwise, try to handle it as an operation on existing intrinsic
-	intrinsic, err := e.tryGetIntrinsic(address)
+	intrinsic, err := e.tryGetIntrinsic(domain)
 	if err != nil {
 		return nil, errors.Wrap(err, "process individual message")
 	}

node/execution/manager/execution_manager.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"math/big"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -638,6 +639,28 @@ func (m *ExecutionEngineManager) Lock(
 	return engine.Lock(frameNumber, address, message)
 }
 
+func (m *ExecutionEngineManager) Unlock() error {
+	m.enginesMu.RLock()
+	defer m.enginesMu.RUnlock()
+
+	errs := []string{}
+	for _, engine := range m.engines {
+		err := engine.Unlock()
+		if err != nil {
+			errs = append(errs, err.Error())
+		}
+	}
+
+	if len(errs) != 0 {
+		return errors.Wrap(
+			errors.Errorf("multiple errors: %s", strings.Join(errs, ", ")),
+			"unlock",
+		)
+	}
+
+	return nil
+}
+
 // ValidateMessage validates a message without materializing state changes
 func (m *ExecutionEngineManager) ValidateMessage(
 	frameNumber uint64,

types/execution/execution_engine.go

@@ -30,7 +30,7 @@ type ShardExecutionEngine interface {
 		message []byte,
 	) (*protobufs.MessageRequest, error)
 	Lock(frameNumber uint64, address []byte, message []byte) error
-	Unlock(frameNumber uint64, address []byte, message []byte) error
+	Unlock() error
 	GetCost(message []byte) (*big.Int, error)
 	GetCapabilities() []*protobufs.Capability
 }