Passing result to subsequent steps, implement JavaScript runner

aggregator/aggregator.go

@@ -107,7 +107,8 @@ func NewAggregator(c *config.Config) (*Aggregator, error) {
 	avsWriter, err := chainio.BuildAvsWriterFromConfig(c)
 	if err != nil {
 		c.Logger.Errorf("Cannot create avsWriter", "err", err)
-		return nil, err
+		// EigenLayer has update the contract and we cannot fetch the slasher anymore, we should upgrade the EigenSDK, right now we don't use it so it's ok to ignore this error
+		//return nil, err
 	}
 
 	go func() {
@@ -122,8 +123,8 @@ func NewAggregator(c *config.Config) (*Aggregator, error) {
 		clients, err := clients.BuildAll(chainioConfig, c.EcdsaPrivateKey, c.Logger)
 		if err != nil {
 			c.Logger.Errorf("Cannot create sdk clients", "err", err)
-			panic(err)
-			//return nil, err
+			// EigenLayer has update the contract and we cannot fetch the slasher anymore, we should upgrade the EigenSDK, right now we don't use it so it's ok to ignore this error
+			//panic(err)
 		}
 		c.Logger.Info("create avsrrader and client", "avsReader", avsReader, "clients", clients)
 	}()

aggregator/rpc_server.go

@@ -53,6 +53,7 @@ func (r *RpcServer) GetWallet(ctx context.Context, payload *avsproto.GetWalletRe
 	r.config.Logger.Info("process create wallet",
 		"user", user.Address.String(),
 		"salt", payload.Salt,
+		"factory", payload.FactoryAddress,
 	)
 
 	return r.engine.CreateSmartWallet(user, payload)
@@ -82,7 +83,7 @@ func (r *RpcServer) ListWallets(ctx context.Context, payload *avsproto.ListWalle
 	r.config.Logger.Info("process list wallet",
 		"address", user.Address.String(),
 	)
-	wallets, err := r.engine.GetSmartWallets(user.Address)
+	wallets, err := r.engine.GetSmartWallets(user.Address, payload)
 	if err != nil {
 		return nil, status.Errorf(codes.Unavailable, "rpc server is unavailable, retry later. %s", err.Error())
 	}
@@ -157,6 +158,7 @@ func (r *RpcServer) ListTasks(ctx context.Context, payload *avsproto.ListTasksRe
 	r.config.Logger.Info("process list task",
 		"user", user.Address.String(),
 		"smart_wallet_address", payload.SmartWalletAddress,
+		"cursor", payload.Cursor,
 	)
 	return r.engine.ListTasksByUser(user, payload)
 }
@@ -169,11 +171,26 @@ func (r *RpcServer) ListExecutions(ctx context.Context, payload *avsproto.ListEx
 
 	r.config.Logger.Info("process list execution",
 		"user", user.Address.String(),
-		"task_id", payload.Id,
+		"task_id", payload.TaskIds,
+		"cursor", payload.Cursor,
 	)
 	return r.engine.ListExecutions(user, payload)
 }
 
+func (r *RpcServer) GetExecution(ctx context.Context, payload *avsproto.GetExecutionReq) (*avsproto.Execution, error) {
+	user, err := r.verifyAuth(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Unauthenticated, "%s: %s", auth.AuthenticationError, err.Error())
+	}
+
+	r.config.Logger.Info("process get execution",
+		"user", user.Address.String(),
+		"task_id", payload.TaskId,
+		"execution_id", payload.ExecutionId,
+	)
+	return r.engine.GetExecution(user, payload)
+}
+
 func (r *RpcServer) GetTask(ctx context.Context, payload *avsproto.IdReq) (*avsproto.Task, error) {
 	user, err := r.verifyAuth(ctx)
 	if err != nil {

core/chainio/aa/aa.go

@@ -77,7 +77,7 @@ func GetSenderAddressForFactory(conn *ethclient.Client, ownerAddress common.Addr
 	}
 
 	sender, err := simpleFactory.GetAddress(nil, ownerAddress, salt)
-	return &sender, nil
+	return &sender, err
 }
 
 func GetNonce(conn *ethclient.Client, ownerAddress common.Address, salt *big.Int) (*big.Int, error) {

core/taskengine/cursor.go

@@ -33,6 +33,10 @@ func CursorFromString(data string) (*Cursor, error) {
 		ulidPos:   ulid.Zero,
 	}
 
+	if data == "" {
+		return c, nil
+	}
+
 	decoded, err := base64.StdEncoding.DecodeString(data)
 	if err != nil {
 		return c, err
@@ -54,6 +58,11 @@ func NewCursor(direction CursorDirection, position string) *Cursor {
 		int64Pos: 0,
 	}
 }
+
+func (c *Cursor) IsZero() bool {
+	return c.Position == "0"
+}
+
 func (c *Cursor) String() string {
 	var d []byte
 	d, err := json.Marshal(c)
@@ -68,7 +77,7 @@ func (c *Cursor) String() string {
 }
 
 // Given a value, return true if the value is after the cursor
-func (c *Cursor) AfterInt64(value int64) bool {
+func (c *Cursor) LessThanInt64(value int64) bool {
 	if !c.parsePos {
 		c.int64Pos, _ = strconv.ParseInt(c.Position, 10, 64)
 		c.parsePos = true
@@ -81,7 +90,7 @@ func (c *Cursor) AfterInt64(value int64) bool {
 }
 
 // Given a value, return true if the value is after the cursor
-func (c *Cursor) AfterUlid(value ulid.ULID) bool {
+func (c *Cursor) LessThanUlid(value ulid.ULID) bool {
 	if !c.parsePos {
 		var err error
 		c.ulidPos, err = ulid.Parse(c.Position)
@@ -95,3 +104,32 @@ func (c *Cursor) AfterUlid(value ulid.ULID) bool {
 	}
 	return c.ulidPos.Compare(value) > 0
 }
+
+// Given a value, return true if the value is after the cursor
+func (c *Cursor) LessThanOrEqualInt64(value int64) bool {
+	if !c.parsePos {
+		c.int64Pos, _ = strconv.ParseInt(c.Position, 10, 64)
+		c.parsePos = true
+	}
+	if c.Direction == CursorDirectionNext {
+		return c.int64Pos <= value
+	}
+
+	return c.int64Pos >= value
+}
+
+// Given a value, return true if the value is after the cursor
+func (c *Cursor) LessThanOrEqualUlid(value ulid.ULID) bool {
+	if !c.parsePos {
+		var err error
+		c.ulidPos, err = ulid.Parse(c.Position)
+		if err != nil {
+			c.ulidPos = ulid.Zero
+		}
+		c.parsePos = true
+	}
+	if c.Direction == CursorDirectionNext {
+		return c.ulidPos.Compare(value) <= 0
+	}
+	return c.ulidPos.Compare(value) >= 0
+}

core/taskengine/doc.go

@@ -10,7 +10,7 @@ w:<eoa>:<smart-wallet-address> = {factory_address: address, salt: salt}
 w:<eoa>:<smart-wallet-address> -> {factory, salt}
 t:<task-status>:<task-id>      -> task payload, the source of truth of task information
 u:<eoa>:<smart-wallet-address>:<task-id>    -> task status
-h:<smart-wallet-address>:<task-id>:<execution-id> -> an execution history
+history:<task-id>:<execution-id> -> an execution history
 
 The task storage was designed for fast retrieve time at the cost of extra storage.