Implement the execution engine

.gitignore

@@ -10,6 +10,7 @@ out/
 # Dotenv file
 .env
 *.env
+.env*
 
 # binary
 ap-avs
@@ -24,4 +25,7 @@ metadata.json
 # Ignores yaml configuration files in the config directory
 config/*.yaml
 
-contracts/script/output/
+contracts/script/output*
+
+contracts/broadcast/
+alias-ecdsa.key.json

Makefile

@@ -91,3 +91,8 @@ build-docker:
 ## up: bring up docker compose stack
 up:
 	docker compose up
+
+## unstable-build: generate an unstable for internal test
+unstable-build:
+	docker build --platform=linux/amd64 --build-arg RELEASE_TAG=unstable -t avaprotocol/ap-avs:unstable -f dockerfiles/operator.Dockerfile .
+	docker push avaprotocol/ap-avs:unstable

aggregator/aggregator.go

@@ -15,8 +15,11 @@ import (
 
 	"github.com/AvaProtocol/ap-avs/aggregator/types"
 	"github.com/AvaProtocol/ap-avs/core"
+	"github.com/AvaProtocol/ap-avs/core/apqueue"
 	"github.com/AvaProtocol/ap-avs/core/chainio"
+	"github.com/AvaProtocol/ap-avs/core/chainio/aa"
 	"github.com/AvaProtocol/ap-avs/core/config"
+	"github.com/AvaProtocol/ap-avs/core/taskengine"
 	"github.com/Layr-Labs/eigensdk-go/chainio/clients"
 	sdkclients "github.com/Layr-Labs/eigensdk-go/chainio/clients"
 	blsagg "github.com/Layr-Labs/eigensdk-go/services/bls_aggregation"
@@ -36,12 +39,19 @@ const (
 	avsName                  = "oak-avs"
 )
 
+type AggregatorStatus string
+
+const (
+	initStatus     AggregatorStatus = "init"
+	runningStatus  AggregatorStatus = "running"
+	shutdownStatus AggregatorStatus = "shutdown"
+)
+
 func RunWithConfig(configPath string) error {
 	nodeConfig, err := config.NewConfig(configPath)
 	if err != nil {
 		panic(fmt.Errorf("failed to parse config file: %w\nMake sure %s is exist and a valid yaml file %w.", configPath, err))
 	}
-	fmt.Printf("loaded config: %v\n", nodeConfig)
 
 	aggregator, err := NewAggregator(nodeConfig)
 	if err != nil {
@@ -55,6 +65,7 @@ func RunWithConfig(configPath string) error {
 type Aggregator struct {
 	logger    logging.Logger
 	avsWriter chainio.AvsWriterer
+
 	// aggregation related fields
 	blsAggregationService blsagg.BlsAggregationService
 	tasks                 map[types.TaskIndex]cstaskmanager.IAutomationTaskManagerTask
@@ -69,6 +80,16 @@ type Aggregator struct {
 	chainID      *big.Int
 
 	operatorPool *OperatorPool
+
+	// task engines handles trigger scheduling and send distribute checks to
+	// operator to checks
+	engine *taskengine.Engine
+	// upon a task condition is met, taskengine will schedule it in our queue to
+	// be executed.
+	queue  *apqueue.Queue
+	worker *apqueue.Worker
+
+	status AggregatorStatus
 }
 
 // NewAggregator creates a new Aggregator with the provided config.
@@ -100,9 +121,7 @@ func NewAggregator(c *config.Config) (*Aggregator, error) {
 			panic(err)
 			//return nil, err
 		}
-
-		fmt.Println("avsReader", avsReader, "clients", clients)
-
+		c.Logger.Info("create avsrrader and client", "avsReader", avsReader, "clients", clients)
 	}()
 
 	// TODO: These are erroring out and we don't need them now yet
@@ -122,6 +141,7 @@ func NewAggregator(c *config.Config) (*Aggregator, error) {
 		config: c,
 
 		operatorPool: &OperatorPool{},
+		status:       initStatus,
 	}, nil
 }
 
@@ -160,22 +180,36 @@ func (agg *Aggregator) init() {
 	} else {
 		config.CurrentChainEnv = config.HoleskyEnv
 	}
+
+	// Setup account abstraction config
+	aa.SetFactoryAddress(agg.config.SmartWallet.FactoryAddress)
+	aa.SetEntrypointAddress(agg.config.SmartWallet.EntrypointAddress)
 }
 
 func (agg *Aggregator) Start(ctx context.Context) error {
+	agg.status = runningStatus
+
 	agg.logger.Infof("Starting aggregator")
 
 	agg.init()
 
 	agg.logger.Infof("Initialize Storagre")
-	agg.initDB(ctx)
+	if err := agg.initDB(ctx); err != nil {
+		agg.logger.Fatalf("failed to initialize storage", "error", err)
+	}
 
-	agg.logger.Infof("Starting rpc server.")
-	go agg.startRpcServer(ctx)
+	agg.logger.Infof("Starting Task engine")
+	go agg.startTaskEngine(ctx)
+
+	agg.logger.Info("Starting repl")
+	go agg.startRepl()
 
-	agg.logger.Infof("Starting http server.")
+	agg.logger.Infof("Starting http server")
 	go agg.startHttpServer(ctx)
 
+	agg.logger.Infof("Starting rpc server")
+	go agg.startRpcServer(ctx)
+
 	// Setup wait signal
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
@@ -186,10 +220,12 @@ func (agg *Aggregator) Start(ctx context.Context) error {
 	}()
 
 	<-done
-	agg.logger.Infof("Shutting down.")
+	agg.logger.Infof("Shutting down...")
 
-	// Shutdown the db
 	// TODO: handle ongoing client and fanout closing
+	agg.status = shutdownStatus
+	agg.stopRepl()
+	agg.stopTaskEngine()
 	agg.db.Close()
 
 	return nil
@@ -235,3 +271,7 @@ func (agg *Aggregator) sendAggregatedResponseToContract(blsAggServiceResp blsagg
 		agg.logger.Error("Aggregator failed to respond to task", "err", err)
 	}
 }
+
+func (agg *Aggregator) IsShutdown() bool {
+	return agg.status == shutdownStatus
+}

aggregator/pool.go

@@ -68,6 +68,8 @@ func (o *OperatorPool) Checkin(payload *avsproto.Checkin) error {
 		Version:       payload.Version,
 	}
 
+	fmt.Println("signature", payload.Signature)
+
 	data, err := json.Marshal(status)
 
 	if err != nil {

aggregator/repl.go

@@ -0,0 +1,132 @@
+package aggregator
+
+import (
+	"bufio"
+	"fmt"
+	"log"
+	"net"
+	"os"
+	"strings"
+)
+
+var (
+	repListener net.Listener
+)
+
+func (agg *Aggregator) stopRepl() {
+	if repListener != nil {
+		repListener.Close()
+	}
+
+}
+func (agg *Aggregator) startRepl() {
+	var err error
+	repListener, err = net.Listen("unix", agg.config.SocketPath)
+	if err != nil {
+		return
+	}
+
+	for {
+		if agg.IsShutdown() {
+			return
+		}
+		conn, err := repListener.Accept()
+		if err != nil {
+			log.Println("Failed to accept connection:", err)
+			continue
+		}
+
+		go handleConnection(agg, conn)
+	}
+}
+
+func handleConnection(agg *Aggregator, conn net.Conn) {
+	defer func() {
+		agg.logger.Info("Close repl connection", "remote", conn.RemoteAddr().String())
+		conn.Close()
+	}()
+
+	reader := bufio.NewReader(conn)
+	fmt.Fprintln(conn, "AP CLI REPL")
+	fmt.Fprintln(conn, "-------------------------")
+
+	for {
+		fmt.Fprint(conn, "> ")
+		input, err := reader.ReadString('\n')
+
+		// Check for EOF (Ctrl+D)
+		if err != nil {
+			if err.Error() == "EOF" {
+				fmt.Fprintln(conn, "\nExiting...")
+				break
+			}
+			log.Println("Error reading input:", err)
+			continue
+		}
+
+		input = strings.TrimSpace(input)
+
+		if input == "" {
+			continue
+		}
+
+		parts := strings.SplitN(input, " ", 2)
+		command := strings.ToLower(parts[0])
+
+		switch command {
+		case "list":
+			if len(parts) == 2 {
+				if keys, err := agg.db.ListKeys(parts[1]); err == nil {
+					for _, k := range keys {
+						fmt.Fprintln(conn, k)
+					}
+				}
+
+			} else {
+				fmt.Fprintln(conn, "Usage: list <prefix>* or list *")
+			}
+		case "get":
+			if len(parts) == 2 {
+				if key, err := agg.db.GetKey([]byte(parts[1])); err == nil {
+					fmt.Fprintln(conn, string(key))
+				}
+			} else {
+				fmt.Fprintln(conn, "Usage: get <key>")
+			}
+		case "set":
+			parts = strings.SplitN(input, " ", 3)
+			if len(parts) >= 3 {
+				if parts[2][0] == '@' {
+					if content, err := os.ReadFile(parts[2][1:]); err == nil {
+						if err = agg.db.Set([]byte(parts[1]), content); err == nil {
+							fmt.Fprintln(conn, "written "+string(parts[1]))
+						}
+					} else {
+						fmt.Fprintln(conn, "invalid file "+parts[2][1:])
+					}
+				} else {
+					if err = agg.db.Set([]byte(parts[1]), []byte(parts[2])); err == nil {
+						fmt.Fprintln(conn, "written "+parts[1])
+					}
+
+				}
+			} else {
+				fmt.Fprintln(conn, "Usage: set <key> @/path-to-file")
+			}
+		case "gc":
+			fmt.Fprintln(conn, "start gc with 0.7")
+			err := agg.db.Vacuum()
+			if err == nil {
+				fmt.Fprintln(conn, "gc success. still have more to run")
+			} else {
+				fmt.Fprintln(conn, "gc is done. no more log file to be removed")
+			}
+
+		case "exit":
+			fmt.Fprintln(conn, "Exiting...")
+			return
+		default:
+			fmt.Fprintln(conn, "Unknown command:", command)
+		}
+	}
+}