Skip to content

fix: resolve input data shows as undefined in simulateTask problem #333

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

Merged
merged 7 commits into from
Jun 18, 2025
+2,035 −487
Merged

fix: resolve input data shows as undefined in simulateTask problem #333

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
fdb2376
Fixed runTask returns undefined for eventTrigger problem (#331)
chrisli30 Jun 17, 2025
ca26272
Apply node.input to the Step of Execution and simulateTask (#332)
chrisli30 Jun 17, 2025
08b6300
fix: resolve input data shows as undefined in simulateTask problem
chrisli30 Jun 17, 2025
8980e08
Address the dual-access mapping logic for processing input variables …
chrisli30 Jun 17, 2025
44c39cc
Extracted common functions out in vm_runner
chrisli30 Jun 17, 2025
08666a0
fix: udpate operator.go to solve contract not found startup issue
chrisli30 Jun 18, 2025
d1fd5d2
Set correct console.warn level and reuse getMapVar code
chrisli30 Jun 18, 2025
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
342 changes: 284 additions & 58 deletions core/taskengine/engine.go
View file
Open in desktop

Large diffs are not rendered by default.

60 changes: 30 additions & 30 deletions core/taskengine/engine_trigger_output_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,26 +239,26 @@ func TestBuildTriggerDataMapFromProtobufEventTriggerComprehensive(t *testing.T)
description: "Should map all TransferLog fields including the critical log_index field",
verifyFunc: func(t *testing.T, result map[string]interface{}) {
expected := map[string]interface{}{
"token_name": "Test Token",
"token_symbol": "TEST",
"token_decimals": uint32(18),
"transaction_hash": "0x1234567890abcdef",
"address": "0xabcdef1234567890",
"block_number": uint64(12345678),
"block_timestamp": uint64(1672531200),
"from_address": "0x1111111111111111",
"to_address": "0x2222222222222222",
"value": "1000000000000000000",
"value_formatted": "1.0",
"transaction_index": uint32(5),
"log_index": uint32(3),
"type": "TRIGGER_TYPE_EVENT",
"tokenName": "Test Token",
"tokenSymbol": "TEST",
"tokenDecimals": uint32(18),
"transactionHash": "0x1234567890abcdef",
"address": "0xabcdef1234567890",
"blockNumber": uint64(12345678),
"blockTimestamp": uint64(1672531200),
"fromAddress": "0x1111111111111111",
"toAddress": "0x2222222222222222",
"value": "1000000000000000000",
"valueFormatted": "1.0",
"transactionIndex": uint32(5),
"logIndex": uint32(3),
"type": "TRIGGER_TYPE_EVENT",
}

require.Equal(t, expected, result, "All TransferLog fields should be properly mapped")

require.Contains(t, result, "log_index", "log_index field should be present")
require.Equal(t, uint32(3), result["log_index"], "log_index should have correct value")
require.Contains(t, result, "logIndex", "logIndex field should be present")
require.Equal(t, uint32(3), result["logIndex"], "logIndex should have correct value")
},
},
{
Expand All @@ -281,29 +281,29 @@ func TestBuildTriggerDataMapFromProtobufEventTriggerComprehensive(t *testing.T)
description: "Should map all EvmLog fields including log_index",
verifyFunc: func(t *testing.T, result map[string]interface{}) {
expected := map[string]interface{}{
"block_number": uint64(12345678),
"log_index": uint32(3),
"tx_hash": "0x1234567890abcdef",
"address": "0xabcdef1234567890",
"topics": []string{"0xtopic1", "0xtopic2", "0xtopic3"},
"data": "0xdeadbeef",
"block_hash": "0xblockhash123456",
"transaction_index": uint32(5),
"removed": false,
"type": "TRIGGER_TYPE_EVENT",
"blockNumber": uint64(12345678),
"logIndex": uint32(3),
"transactionHash": "0x1234567890abcdef",
"address": "0xabcdef1234567890",
"topics": []string{"0xtopic1", "0xtopic2", "0xtopic3"},
"data": "0xdeadbeef",
"blockHash": "0xblockhash123456",
"transactionIndex": uint32(5),
"removed": false,
"type": "TRIGGER_TYPE_EVENT",
}

require.Equal(t, expected, result, "All EvmLog fields should be properly mapped")

require.Contains(t, result, "log_index", "log_index field should be present")
require.Equal(t, uint32(3), result["log_index"], "log_index should have correct value")
require.Contains(t, result, "logIndex", "logIndex field should be present")
require.Equal(t, uint32(3), result["logIndex"], "logIndex should have correct value")
},
},
}

for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
result := buildTriggerDataMapFromProtobuf(avsproto.TriggerType_TRIGGER_TYPE_EVENT, test.input)
result := buildTriggerDataMapFromProtobuf(avsproto.TriggerType_TRIGGER_TYPE_EVENT, test.input, nil)

require.NotNil(t, result, "buildTriggerDataMapFromProtobuf should never return nil")

Expand All @@ -325,7 +325,7 @@ func TestBuildTriggerDataMapFromProtobufFieldCompleteness(t *testing.T) {
}

for _, triggerType := range triggerTypes {
result := buildTriggerDataMapFromProtobuf(triggerType, nil)
result := buildTriggerDataMapFromProtobuf(triggerType, nil, nil)
require.Contains(t, result, "type", "All trigger types should include type field")
require.Equal(t, triggerType.String(), result["type"], "Type field should match trigger type")
}
Expand Down
36 changes: 35 additions & 1 deletion core/taskengine/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"time"

"google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/types/known/structpb"

"github.com/AvaProtocol/EigenLayer-AVS/model"
avsproto "github.com/AvaProtocol/EigenLayer-AVS/protobuf"
Expand Down Expand Up @@ -160,6 +161,28 @@ func (x *TaskExecutor) RunTask(task *model.Task, queueData *QueueExecutionData)
vm.WithLogger(x.logger).WithDb(x.db)
initialTaskStatus := task.Status

// Extract and add trigger input data if available
triggerInputData := ExtractTriggerInputData(task.Trigger)
if triggerInputData != nil {
// Get the trigger variable name and add input data
triggerVarName := sanitizeTriggerNameForJS(task.Trigger.GetName())
vm.mu.Lock()
existingTriggerVar := vm.vars[triggerVarName]
if existingMap, ok := existingTriggerVar.(map[string]any); ok {
// Apply dual-access mapping to trigger input data
processedTriggerInput := ProcessInputVariableWithDualAccess(triggerInputData)
existingMap["input"] = processedTriggerInput
vm.vars[triggerVarName] = existingMap
} else {
// Create new trigger variable with input data
processedTriggerInput := ProcessInputVariableWithDualAccess(triggerInputData)
vm.vars[triggerVarName] = map[string]any{
"input": processedTriggerInput,
}
}
vm.mu.Unlock()
}

if err != nil {
return nil, fmt.Errorf("vm failed to initialize: %w", err)
}
Expand Down Expand Up @@ -189,6 +212,16 @@ func (x *TaskExecutor) RunTask(task *model.Task, queueData *QueueExecutionData)
// This ensures regular workflows have complete execution history (trigger + nodes)

// Create trigger step similar to SimulateTask
// Extract trigger input data using the proper extraction function
triggerInputData := ExtractTriggerInputData(task.Trigger)
var triggerInputProto *structpb.Value
if triggerInputData != nil {
triggerInputProto, err = structpb.NewValue(triggerInputData)
Comment on lines +216 to +219
Copy link
Preview

Copilot AI Jun 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

triggerInputData is redeclared later when building the trigger step, shadowing the earlier variable—rename one of them (e.g., triggerInputForStep) to avoid confusion.

Suggested change
triggerInputData := ExtractTriggerInputData(task.Trigger)
var triggerInputProto *structpb.Value
if triggerInputData != nil {
triggerInputProto, err = structpb.NewValue(triggerInputData)
triggerInputForStep := ExtractTriggerInputData(task.Trigger)
var triggerInputProto *structpb.Value
if triggerInputForStep != nil {
triggerInputProto, err = structpb.NewValue(triggerInputForStep)

Copilot uses AI. Check for mistakes.

if err != nil {
x.logger.Warn("Failed to convert trigger input data to protobuf", "error", err)
}
}

triggerStep := &avsproto.Execution_Step{
Id: task.Trigger.Id,
Success: true,
Expand All @@ -199,6 +232,7 @@ func (x *TaskExecutor) RunTask(task *model.Task, queueData *QueueExecutionData)
Inputs: []string{}, // Empty inputs for trigger steps
Type: queueData.TriggerType.String(),
Name: task.Trigger.Name,
Input: triggerInputProto, // Include extracted trigger input data for debugging
}

// Set trigger output data in the step based on trigger type
Expand Down Expand Up @@ -307,7 +341,7 @@ func (x *TaskExecutor) RunTask(task *model.Task, queueData *QueueExecutionData)
if executionSuccess {
x.logger.Info("successfully executing task", "task_id", task.Id, "triggermark", queueData)
} else {
x.logger.Info("task execution completed with step failures", "task_id", task.Id, "failed_steps", failedStepCount)
x.logger.Warn("task execution completed with step failures", "task_id", task.Id, "failed_steps", failedStepCount)
}

return execution, nil
Expand Down
4 changes: 2 additions & 2 deletions core/taskengine/executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ func TestExecutorRunTaskSucess(t *testing.T) {
Id: "a1",
Type: "if",
// The test data is of this transaction https://sepolia.etherscan.io/tx/0x53beb2163994510e0984b436ebc828dc57e480ee671cfbe7ed52776c2a4830c8 which is 3.45 token
Expression: "Number(triggertest.data.value_formatted) >= 3",
Expression: "Number(triggertest.data.valueFormatted) >= 3",
},
},
},
Expand Down Expand Up @@ -390,7 +390,7 @@ func TestExecutorRunTaskReturnAllExecutionData(t *testing.T) {
{
Id: "condition1",
Type: "if",
Expression: "Number(triggertest.data.value_formatted) >= 3",
Expression: "Number(triggertest.data.valueFormatted) >= 3",
},
},
},
Expand Down
25 changes: 16 additions & 9 deletions core/taskengine/run_node_immediately.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -720,8 +720,10 @@ func (n *Engine) runProcessingNodeWithInputs(nodeType string, nodeConfig map[str
vm.WithLogger(n.logger).WithDb(n.db)

// Add input variables to VM for template processing and node access
for key, value := range inputVariables {
vm.AddVar(key, value)
// Apply dual-access mapping to enable both camelCase and snake_case field access
processedInputVariables := ProcessInputVariablesWithDualAccess(inputVariables)
for key, processedValue := range processedInputVariables {
vm.AddVar(key, processedValue)
}

// Create node from type and config
Expand All @@ -730,8 +732,8 @@ func (n *Engine) runProcessingNodeWithInputs(nodeType string, nodeConfig map[str
return nil, fmt.Errorf("failed to create node: %w", err)
}

// Execute the node
executionStep, err := vm.RunNodeWithInputs(node, inputVariables)
// Execute the node with processed input variables
executionStep, err := vm.RunNodeWithInputs(node, processedInputVariables)
if err != nil {
return nil, fmt.Errorf("node execution failed: %w", err)
}
Expand Down Expand Up @@ -1071,8 +1073,7 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi

// Log successful execution
if n.logger != nil {
n.logger.Info("RunNodeImmediatelyRPC: Executed successfully", "nodeTypeStr", nodeTypeStr, "originalNodeType", req.NodeType, "configKeys", getStringMapKeys(nodeConfig), "inputKeys", getStringMapKeys(inputVariables))

n.logger.Info("RunNodeImmediatelyRPC: Executed successfully", "nodeTypeStr", nodeTypeStr, "originalNodeType", req.NodeType)
}

// Convert result to the appropriate protobuf output type
Expand Down Expand Up @@ -1500,6 +1501,12 @@ func (n *Engine) RunTriggerRPC(user *model.User, req *avsproto.RunTriggerReq) (*
triggerConfig[k] = v.AsInterface()
}

// Extract trigger input data from the request
triggerInput := make(map[string]interface{})
for k, v := range req.TriggerInput {
triggerInput[k] = v.AsInterface()
}

// Convert TriggerType enum to string
triggerTypeStr := TriggerTypeToString(req.TriggerType)
if triggerTypeStr == "" {
Expand All @@ -1516,8 +1523,8 @@ func (n *Engine) RunTriggerRPC(user *model.User, req *avsproto.RunTriggerReq) (*
return resp, nil
}

// Execute the trigger immediately (triggers don't accept input variables)
result, err := n.runTriggerImmediately(triggerTypeStr, triggerConfig, nil)
// Execute the trigger immediately with trigger input data
result, err := n.runTriggerImmediately(triggerTypeStr, triggerConfig, triggerInput)
if err != nil {
if n.logger != nil {
// Categorize errors to avoid unnecessary stack traces for expected validation errors
Expand Down Expand Up @@ -1572,7 +1579,7 @@ func (n *Engine) RunTriggerRPC(user *model.User, req *avsproto.RunTriggerReq) (*

// Log successful execution
if n.logger != nil {
n.logger.Info("RunTriggerRPC: Executed successfully", "triggerTypeStr", triggerTypeStr, "originalTriggerType", req.TriggerType, "configKeys", getStringMapKeys(triggerConfig))
n.logger.Info("RunTriggerRPC: Executed successfully", "triggerTypeStr", triggerTypeStr, "originalTriggerType", req.TriggerType)
}

// Convert result to the appropriate protobuf output type
Expand Down
Loading
Loading