feat: "persistent" run mode

MarcStdt · MarcStdt · commit 09d5a8ffd3db · 2025-03-22T23:04:34.000+01:00
diff --git a/.vscode/launch.json b/.vscode/launch.json
@@ -27,7 +27,7 @@
                 "--cwd",
                 "${workspaceFolder}/examples/restore-test",
                 "https://gist.github.com/jlmaccal/ca6b08a307800f80986661075c82c432/raw/dbd5e7a38d0660b7047476de33da142bd79c22ef/test.tgz"
-            ],
+            ]
         },
         {
             "name": "Debug Daemon serve (restore example)",
@@ -46,6 +46,32 @@
                 "artifacts.druid.gg/druid-team/scroll-lgsm:arkserver",
                 "--coldstarter"
             ],
+            "env": {
+                "DRUID_PORT_QUERY": "2556",
+                "DRUID_PORT_MAIN": "7771"
+            }
+        },
+        {
+            "name": "Debug Daemon serve (restore example no snap)",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "console": "integratedTerminal",
+            "program": "${workspaceFolder}/main.go",
+            "args": [
+                "serve",
+                "--cwd",
+                "${workspaceFolder}/examples/ark-restore-t",
+                "-p",
+                "9190",
+                "artifacts.druid.gg/druid-team/scroll-minecraft-spigot:1.21.1",
+                "--coldstarter",
+                "--init-snapshot-url", "nil"
+            ],
+            "env": {
+                "DRUID_PORT_QUERY": "2556",
+                "DRUID_PORT_MAIN": "7771"
+            }
         },
         {
             "name": "Debug Daemon serve (htop example)",
@@ -114,7 +140,8 @@
             ],
             "env": {
                 "DRUID_PORT_QUERY": "2556",
-                "DRUID_PORT_MAIN": "7771"
+                "DRUID_PORT_MAIN": "7771",
+                "LOG_LEVEL": "debug"
             }
         },
         {
diff --git a/cmd/run.go b/cmd/run.go
@@ -46,6 +46,9 @@ var RunCmd = &cobra.Command{
 		snapshotService := snapshotService.NewSnapshotService()
 
 		_, err = initScroll(scrollService, snapshotService, processLauncher, queueManager)
+		if err != nil {
+			return fmt.Errorf("error initializing scroll: %w", err)
+		}
 
 		logger.Log().Info("Adding command to queue", zap.String("command", command))
 		err = queueManager.AddTempItem(command)
diff --git a/cmd/serve.go b/cmd/serve.go
@@ -140,11 +140,14 @@ to interact and monitor the Scroll Application`,
 
 		websocketHandler := handler.NewWebsocketHandler(authorizer, scrollService, consoleService)
 
-		s := web.NewServer(jwksUrl, scrollHandler, scrollLogHandler, scrollMetricHandler, annotationHandler, processHandler, queueHandler, websocketHandler, portHandler, healthHandler, coldstarterHandler, authorizer, cwd)
+		signalHandler := signals.NewSignalHandler(ctx, queueManager, processManager, nil, shutdownWait)
+		daemonHander := handler.NewDaemonHandler(signalHandler)
+
+		s := web.NewServer(jwksUrl, scrollHandler, scrollLogHandler, scrollMetricHandler, annotationHandler, processHandler, queueHandler, websocketHandler, portHandler, healthHandler, coldstarterHandler, daemonHander, authorizer, cwd)
 
 		a := s.Initialize()
 
-		signalHandler := signals.NewSignalHandler(ctx, queueManager, processManager, a, shutdownWait)
+		signalHandler.SetApp(a)
 
 		if watchPorts {
 			logger.Log().Info("Starting port watcher", zap.Strings("interfaces", watchPortsInterfaces))
@@ -219,7 +222,7 @@ to interact and monitor the Scroll Application`,
 								}
 							}
 
-							signalHandler.ShutdownRoutine()
+							signalHandler.ExtendedShutdownRoutine()
 						}
 					} else {
 						logger.Log().Warn("No ports to start, skipping coldstarter")
@@ -248,7 +251,7 @@ to interact and monitor the Scroll Application`,
 func init() {
 	ServeCommand.Flags().IntVarP(&port, "port", "p", 8081, "Port")
 
-	ServeCommand.Flags().IntVarP(&shutdownWait, "shutdown-wait", "", 60, "Wait interval how long the process is allowed to shutdown. First normal shutdown, then forced shutdown")
+	ServeCommand.Flags().IntVarP(&shutdownWait, "shutdown-wait", "", 10, "Wait interval how long the process is allowed to shutdown. First normal shutdown, then forced shutdown")
 
 	ServeCommand.Flags().StringVarP(&jwksUrl, "jwks-server", "", "", "JWKS Server to authenticate requests against")
 
diff --git a/cmd/server/web/server.go b/cmd/server/web/server.go
@@ -36,6 +36,7 @@ type Server struct {
 	portHandler                   ports.PortHandlerInterface
 	healthHandler                 ports.HealthHandlerInterface
 	coldstarterHandler            ports.ColdstarterHandlerInterface
+	daemonHander                  ports.SignalHandlerInterface
 	webdavPath                    string
 }
 
@@ -51,6 +52,7 @@ func NewServer(
 	portHandler ports.PortHandlerInterface,
 	healthHandler ports.HealthHandlerInterface,
 	coldstarterHandler ports.ColdstarterHandlerInterface,
+	daemonHander ports.SignalHandlerInterface,
 	authorizerService ports.AuthorizerServiceInterface,
 	webdavPath string,
 ) *Server {
@@ -73,6 +75,7 @@ func NewServer(
 		healthHandler:                 healthHandler,
 		coldstarterHandler:            coldstarterHandler,
 		webdavPath:                    webdavPath,
+		daemonHander:                  daemonHander,
 	}
 
 	if jwlsUrl != "" {
@@ -153,6 +156,8 @@ func (s *Server) SetAPI(app *fiber.App) *fiber.App {
 
 	apiRoutes.Get("/health", s.healthHandler.Health).Name("health-authenticated")
 
+	apiRoutes.Post("/daemon/stop", s.daemonHander.Stop).Name("daemon.stop")
+
 	// Create the WebDAV handler
 	webdavHandler := &webdav.Handler{
 		Prefix:     "/webdav",
@@ -189,6 +194,10 @@ func (s *Server) SetAPI(app *fiber.App) *fiber.App {
 	return app
 }
 
+func (s *Server) SetDaemonRoute(app *fiber.App, signalHandler ports.SignalHandlerInterface) {
+	app.Post("/stop", signalHandler.Stop).Name("daemon.stop")
+}
+
 func (s *Server) Serve(app *fiber.App, port int) error {
 	addr := fmt.Sprintf(":%d", port)
 	if err := app.Listen(addr); err != nil {
diff --git a/examples/minecraft/.scroll/scroll.yaml b/examples/minecraft/.scroll/scroll.yaml
@@ -32,7 +32,7 @@ init: "start"
 commands:
   start:
     needs: [install]
-    run: restart
+    run: persistent
     procedures:
       - mode: exec
         data:
diff --git a/go.mod b/go.mod
@@ -1,8 +1,6 @@
 module github.com/highcard-dev/daemon
 
-go 1.22.0
-
-toolchain go1.22.8
+go 1.23.0
 
 require (
 	github.com/Masterminds/semver/v3 v3.2.1
diff --git a/internal/core/domain/scroll.go b/internal/core/domain/scroll.go
@@ -15,9 +15,10 @@ import (
 type RunMode string
 
 const (
-	RunModeAlways  RunMode = "always" //default
-	RunModeOnce    RunMode = "once"
-	RunModeRestart RunMode = "restart"
+	RunModeAlways     RunMode = "always" //default
+	RunModeOnce       RunMode = "once"
+	RunModeRestart    RunMode = "restart"
+	RunModePersistent RunMode = "persistent" //restarts on failure and on program restart
 )
 
 type Cronjob struct {
diff --git a/internal/core/ports/handler_ports.go b/internal/core/ports/handler_ports.go
@@ -50,3 +50,7 @@ type HealthHandlerInterface interface {
 type ColdstarterHandlerInterface interface {
 	Finish(ctx *fiber.Ctx) error
 }
+
+type SignalHandlerInterface interface {
+	Stop(ctx *fiber.Ctx) error
+}
diff --git a/internal/core/services/process_manager.go b/internal/core/services/process_manager.go
@@ -140,6 +140,10 @@ func (po *ProcessManager) Run(commandName string, command []string, dir string)
 	process.Cmd = exec.Command(name, args...)
 	process.Cmd.Dir = dir
 
+	//process.Cmd.SysProcAttr = &syscall.SysProcAttr{
+	//	Setpgid: true,
+	//}
+
 	stdoutReader, err := process.Cmd.StdoutPipe()
 	if err != nil {
 		cmdDone()
@@ -171,7 +175,9 @@ func (po *ProcessManager) Run(commandName string, command []string, dir string)
 		defer wg.Done()
 		scanner := bufio.NewScanner(stdoutReader)
 		for scanner.Scan() {
-			combinedChannel <- scanner.Text() + "\n"
+			text := scanner.Text()
+			logger.Log().Debug(text)
+			combinedChannel <- text + "\n"
 		}
 	}()
 
@@ -181,7 +187,9 @@ func (po *ProcessManager) Run(commandName string, command []string, dir string)
 		defer wg.Done()
 		scanner := bufio.NewScanner(stderrReader)
 		for scanner.Scan() {
-			combinedChannel <- scanner.Text() + "\n"
+			text := scanner.Text()
+			logger.Log().Debug(text)
+			combinedChannel <- text + "\n"
 		}
 
 	}()
diff --git a/internal/core/services/process_monitor.go b/internal/core/services/process_monitor.go
@@ -89,6 +89,8 @@ func (po *ProcessMonitor) StartMonitoring() {
 }
 
 func (po *ProcessMonitor) RefreshMetrics() {
+	po.mu.Lock()
+	defer po.mu.Unlock()
 	for name, process := range po.processes {
 
 		_, err := po.GetProcessMetric(name, process)
diff --git a/internal/core/services/queue_manager.go b/internal/core/services/queue_manager.go
@@ -194,8 +194,8 @@ func (sc *QueueManager) QueueLockFile() error {
 				callback()
 			}
 
-			//not sure if this can even happen, maybe on updates
-			if command.Run != domain.RunModeRestart {
+			//not sure if this can even happen for "restart", maybe on updates
+			if command.Run != domain.RunModeRestart && command.Run != domain.RunModePersistent {
 
 				//TODO: use addQueueItem here
 				sc.mu.Lock()
@@ -276,7 +276,7 @@ func (sc *QueueManager) RunQueue() {
 		}
 
 		//if run Mode is restart, we need to run it again
-		if (status == domain.ScrollLockStatusDone) && command.Run != domain.RunModeRestart {
+		if status == domain.ScrollLockStatusDone && command.Run != domain.RunModeRestart {
 			continue
 		}
 
diff --git a/internal/handler/daemon_handler.go b/internal/handler/daemon_handler.go
@@ -0,0 +1,28 @@
+package handler
+
+import (
+	"github.com/gofiber/fiber/v2"
+	"github.com/highcard-dev/daemon/internal/signals"
+)
+
+type DaemonHandler struct {
+	shutdown *signals.SignalHandler
+}
+
+func NewDaemonHandler(shutdown *signals.SignalHandler) *DaemonHandler {
+	return &DaemonHandler{
+		shutdown: shutdown,
+	}
+}
+
+// @Summary Finish Coldstarter
+// @ID finishColdStarter
+// @Tags druid, daemon
+// @Accept */*
+// @Success 202
+// @Router /api/v1/daemon/stop [POST]
+func (ah DaemonHandler) Stop(c *fiber.Ctx) error {
+	ah.shutdown.Stop()
+	c.Status(201)
+	return nil
+}
diff --git a/internal/signals/process_shutdown.go b/internal/signals/process_shutdown.go
@@ -3,8 +3,10 @@ package signals
 import (
 	"context"
 	"fmt"
+	"maps"
 	"os"
 	"os/signal"
+	"slices"
 	"syscall"
 	"time"
 
@@ -37,6 +39,10 @@ func NewSignalHandler(ctx context.Context, queueManager ports.QueueManagerInterf
 	return sh
 }
 
+func (sh *SignalHandler) SetApp(app *fiber.App) {
+	sh.app = app
+}
+
 func (sh *SignalHandler) SetupSignals(ctx context.Context) {
 
 	signal.Notify(sh.SigC,
@@ -58,25 +64,27 @@ func (sh *SignalHandler) SetupSignals(ctx context.Context) {
 			//debug timeout for testing
 			//case <-time.After(time.Duration(25) * time.Second):
 			//	s = syscall.SIGTERM
+			go sh.queueManager.AddShutdownItem("stop")
 		}
 
 		sh.GracefulShutdown()
 	}()
 }
 
-func (sh *SignalHandler) ShutdownRoutine() {
+func (sh *SignalHandler) ExtendedShutdownRoutine() {
 
 	shudownDone := make(chan struct{})
 	go func() {
 		waitForProcessesToStop(sh.processManager)
 		shudownDone <- struct{}{}
 	}()
-	//we do that non block, in case something is allready down
-	go sh.queueManager.AddShutdownItem("stop")
 
 	//TODO: refactor this
 	done := false
 	go func() {
+		//wait for some time to await the sigterm
+		<-time.After(time.Duration(sh.waitSeconds) * time.Second)
+		go sh.queueManager.AddShutdownItem("stop")
 		<-time.After(time.Duration(sh.waitSeconds) * time.Second)
 		if done {
 			return
@@ -98,10 +106,12 @@ func (sh *SignalHandler) GracefulShutdown() {
 	logger.Log().Info("Graceful shutdown started")
 
 	logger.Log().Info("Shutdown Routine")
-	sh.ShutdownRoutine()
+	sh.ExtendedShutdownRoutine()
 
 	logger.Log().Info("Shutting down app")
-	sh.app.Shutdown()
+	if sh.app != nil {
+		sh.app.Shutdown()
+	}
 
 	logger.Log().Info("Shutdown done")
 
@@ -113,7 +123,10 @@ func waitForProcessesToStop(processManager ports.ProcessManagerInterface) {
 			logger.Log().Info("No running processes")
 			break
 		}
-		logger.Log().Info(fmt.Sprintf("Waiting for %d processes to stop...", len(processManager.GetRunningProcesses())))
+		runningPorcesses := processManager.GetRunningProcesses()
+		keys := slices.Collect(maps.Keys(runningPorcesses))
+
+		logger.Log().Info(fmt.Sprintf("Waiting for %d processes to stop...", len(runningPorcesses)), zap.Strings("processes", keys))
 		time.Sleep(time.Second)
 	}
 }

Original file line number	Diff line number	Diff line change
`@@ -50,3 +50,7 @@ type HealthHandlerInterface interface {`
`50`	`50`	`type ColdstarterHandlerInterface interface {`
`51`	`51`	`Finish(ctx *fiber.Ctx) error`
`52`	`52`	`}`
	`53`	`+`
	`54`	`+type SignalHandlerInterface interface {`
	`55`	`+ Stop(ctx *fiber.Ctx) error`
	`56`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -89,6 +89,8 @@ func (po *ProcessMonitor) StartMonitoring() {`
`89`	`89`	`}`
`90`	`90`
`91`	`91`	`func (po *ProcessMonitor) RefreshMetrics() {`
	`92`	`+ po.mu.Lock()`
	`93`	`+ defer po.mu.Unlock()`
`92`	`94`	`for name, process := range po.processes {`
`93`	`95`
`94`	`96`	`_, err := po.GetProcessMetric(name, process)`