Tweak booting order and update workflow

.github/workflows/pr.yml → .github/workflows/push.yml

@@ -1,4 +1,4 @@
-name: Pull Request
+name: Run on Push
 
 on:
   push:

.github/workflows/release.yaml

@@ -9,7 +9,7 @@ on:
         description: tag the release
         required: true
 
-name: Latest Release
+name: Public A Release
 
 defaults:
   run:

aggregator/aggregator.go

@@ -232,8 +232,6 @@ func (agg *Aggregator) init() {
 }
 
 func (agg *Aggregator) Start(ctx context.Context) error {
-	agg.status = runningStatus
-
 	agg.logger.Infof("Starting aggregator %s", version.Get())
 
 	agg.init()
@@ -244,16 +242,17 @@ func (agg *Aggregator) Start(ctx context.Context) error {
 	}
 
 	agg.logger.Infof("Starting Task engine")
-	go agg.startTaskEngine(ctx)
-
-	agg.logger.Info("Starting repl")
-	go agg.startRepl()
+	agg.startTaskEngine(ctx)
 
 	agg.logger.Infof("Starting rpc server")
-	go agg.startRpcServer(ctx)
+	agg.startRpcServer(ctx)
+
+	agg.logger.Info("Starting repl")
+	agg.startRepl()
 
 	agg.logger.Infof("Starting http server")
-	go agg.startHttpServer(ctx)
+	agg.startHttpServer(ctx)
+	agg.status = runningStatus
 
 	// Setup wait signal
 	sigs := make(chan os.Signal, 1)

aggregator/http_server.go

@@ -24,7 +24,11 @@ func (agg *Aggregator) startHttpServer(ctx context.Context) {
 	e := echo.New()
 
 	e.GET("/up", func(c echo.Context) error {
-		return c.String(http.StatusOK, "AvaProtocol is up")
+		if agg.status == runningStatus {
+			return c.String(http.StatusOK, "up")
+		}
+
+		return c.String(http.StatusServiceUnavailable, "pending...")
 	})
 
 	e.GET("/operator", func(c echo.Context) error {
@@ -51,5 +55,7 @@ func (agg *Aggregator) startHttpServer(ctx context.Context) {
 		return c.HTMLBlob(http.StatusOK, buf.Bytes())
 	})
 
-	e.Logger.Fatal(e.Start(":1323"))
+	go func() {
+		e.Logger.Fatal(e.Start(":1323"))
+	}()
 }

aggregator/repl.go

@@ -35,18 +35,20 @@ func (agg *Aggregator) startRepl() {
 		return
 	}
 
-	for {
-		if agg.IsShutdown() {
-			return
-		}
-		conn, err := repListener.Accept()
-		if err != nil {
-			log.Println("Failed to accept connection:", err)
-			continue
-		}
+	go func() {
+		for {
+			if agg.IsShutdown() {
+				return
+			}
+			conn, err := repListener.Accept()
+			if err != nil {
+				log.Println("Failed to accept connection:", err)
+				continue
+			}
 
-		go handleConnection(agg, conn)
-	}
+			go handleConnection(agg, conn)
+		}
+	}()
 }
 
 func handleConnection(agg *Aggregator, conn net.Conn) {

aggregator/rpc_server.go

@@ -251,8 +251,11 @@ func (agg *Aggregator) startRpcServer(ctx context.Context) error {
 	agg.logger.Info("start grpc server",
 		"address", lis.Addr(),
 	)
-	if err := s.Serve(lis); err != nil {
-		log.Fatalf("failed to serve: %v", err)
-	}
+
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("failed to serve: %v", err)
+		}
+	}()
 	return nil
 }