From e3fae3d96faabf21633a829ae3304a185f3e018f Mon Sep 17 00:00:00 2001
From: Ben Duncan <5178243+benduncan@users.noreply.github.com>
Date: Mon, 19 Jan 2026 14:08:18 +1000
Subject: [PATCH] Latest predastore changes for multi-node

---
 hive/daemon/daemon.go                  |   2 +-
 hive/services/predastore/predastore.go | 117 ++++++++++++-------------
 2 files changed, 55 insertions(+), 64 deletions(-)

diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go
index 5f0cb5c1c..14fb580a9 100644
--- a/hive/daemon/daemon.go
+++ b/hive/daemon/daemon.go
@@ -852,7 +852,7 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
 			return
 		}
 
-		// Last, delete the instance volumes
+		// TODO: Last, delete the instance volumes
 
 		// Free resources
 		instanceType := d.resourceMgr.instanceTypes[instance.InstanceType]
diff --git a/hive/services/predastore/predastore.go b/hive/services/predastore/predastore.go
index a69f21ea8..96993feb7 100644
--- a/hive/services/predastore/predastore.go
+++ b/hive/services/predastore/predastore.go
@@ -2,8 +2,6 @@ package predastore
 
 import (
 	"context"
-	"fmt"
-	"log"
 	"log/slog"
 	"os"
 	"os/signal"
@@ -12,11 +10,14 @@ import (
 
 	"github.com/mulgadc/hive/hive/utils"
 	"github.com/mulgadc/predastore/s3"
-	"go.uber.org/automaxprocs/maxprocs"
+
+	// Import backends to trigger their init() registration
+	_ "github.com/mulgadc/predastore/backend/filesystem"
 )
 
 var serviceName = "predastore"
 
+// Config holds the configuration for the predastore service
 type Config struct {
 	ConfigPath string
 	Port       int
@@ -27,91 +28,81 @@ type Config struct {
 	TlsKey     string
 }
 
+// Service wraps the predastore S3 server
 type Service struct {
 	Config *Config
+	server *s3.Server
 }
 
+// New creates a new predastore service
 func New(config interface{}) (svc *Service, err error) {
 	svc = &Service{
 		Config: config.(*Config),
 	}
-
 	return svc, nil
 }
 
+// Start starts the predastore service
 func (svc *Service) Start() (int, error) {
-
 	utils.WritePidFile(serviceName, os.Getpid())
-	launchService(svc.Config)
+
+	server, err := s3.NewServer(
+		s3.WithConfigPath(svc.Config.ConfigPath),
+		s3.WithAddress(svc.Config.Host, svc.Config.Port),
+		s3.WithTLS(svc.Config.TlsCert, svc.Config.TlsKey),
+		s3.WithBasePath(svc.Config.BasePath),
+		s3.WithDebug(svc.Config.Debug),
+	)
+	if err != nil {
+		slog.Error("Failed to create predastore server", "error", err)
+		return 0, err
+	}
+
+	svc.server = server
+
+	// Start server asynchronously
+	if err := server.ListenAndServeAsync(); err != nil {
+		slog.Error("Failed to start predastore server", "error", err)
+		return 0, err
+	}
+
+	// Wait for shutdown signal
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+	<-sigChan
+
+	slog.Info("Shutting down predastore service")
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := server.Shutdown(ctx); err != nil {
+		slog.Error("Error during shutdown", "error", err)
+	}
 
 	return os.Getpid(), nil
 }
 
-func (svc *Service) Stop() (err error) {
-
-	err = utils.StopProcess(serviceName)
-	return err
+// Stop stops the predastore service
+func (svc *Service) Stop() error {
+	return utils.StopProcess(serviceName)
 }
 
+// Status returns the status of the predastore service
 func (svc *Service) Status() (string, error) {
 	return "", nil
 }
 
-func (svc *Service) Shutdown() (err error) {
+// Shutdown gracefully shuts down the predastore service
+func (svc *Service) Shutdown() error {
+	if svc.server != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		return svc.server.Shutdown(ctx)
+	}
 	return svc.Stop()
 }
 
-func (svc *Service) Reload() (err error) {
+// Reload reloads the predastore service configuration
+func (svc *Service) Reload() error {
 	return nil
 }
-
-func launchService(config *Config) {
-
-	s3 := s3.New(&s3.Config{
-		ConfigPath: config.ConfigPath,
-		Port:       config.Port,
-		Host:       config.Host,
-		Debug:      config.Debug,
-		BasePath:   config.BasePath,
-	})
-
-	// Adjust MAXPROCS if running under linux/cgroups quotas.
-	undo, err := maxprocs.Set(maxprocs.Logger(log.Printf))
-	if err != nil {
-		log.Printf("Failed to set GOMAXPROCS: %v", err)
-	} else {
-		defer undo()
-	}
-
-	err = s3.ReadConfig()
-
-	if err != nil {
-		slog.Warn("Error reading config file", "error", err)
-		os.Exit(-1)
-	}
-
-	app := s3.SetupRoutes()
-
-	go func() {
-		log.Fatal(app.ListenTLS(fmt.Sprintf("%s:%d", config.Host, config.Port), config.TlsCert, config.TlsKey))
-	}()
-
-	// Create a channel to receive shutdown signals
-	sigChan := make(chan os.Signal, 1)
-	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
-
-	// Wait for shutdown signal
-	<-sigChan
-	log.Println("Shutting down gracefully...")
-
-	// Graceful shutdown with timeout
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	if err := app.ShutdownWithContext(ctx); err != nil {
-		log.Printf("Error during shutdown: %v", err)
-	}
-
-	log.Println("Server stopped")
-
-}