fix: error handling (#76)

cmd/hive/cmd/root.go

@@ -116,7 +116,11 @@ func initConfig() {
 	appConfig, err = config.LoadConfig(cfgFile)
 
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "Warning: %v\n", err)
-		fmt.Fprintln(os.Stderr, "Continuing with environment variables and defaults...")
+		// If a config file was explicitly provided, treat load failure as fatal
+		if cfgFile != "" {
+			fmt.Fprintf(os.Stderr, "Error: failed to load config %s: %v\n", cfgFile, err)
+			os.Exit(1)
+		}
+		// No config specified — continue with env/defaults (e.g., hive --help)
 	}
 }

cmd/hive/cmd/service.go

@@ -148,7 +148,10 @@ var predastoreStartCmd = &cobra.Command{
 			return
 		}
 
-		service.Start()
+		if _, err := service.Start(); err != nil {
+			fmt.Println("Error starting predastore service:", err)
+			os.Exit(1)
+		}
 
 		fmt.Println("Predastore service started", service)
 	},
@@ -167,7 +170,10 @@ var predastoreStopCmd = &cobra.Command{
 			return
 		}
 
-		service.Stop()
+		if err = service.Stop(); err != nil {
+			fmt.Println("Error stopping predastore service:", err)
+			os.Exit(1)
+		}
 
 		fmt.Println("Predastore service stopped")
 
@@ -319,7 +325,10 @@ var viperblockStopCmd = &cobra.Command{
 			return
 		}
 
-		service.Stop()
+		if err = service.Stop(); err != nil {
+			fmt.Println("Error stopping viperblock service:", err)
+			os.Exit(1)
+		}
 
 		fmt.Println("Viperblock service stopped")
 
@@ -363,7 +372,10 @@ var natsStartCmd = &cobra.Command{
 			return
 		}
 
-		service.Start()
+		if _, err = service.Start(); err != nil {
+			fmt.Println("Error starting nats service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("NATS service started")
 	},
 }
@@ -381,7 +393,10 @@ var natsStopCmd = &cobra.Command{
 			return
 		}
 
-		service.Stop()
+		if err = service.Stop(); err != nil {
+			fmt.Println("Error stopping nats service:", err)
+			os.Exit(1)
+		}
 
 		fmt.Println("Nats service stopped")
 	},
@@ -448,7 +463,10 @@ var hiveStartCmd = &cobra.Command{
 			hiveSvc.SetConfigPath(cfgFile)
 		}
 
-		svc.Start()
+		if _, err = svc.Start(); err != nil {
+			fmt.Println("Error starting hive service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("HIVE service started")
 	},
 }
@@ -466,7 +484,10 @@ var hiveStopCmd = &cobra.Command{
 			return
 		}
 
-		service.Stop()
+		if err = service.Stop(); err != nil {
+			fmt.Println("Error stopping hive service:", err)
+			os.Exit(1)
+		}
 
 		fmt.Println("Hive service stopped")
 	},
@@ -542,7 +563,10 @@ var awsgwStartCmd = &cobra.Command{
 			return
 		}
 
-		service.Start()
+		if _, err = service.Start(); err != nil {
+			fmt.Println("Error starting awsgw service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("AWSGW service started")
 	},
 }
@@ -560,7 +584,10 @@ var awsgwStopCmd = &cobra.Command{
 			return
 		}
 
-		service.Stop()
+		if err = service.Stop(); err != nil {
+			fmt.Println("Error stopping awsgw service:", err)
+			os.Exit(1)
+		}
 
 		fmt.Println("AWSGW service stopped")
 	},
@@ -597,7 +624,10 @@ var hiveUIStartCmd = &cobra.Command{
 			return
 		}
 
-		svc.Start()
+		if _, err = svc.Start(); err != nil {
+			fmt.Println("Error starting hive-ui service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("hive-ui service started")
 	},
 }
@@ -615,7 +645,10 @@ var hiveUIStopCmd = &cobra.Command{
 			return
 		}
 
-		svc.Stop()
+		if err = svc.Stop(); err != nil {
+			fmt.Println("Error stopping hive-ui service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("hive-ui service stopped")
 	},
 }
@@ -674,7 +707,10 @@ var vpcdStartCmd = &cobra.Command{
 			return
 		}
 
-		svc.Start()
+		if _, err = svc.Start(); err != nil {
+			fmt.Println("Error starting vpcd service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("vpcd service started")
 	},
 }
@@ -691,7 +727,10 @@ var vpcdStopCmd = &cobra.Command{
 			return
 		}
 
-		svc.Stop()
+		if err = svc.Stop(); err != nil {
+			fmt.Println("Error stopping vpcd service:", err)
+			os.Exit(1)
+		}
 		fmt.Println("vpcd service stopped")
 	},
 }

hive/daemon/daemon.go

@@ -15,6 +15,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"runtime"
+	"runtime/debug"
 	"strconv"
 	"strings"
 	"sync"
@@ -22,6 +23,8 @@ import (
 	"syscall"
 	"time"
 
+	"net/http"
+
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/go-chi/chi/v5"
@@ -46,7 +49,6 @@ import (
 	"github.com/mulgadc/viperblock/viperblock"
 	"github.com/nats-io/nats.go"
 	"github.com/pelletier/go-toml/v2"
-	"net/http"
 )
 
 type BlockDeviceMapping struct {
@@ -910,6 +912,11 @@ func (d *Daemon) restoreInstances() {
 			go func(inst *vm.VM) {
 				defer wg.Done()
 				defer func() { <-sem }() // release
+				defer func() {
+					if r := recover(); r != nil {
+						slog.Error("Panic during instance recovery", "instanceId", inst.ID, "panic", r, "stack", string(debug.Stack()))
+					}
+				}()
 				slog.Info("Launching instance (recovery)", "instance", inst.ID)
 				if err := d.LaunchInstance(inst); err != nil {
 					slog.Error("Failed to launch instance", "instanceId", inst.ID, "err", err)
@@ -1087,7 +1094,9 @@ func (d *Daemon) ClusterManager() error {
 		}
 
 		w.Header().Set("Content-Type", "application/json")
-		_ = json.NewEncoder(w).Encode(response)
+		if err := json.NewEncoder(w).Encode(response); err != nil {
+			slog.Error("Failed to encode health response", "error", err)
+		}
 	})
 
 	// Join endpoint - accepts new nodes joining the cluster
@@ -1096,10 +1105,12 @@ func (d *Daemon) ClusterManager() error {
 		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 			w.Header().Set("Content-Type", "application/json")
 			w.WriteHeader(400)
-			_ = json.NewEncoder(w).Encode(types.NodeJoinResponse{
+			if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{
 				Success: false,
 				Message: "invalid request body",
-			})
+			}); err != nil {
+				slog.Error("Failed to encode join error response", "error", err)
+			}
 			return
 		}
 
@@ -1109,10 +1120,12 @@ func (d *Daemon) ClusterManager() error {
 		if req.Node == "" || req.Region == "" || req.AZ == "" {
 			w.Header().Set("Content-Type", "application/json")
 			w.WriteHeader(400)
-			_ = json.NewEncoder(w).Encode(types.NodeJoinResponse{
+			if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{
 				Success: false,
 				Message: "node, region, and az are required",
-			})
+			}); err != nil {
+				slog.Error("Failed to encode join validation response", "error", err)
+			}
 			return
 		}
 
@@ -1120,10 +1133,12 @@ func (d *Daemon) ClusterManager() error {
 		if _, exists := d.clusterConfig.Nodes[req.Node]; exists {
 			w.Header().Set("Content-Type", "application/json")
 			w.WriteHeader(409)
-			_ = json.NewEncoder(w).Encode(types.NodeJoinResponse{
+			if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{
 				Success: false,
 				Message: fmt.Sprintf("node %s already exists in cluster", req.Node),
-			})
+			}); err != nil {
+				slog.Error("Failed to encode join conflict response", "error", err)
+			}
 			return
 		}
 
@@ -1153,10 +1168,12 @@ func (d *Daemon) ClusterManager() error {
 			slog.Error("Failed to save cluster config", "error", err)
 			w.Header().Set("Content-Type", "application/json")
 			w.WriteHeader(500)
-			_ = json.NewEncoder(w).Encode(types.NodeJoinResponse{
+			if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{
 				Success: false,
 				Message: "failed to save cluster config",
-			})
+			}); err != nil {
+				slog.Error("Failed to encode join save-error response", "error", err)
+			}
 			return
 		}
 
@@ -1209,7 +1226,7 @@ func (d *Daemon) ClusterManager() error {
 		}
 
 		w.Header().Set("Content-Type", "application/json")
-		_ = json.NewEncoder(w).Encode(types.NodeJoinResponse{
+		if err := json.NewEncoder(w).Encode(types.NodeJoinResponse{
 			Success:          true,
 			Message:          fmt.Sprintf("node %s successfully joined cluster", req.Node),
 			SharedData:       sharedData,
@@ -1218,7 +1235,9 @@ func (d *Daemon) ClusterManager() error {
 			CACert:           caCert,
 			CAKey:            caKey,
 			PredastoreConfig: predastoreConfig,
-		})
+		}); err != nil {
+			slog.Error("Failed to encode join success response", "error", err)
+		}
 	})
 
 	// Get cluster config endpoint
@@ -1226,10 +1245,12 @@ func (d *Daemon) ClusterManager() error {
 		configHash, _ := d.computeConfigHash()
 
 		w.Header().Set("Content-Type", "application/json")
-		_ = json.NewEncoder(w).Encode(map[string]any{
+		if err := json.NewEncoder(w).Encode(map[string]any{
 			"config":      d.clusterConfig,
 			"config_hash": configHash,
-		})
+		}); err != nil {
+			slog.Error("Failed to encode config response", "error", err)
+		}
 	})
 
 	// Start HTTP server in goroutine

hive/daemon/jetstream.go

@@ -3,6 +3,7 @@ package daemon
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"strings"
 	"sync"
@@ -301,11 +302,14 @@ func (m *JetStreamManager) WriteShutdownMarker(nodeID string) error {
 	if m.clusterKV == nil {
 		return errors.New("cluster state KV not initialized")
 	}
-	data, _ := json.Marshal(map[string]any{
+	data, err := json.Marshal(map[string]any{
 		"node":      nodeID,
 		"timestamp": time.Now().UTC().Format(time.RFC3339),
 	})
-	_, err := m.clusterKV.Put("shutdown."+nodeID, data)
+	if err != nil {
+		return fmt.Errorf("marshal shutdown marker: %w", err)
+	}
+	_, err = m.clusterKV.Put("shutdown."+nodeID, data)
 	return err
 }
 
@@ -341,14 +345,17 @@ func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string
 	if m.clusterKV == nil {
 		return errors.New("cluster state KV not initialized")
 	}
-	data, _ := json.Marshal(map[string]any{
+	data, err := json.Marshal(map[string]any{
 		"node":            nodeID,
 		"services":        services,
 		"nats_host":       natsHost,
 		"predastore_host": predastoreHost,
 		"timestamp":       time.Now().UTC().Format(time.RFC3339),
 	})
-	_, err := m.clusterKV.Put("node."+nodeID+".services", data)
+	if err != nil {
+		return fmt.Errorf("marshal service manifest: %w", err)
+	}
+	_, err = m.clusterKV.Put("node."+nodeID+".services", data)
 	return err
 }
 

hive/daemon/shutdown.go

@@ -53,6 +53,7 @@ func (d *Daemon) handleShutdownGate(msg *nats.Msg) {
 	var req ShutdownRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleShutdownGate: failed to unmarshal request", "error", err)
+		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "gate", Error: err.Error()})
 		return
 	}
 
@@ -105,6 +106,7 @@ func (d *Daemon) handleShutdownDrain(msg *nats.Msg) {
 	var req ShutdownRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleShutdownDrain: failed to unmarshal request", "error", err)
+		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "drain", Error: err.Error()})
 		return
 	}
 
@@ -162,6 +164,7 @@ func (d *Daemon) handleShutdownStorage(msg *nats.Msg) {
 	var req ShutdownRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleShutdownStorage: failed to unmarshal request", "error", err)
+		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "storage", Error: err.Error()})
 		return
 	}
 
@@ -196,6 +199,7 @@ func (d *Daemon) handleShutdownPersist(msg *nats.Msg) {
 	var req ShutdownRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleShutdownPersist: failed to unmarshal request", "error", err)
+		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "persist", Error: err.Error()})
 		return
 	}
 
@@ -226,6 +230,7 @@ func (d *Daemon) handleShutdownInfra(msg *nats.Msg) {
 	var req ShutdownRequest
 	if err := json.Unmarshal(msg.Data, &req); err != nil {
 		slog.Error("handleShutdownInfra: failed to unmarshal request", "error", err)
+		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "infra", Error: err.Error()})
 		return
 	}
 
@@ -290,9 +295,10 @@ func (d *Daemon) publishShutdownProgress(phase string, total, remaining int) {
 	}
 	data, err := json.Marshal(progress)
 	if err != nil {
+		slog.Error("Failed to marshal shutdown progress", "error", err)
 		return
 	}
 	if err := d.natsConn.Publish("hive.cluster.shutdown.progress", data); err != nil {
-		slog.Debug("Failed to publish shutdown progress", "error", err)
+		slog.Warn("Failed to publish shutdown progress", "error", err)
 	}
 }

hive/gateway/ec2.go

@@ -234,7 +234,11 @@ var ec2LocalActions = map[string]bool{
 }
 
 func (gw *GatewayConfig) EC2_Request(w http.ResponseWriter, r *http.Request) error {
-	body, _ := io.ReadAll(r.Body)
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		slog.Error("Failed to read EC2 request body", "error", err)
+		return errors.New(awserrors.ErrorInternalError)
+	}
 	queryArgs := ParseAWSQueryArgs(string(body))
 
 	action := queryArgs["Action"]

hive/gateway/iam.go

@@ -94,7 +94,11 @@ var iamActions = map[string]IAMHandler{
 }
 
 func (gw *GatewayConfig) IAM_Request(w http.ResponseWriter, r *http.Request) error {
-	body, _ := io.ReadAll(r.Body)
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		slog.Error("Failed to read IAM request body", "error", err)
+		return errors.New(awserrors.ErrorInternalError)
+	}
 	queryArgs := ParseAWSQueryArgs(string(body))
 
 	action := queryArgs["Action"]

hive/qmp/qmp.go

@@ -2,8 +2,10 @@ package qmp
 
 import (
 	"encoding/json"
+	"fmt"
 	"net"
 	"sync"
+	"time"
 )
 
 type unmarshalTarget any
@@ -117,7 +119,11 @@ func NewQMPClient(path string) (*QMPClient, error) {
 		return nil, err
 	}
 
-	//conn.SetReadDeadline(time.Now().Add(30 * time.Second))
+	// Set deadline for greeting — a hung QEMU would block forever without this
+	if err := conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil {
+		_ = conn.Close()
+		return nil, fmt.Errorf("set read deadline: %w", err)
+	}
 
 	client := &QMPClient{
 		Conn:    conn,
@@ -129,7 +135,13 @@ func NewQMPClient(path string) (*QMPClient, error) {
 	var greeting QMPGreeting
 	if err := client.Decoder.Decode(&greeting); err != nil {
 		_ = conn.Close()
-		return nil, err
+		return nil, fmt.Errorf("waiting for QMP greeting: %w", err)
+	}
+
+	// Clear deadline for normal operations
+	if err := conn.SetReadDeadline(time.Time{}); err != nil {
+		_ = conn.Close()
+		return nil, fmt.Errorf("clear read deadline: %w", err)
 	}
 
 	// enable capabilities

hive/services/awsgw/awsgw.go

@@ -96,7 +96,9 @@ func launchService(config *config.ClusterConfig) error {
 
 	// First boot: consume bootstrap.json → seed IAM users into NATS KV → delete file
 	bootstrapPath := filepath.Join(nodeConfig.BaseDir, "config", "bootstrap.json")
-	if data, err := handlers_iam.LoadBootstrapData(bootstrapPath); err == nil {
+	data, err := handlers_iam.LoadBootstrapData(bootstrapPath)
+	switch {
+	case err == nil:
 		slog.Info("Bootstrap file found, seeding IAM users")
 		if err := iamService.SeedBootstrap(data); err != nil {
 			return fmt.Errorf("seed bootstrap from bootstrap.json: %w", err)
@@ -105,6 +107,10 @@ func launchService(config *config.ClusterConfig) error {
 			slog.Warn("Failed to delete bootstrap file", "path", bootstrapPath, "err", err)
 		}
 		slog.Info("Bootstrap complete, bootstrap.json deleted")
+	case os.IsNotExist(err):
+		// No bootstrap file — normal after first boot
+	default:
+		return fmt.Errorf("load bootstrap from %s: %w", bootstrapPath, err)
 	}
 
 	// Create gateway with NATS connection

hive/services/viperblockd/viperblockd.go

@@ -240,7 +240,7 @@ func launchService(cfg *Config) (err error) {
 			slog.Error("Failed to respond to ebs.delete request", "err", err)
 		}
 	}); err != nil {
-		slog.Error("Failed to subscribe to ebs.delete", "err", err)
+		return fmt.Errorf("failed to subscribe to ebs.delete: %w", err)
 	}
 
 	// Subscribe to node-specific unmount topic if NodeName is set, otherwise fall back to generic queue group
@@ -261,7 +261,11 @@ func launchService(cfg *Config) (err error) {
 		var ebsRequest types.EBSRequest
 		err := json.Unmarshal(msg.Data, &ebsRequest)
 		if err != nil {
-			slog.Error("Failed to unmarshal message", "err", err)
+			slog.Error("Failed to unmarshal ebs.unmount message", "err", err)
+			errResp, _ := json.Marshal(types.EBSUnMountResponse{Error: fmt.Sprintf("bad request: %v", err)})
+			if err := msg.Respond(errResp); err != nil {
+				slog.Error("Failed to respond to ebs.unmount request", "err", err)
+			}
 			return
 		}
 
@@ -335,7 +339,7 @@ func launchService(cfg *Config) (err error) {
 			slog.Error("Failed to publish ebs.unmount.response", "err", err)
 		}
 	}); err != nil {
-		slog.Error("Failed to subscribe to unmount topic", "topic", unmountTopic, "err", err)
+		return fmt.Errorf("failed to subscribe to %s: %w", unmountTopic, err)
 	}
 
 	if _, err := nc.QueueSubscribe("ebs.sync", "hive-workers", func(msg *nats.Msg) {
@@ -389,7 +393,7 @@ func launchService(cfg *Config) (err error) {
 			slog.Error("Failed to respond to ebs.sync request", "err", err)
 		}
 	}); err != nil {
-		slog.Error("Failed to subscribe to ebs.sync", "err", err)
+		return fmt.Errorf("failed to subscribe to ebs.sync: %w", err)
 	}
 
 	// Note: ebs.snapshot is handled per-volume via ebs.snapshot.{volumeID} topics,
@@ -414,7 +418,11 @@ func launchService(cfg *Config) (err error) {
 		var ebsRequest types.EBSRequest
 		err := json.Unmarshal(msg.Data, &ebsRequest)
 		if err != nil {
-			slog.Error("Failed to unmarshal message", "err", err)
+			slog.Error("Failed to unmarshal ebs.mount message", "err", err)
+			errResp, _ := json.Marshal(types.EBSMountResponse{Error: fmt.Sprintf("bad request: %v", err)})
+			if err := msg.Respond(errResp); err != nil {
+				slog.Error("Failed to respond to ebs.mount request", "err", err)
+			}
 			return
 		}
 
@@ -546,6 +554,14 @@ func launchService(cfg *Config) (err error) {
 			nbdPort, err = strconv.Atoi(parts[len(parts)-1])
 			if err != nil {
 				slog.Error("Failed to convert port to int", "err", err)
+				ebsResponse.Error = fmt.Sprintf("failed to parse port: %v", err)
+				response, _ := json.Marshal(ebsResponse)
+				if err := msg.Respond(response); err != nil {
+					slog.Error("Failed to respond to ebs.mount request", "err", err)
+				}
+				if err := nc.Publish("ebs.mount.response", response); err != nil {
+					slog.Error("Failed to publish ebs.mount.response", "err", err)
+				}
 				return
 			}
 
@@ -573,7 +589,15 @@ func launchService(cfg *Config) (err error) {
 		// Generate PID file for nbdkit process
 		nbdPidFile, err := utils.GeneratePidFile(fmt.Sprintf("nbdkit-vol-%s", ebsRequest.Name))
 		if err != nil {
-			slog.Error("Failed to generate nbdkit pid file:", "err", err)
+			slog.Error("Failed to generate nbdkit pid file", "err", err)
+			ebsResponse.Error = fmt.Sprintf("failed to generate pid file: %v", err)
+			response, _ := json.Marshal(ebsResponse)
+			if err := msg.Respond(response); err != nil {
+				slog.Error("Failed to respond to ebs.mount request", "err", err)
+			}
+			if err := nc.Publish("ebs.mount.response", response); err != nil {
+				slog.Error("Failed to publish ebs.mount.response", "err", err)
+			}
 			return
 		}
 
@@ -709,7 +733,7 @@ func launchService(cfg *Config) (err error) {
 
 		slog.Debug("Sent ebs.mount response", "response", string(response))
 	}); err != nil {
-		slog.Error("Failed to subscribe to mount topic", "topic", mountTopic, "err", err)
+		return fmt.Errorf("failed to subscribe to %s: %w", mountTopic, err)
 	}
 
 	// Create a channel to receive shutdown signals