Skip to content

framework: guard executor shared state under stateMu #24

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nadahalli wants to merge 1 commit into
base: main
Choose a base branch
from
Open

framework: guard executor shared state under stateMu #24

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 55 additions & 17 deletions capabilities/framework/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -463,12 +463,18 @@ type RealExecutor struct {
capConfigMu sync.RWMutex
capConfig *ParsedConfig
capabilityRegistry core.CapabilitiesRegistry
nodeID string
proposalInFlight atomic.Bool
donMu sync.RWMutex
donMembers [][]byte
donF uint32
enclaves []types.Enclave

// stateMu guards the mutable shared state that the background refresh loop
// (EnsureFreshEnclaves) and getLocalNodeAndCapConfig rewrite while Execute
// reads it concurrently. Access these only through the getter/setter helpers.
// vaultDON.CryptographyThreshold is guarded too (via getVaultThreshold/
// setVaultThreshold); the rest of vaultDON is publish-once at init.
stateMu sync.RWMutex
nodeID string
donMembers [][]byte
donF uint32
enclaves []types.Enclave

// Background enclave-refresh ticker, started once on init and stopped on Close.
refreshCancel context.CancelFunc
Expand Down Expand Up @@ -627,7 +633,7 @@ func (e *RealExecutor) Execute(ctx context.Context, protoBytes []byte, secrets [
// and org, matching the DON-mode base labels (workflowName/workflowOwner/orgID).
// sdk is not available at this layer (not on RequestMetadata), so it is omitted.
metrics := NewScopedEmitter(e.metrics, map[string]any{
"node.id": e.nodeID,
"node.id": e.getNodeID(),
"workflow.owner": metadata.WorkflowOwner,
"workflow.id": metadata.WorkflowID,
"workflow.name": metadata.WorkflowName,
Expand Down Expand Up @@ -1011,7 +1017,7 @@ func (e *RealExecutor) initLazily(ctx context.Context) error {
}

e.enclaveClient = pool
e.enclaves = nodes
e.setEnclaves(nodes)
e.rateLimiter = rateLimiter
e.setCapConfig(parsedConfig)
e.vaultDON = VaultDON{
Expand All @@ -1024,11 +1030,11 @@ func (e *RealExecutor) initLazily(ctx context.Context) error {
e.startEnclaveRefreshLoop()
e.lggr.Infow("executor initialized",
"capabilityID", e.capabilityID,
"nodeID", e.nodeID,
"nodeID", e.getNodeID(),
"maxRetries", parsedConfig.MaxRetries,
"retryBackoffSeconds", parsedConfig.RetryBackoffSeconds,
"enableSecretsCache", parsedConfig.EnableSecretsCache,
"vaultDONThreshold", e.vaultDON.CryptographyThreshold,
"vaultDONThreshold", e.getVaultThreshold(),
"insecureSkipTLS", parsedConfig.InsecureSkipTLSVerify)
return nil
}
Expand Down Expand Up @@ -1094,14 +1100,14 @@ func (e *RealExecutor) EnsureFreshEnclaves(ctx context.Context) error {
"endpoint": "publicKeys",
})
} else {
e.enclaves = nodes
e.setEnclaves(nodes)
}

vaultDONPossibleFaultyNodes, err := getVaultDONPossibleFaultyNodes(ctx, e.vaultDON.Capability, int(localNode.WorkflowDON.F))
if err != nil {
return fmt.Errorf("failed to get VaultDON possible faulty nodes: %w", err)
}
e.vaultDON.CryptographyThreshold = getVaultDONThreshold(vaultDONPossibleFaultyNodes)
e.setVaultThreshold(getVaultDONThreshold(vaultDONPossibleFaultyNodes))

newMembers := peerIDsToSortedBytes(localNode.WorkflowDON.Members)
newF := uint32(localNode.WorkflowDON.F)
Expand All @@ -1111,18 +1117,48 @@ func (e *RealExecutor) EnsureFreshEnclaves(ctx context.Context) error {
}

func (e *RealExecutor) getDONMembership() (members [][]byte, f uint32) {
e.donMu.RLock()
defer e.donMu.RUnlock()
e.stateMu.RLock()
defer e.stateMu.RUnlock()
return e.donMembers, e.donF
}

func (e *RealExecutor) setDONMembership(members [][]byte, f uint32) {
e.donMu.Lock()
defer e.donMu.Unlock()
e.stateMu.Lock()
defer e.stateMu.Unlock()
e.donMembers = members
e.donF = f
}

func (e *RealExecutor) getNodeID() string {
e.stateMu.RLock()
defer e.stateMu.RUnlock()
return e.nodeID
}

func (e *RealExecutor) setNodeID(id string) {
e.stateMu.Lock()
defer e.stateMu.Unlock()
e.nodeID = id
}

func (e *RealExecutor) setEnclaves(nodes []types.Enclave) {
e.stateMu.Lock()
defer e.stateMu.Unlock()
e.enclaves = nodes
}

func (e *RealExecutor) getVaultThreshold() int {
e.stateMu.RLock()
defer e.stateMu.RUnlock()
return e.vaultDON.CryptographyThreshold
}

func (e *RealExecutor) setVaultThreshold(t int) {
e.stateMu.Lock()
defer e.stateMu.Unlock()
e.vaultDON.CryptographyThreshold = t
}

func (e *RealExecutor) proposeConfigUpdateIfMembershipChanged(ctx context.Context, newMembers [][]byte, newF uint32) {
curMembers, curF := e.getDONMembership()
// With a recorded baseline, skip the enclave round-trip while membership is unchanged.
Expand Down Expand Up @@ -1215,6 +1251,8 @@ func (e *RealExecutor) broadcastConfigUpdate(ctx context.Context, newMembers [][
}

func (e *RealExecutor) GetEnclaves() []types.Enclave {
e.stateMu.RLock()
defer e.stateMu.RUnlock()
return e.enclaves
}

Expand All @@ -1233,7 +1271,7 @@ func (e *RealExecutor) getLocalNodeAndCapConfig(ctx context.Context) (capabiliti
return capabilities.Node{}, capabilities.CapabilityConfiguration{}, fmt.Errorf("local node does not have a WorkflowDON ID, cannot initialise confidential http action capability")
}
if localNode.PeerID != nil {
e.nodeID = localNode.PeerID.String()
e.setNodeID(localNode.PeerID.String())
}

ownCapabilityConfig, err := e.capabilityRegistry.ConfigForCapability(ctx, e.capabilityID, localNode.WorkflowDON.ID)
Expand Down Expand Up @@ -1602,7 +1640,7 @@ func (e *RealExecutor) GetEncryptedDecryptionShares(
} else {
return nil, nil, fmt.Errorf("no decryption shares found for secret %s: neither binary nor hex-encoded shares present", secretResp.GetId().GetKey())
}
minimumSharesRequired := e.vaultDON.CryptographyThreshold
minimumSharesRequired := e.getVaultThreshold()
if len(encryptedDecryptionSharesForSecret) < minimumSharesRequired {
return nil, nil, fmt.Errorf("not enough encrypted decryption key shares for secret %s, expected at least %d, got %d", secretResp.GetId().GetKey(), minimumSharesRequired, len(encryptedDecryptionSharesForSecret))
}
Expand Down
115 changes: 115 additions & 0 deletions capabilities/framework/executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ import (
"errors"
"fmt"
"slices"
"sync"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -299,6 +301,9 @@ func getMockCapabilitiesRegistry(t *testing.T, mockVaultDON framework.VaultDON)

// MockMetrics is a stub implementation of types.Emitter for testing.
type MockMetrics struct {
// mu guards the maps so Emit is safe under concurrent Execute calls.
mu sync.Mutex

// CallCounts tracks how many times each event was emitted
CallCounts map[string]int

Expand All @@ -316,6 +321,8 @@ func NewMockMetrics() *MockMetrics {

// Emit implements the types.Emitter interface.
func (m *MockMetrics) Emit(event string, details map[string]any) {
m.mu.Lock()
defer m.mu.Unlock()
m.CallCounts[event] += 1
m.EmitRecords[event] = append(m.EmitRecords[event], details)
}
Expand Down Expand Up @@ -1635,6 +1642,114 @@ func TestExecutor_ExecuteWithSecretsCache(t *testing.T) {
})
}

// TestExecutor_ConcurrentRefreshAndExecute_NoDataRace is a regression test for the
// executor data race: the background refresh loop (EnsureFreshEnclaves) and
// getLocalNodeAndCapConfig rewrite nodeID, enclaves, and vaultDON.CryptographyThreshold
// while Execute reads them concurrently. Run under -race; before the stateMu guards
// this trips the detector, and a torn read of the nodeID string / enclaves slice
// header can corrupt memory at runtime.
func TestExecutor_ConcurrentRefreshAndExecute_NoDataRace(t *testing.T) {
mockVaultDONCapability := &MockVaultDONCapability{}
mockVaultDONCapability.ExecuteFunc = func(ctx context.Context, req capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) {
respAny, _ := anypb.New(getValidGetSecretsResponse())
return capabilities.CapabilityResponse{Payload: respAny}, nil
}
mockVaultDON := framework.VaultDON{CryptographyThreshold: 1, Capability: mockVaultDONCapability}

mockEnclaveClient := &MockEnclaveClient{}
mockEnclaveClient.ExecuteBatchFunc = func(ctx context.Context, reqs []enclavetypes.SignedComputeRequest, enclaveIDs [][32]byte) ([]enclavetypes.ExecuteResponse, error) {
return mockEnclaveClient.commonExecuteBatchReturn(t)
}

// The registry flips the local PeerID (source of nodeID) and the enclave config
// (source of e.enclaves) on every call, so the guarded fields actually change value
// under concurrency. WorkflowDON membership and F stay constant so
// validateEnclaveSigners keeps passing and Execute stays on the happy path.
var flip atomic.Uint64
mockRegistry := &MockCapabilitiesRegistry{
LocalNodeFunc: func(ctx context.Context) (capabilities.Node, error) {
pid := mockPeerID1
if flip.Add(1)%2 == 0 {
pid = mockPeerID2
}
return capabilities.Node{
WorkflowDON: capabilities.DON{ID: 1, F: 0, Members: []p2ptypes.PeerID{mockPeerID1, mockPeerID2}},
PeerID: &pid,
}, nil
},
ConfigForCapabilityFunc: func(ctx context.Context, capabilityID string, donID uint32) (capabilities.CapabilityConfiguration, error) {
id, url := [32]byte{1}, "https://enclave-a.example.com"
if flip.Load()%2 == 0 {
id, url = [32]byte{2}, "https://enclave-b.example.com"
}
enclavesList := enclavetypes.EnclavesList{
Enclaves: []enclavetypes.Enclave{{
EnclaveID: id, EnclaveURL: url, EnclaveType: "nitro",
TrustedValues: [][]byte{[]byte("{}")}, Region: "us-west-2",
}},
}
wrappedConfig, err := values.WrapMap(enclavesList)
require.NoError(t, err)
return capabilities.CapabilityConfiguration{DefaultConfig: wrappedConfig}, nil
},
GetExecutableFunc: func(ctx context.Context, ID string) (capabilities.ExecutableCapability, error) {
if ID == vault.CapabilityID {
return mockVaultDONCapability, nil
}
return nil, fmt.Errorf("unknown capability ID: %s", ID)
},
}

executor := framework.NewTestExecutor(
logger.Test(t), getMockKeystore(), mockEnclaveClient, mockVaultDON,
NewMockMetrics(), getDefaultRateLimiter(),
1, 0, "test-capability-id", false, TEST_NODE_ID, mockRegistry,
)

actionInput := getTestExecutorInput()
protoBytes, err := proto.Marshal(actionInput.GetInput())
require.NoError(t, err)

const goroutines = 8
const iterations = 40
var wg sync.WaitGroup
start := make(chan struct{})

worker := func(fn func(i int)) {
defer wg.Done()
<-start
for i := 0; i < iterations; i++ {
fn(i)
}
}

for g := 0; g < goroutines; g++ {
g := g
wg.Add(3)
// Reader: Execute reads nodeID (scoped emitter) and the vault threshold.
go worker(func(i int) {
md := capabilities.RequestMetadata{
WorkflowID: WORKFLOW_ID,
WorkflowExecutionID: fmt.Sprintf("exec-%d-%d", g, i),
WorkflowName: WORKFLOW_NAME,
WorkflowOwner: WORKFLOW_OWNER,
}
_, _ = executor.Execute(context.Background(), protoBytes, actionInput.GetVaultDonSecrets(), md)
})
// Writer: refresh rewrites nodeID, enclaves, and the vault threshold.
go worker(func(i int) {
_ = executor.EnsureFreshEnclaves(context.Background())
})
// Reader: GetEnclaves reads the enclaves slice header.
go worker(func(i int) {
_ = executor.GetEnclaves()
})
}

close(start)
wg.Wait()
}

func TestEnsureFreshEnclaves_ConfigurationChange(t *testing.T) {
t.Run("UpdateNodes is called with new enclaves when registry configuration changes", func(t *testing.T) {
// Track what nodes are passed to UpdateNodes
Expand Down
Loading