feat: modify-volume (resize) (#14)

.claude/settings.json

@@ -1,16 +1,4 @@
 {
-  // "hooks": {
-  //   "Stop": [
-  //     {
-  //       "hooks": [
-  //         {
-  //           "type": "command",
-  //           "command": "make test"
-  //         }
-  //       ]
-  //     }
-  //   ]
-  // },
   "permissions": {
     "allow": [
       "Bash(make build)",
@@ -27,10 +15,10 @@
       "Bash(ls*)",
       "Bash(ls -la*)",
       "Bash(ls -l*)",
+      "Bash(ll*)",
       "Bash(ls -a*)",
       "Bash(ls -lh*)",
       "Bash(ls -la */)",
-      "Bash(ll*)",
       "Bash(cat *)",
       "Bash(cat -n *)",
       "Bash(less *)",
@@ -53,6 +41,13 @@
       "Bash(find . -name *)",
       "Bash(find . -type *)",
       "Bash(find . -path *)",
+      "Bash(diff *)",
+      "Bash(tree *)",
+      "Bash(pwd*)",
+      "Bash(cd *)",
+      "Bash(cd * && *)",
+      "Bash(cd ../viperblock && *)",
+      "Bash(cd ../predastore && *)",
 
       "Bash(pnpm fix)",
       "Bash(pnpm check)",

TODO.md

@@ -28,7 +28,9 @@
 - placement group instances, allow spreading instance group between nodes.
 - capabilities flag
 - more instance types - per cpu gen
-- multi part uploads on frontend
+- [DONE] multi part uploads on frontend
+- delete volumes on termination
+- fix describe instances
 
 ## EC2 create-volume
 

hive/config/config.go

@@ -119,6 +119,16 @@ type EBSUnMountResponse struct {
 	Error   string
 }
 
+type EBSSyncRequest struct {
+	Volume string
+}
+
+type EBSSyncResponse struct {
+	Volume  string
+	Synced  bool
+	Error   string
+}
+
 type EBSDeleteRequest struct {
 	Volume string
 }

hive/daemon/daemon.go

@@ -627,6 +627,15 @@ func (d *Daemon) Start() error {
 		return fmt.Errorf("failed to subscribe to NATS ec2.DescribeVolumes: %w", err)
 	}
 
+	slog.Info("Subscribing to subject pattern", "subject", "ec2.ModifyVolume")
+
+	// Subscribe to EC2 ModifyVolume with queue group
+	d.natsSubscriptions["ec2.ModifyVolume"], err = d.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", d.handleEC2ModifyVolume)
+
+	if err != nil {
+		return fmt.Errorf("failed to subscribe to NATS ec2.ModifyVolume: %w", err)
+	}
+
 	log.Printf("Subscribing to subject pattern: %s", "ec2.DescribeInstances")
 
 	// Subscribe to EC2 DescribeInstances - no queue group for multi-node fan-out
@@ -1601,6 +1610,13 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er
 				} else {
 					slog.Info("Unmounted Viperblock volume", "id", instance.ID, "data", string(msg.Data))
 				}
+
+				// Update volume state to "available" for boot volumes
+				if ebsRequest.Boot {
+					if err := d.volumeService.UpdateVolumeState(ebsRequest.Name, "available", ""); err != nil {
+						slog.Error("Failed to update volume state to available", "volumeId", ebsRequest.Name, "err", err)
+					}
+				}
 			}
 
 			// If flagged for termination (delete Volume)
@@ -1835,6 +1851,17 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) {
 	d.Instances.VMS[instance.ID] = instance
 	d.Instances.Mu.Unlock()
 
+	// Step 10: Mark boot volumes as "in-use" now that instance is confirmed running
+	instance.EBSRequests.Mu.Lock()
+	for _, ebsReq := range instance.EBSRequests.Requests {
+		if ebsReq.Boot {
+			if err := d.volumeService.UpdateVolumeState(ebsReq.Name, "in-use", instance.ID); err != nil {
+				slog.Error("Failed to update volume state to in-use", "volumeId", ebsReq.Name, "err", err)
+			}
+		}
+	}
+	instance.EBSRequests.Mu.Unlock()
+
 	err = d.WriteState()
 
 	if err != nil {
@@ -2458,6 +2485,59 @@ func (d *Daemon) handleEC2DescribeVolumes(msg *nats.Msg) {
 	slog.Info("handleEC2DescribeVolumes completed", "count", len(output.Volumes))
 }
 
+// handleEC2ModifyVolume processes incoming EC2 ModifyVolume requests
+func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) {
+	slog.Debug("Received message", "subject", msg.Subject)
+	slog.Debug("Message data", "data", string(msg.Data))
+
+	modifyVolumeInput := &ec2.ModifyVolumeInput{}
+	var errResp []byte
+
+	errResp = utils.UnmarshalJsonPayload(modifyVolumeInput, msg.Data)
+
+	if errResp != nil {
+		msg.Respond(errResp)
+		slog.Error("Request does not match ModifyVolumeInput")
+		return
+	}
+
+	slog.Info("Processing ModifyVolume request", "volumeId", modifyVolumeInput.VolumeId)
+
+	output, err := d.volumeService.ModifyVolume(modifyVolumeInput)
+
+	if err != nil {
+		slog.Error("handleEC2ModifyVolume service.ModifyVolume failed", "err", err)
+		errResp = utils.GenerateErrorPayload(err.Error())
+		msg.Respond(errResp)
+		return
+	}
+
+	jsonResponse, err := json.Marshal(output)
+	if err != nil {
+		slog.Error("handleEC2ModifyVolume failed to marshal output", "err", err)
+		errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal)
+		msg.Respond(errResp)
+		return
+	}
+	msg.Respond(jsonResponse)
+
+	// Notify viperblockd to reload state after volume modification (e.g. resize)
+	if modifyVolumeInput.VolumeId != nil {
+		syncData, err := json.Marshal(config.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId})
+		if err != nil {
+			slog.Error("failed to marshal ebs.sync request", "volumeId", *modifyVolumeInput.VolumeId, "err", err)
+		} else {
+			_, syncErr := d.natsConn.Request("ebs.sync", syncData, 5*time.Second)
+			if syncErr != nil {
+				slog.Warn("ebs.sync notification failed (volume may not be mounted)",
+					"volumeId", *modifyVolumeInput.VolumeId, "err", syncErr)
+			}
+		}
+	}
+
+	slog.Info("handleEC2ModifyVolume completed", "volumeId", modifyVolumeInput.VolumeId)
+}
+
 // handleEC2DescribeInstanceTypes processes incoming EC2 DescribeInstanceTypes requests
 // This handler responds with instance types that can currently be provisioned on this node
 // based on available resources (CPU and memory not already allocated to running instances)

hive/gateway/ec2.go

@@ -327,6 +327,28 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error {
 			return errors.New("failed to marshal response to XML")
 		}
 
+	case "ModifyVolume":
+		var input = &ec2.ModifyVolumeInput{}
+		err = awsec2query.QueryParamsToStruct(queryArgs, input)
+
+		if err != nil {
+			return err
+		}
+
+		output, err := gateway_ec2_volume.ModifyVolume(input, gw.NATSConn)
+
+		if err != nil {
+			return err
+		}
+
+		// Convert to XML
+		payload := utils.GenerateXMLPayload("ModifyVolumeResponse", output)
+		xmlOutput, err = utils.MarshalToXML(payload)
+
+		if err != nil {
+			return errors.New("failed to marshal response to XML")
+		}
+
 	default:
 		err = errors.New("InvalidAction")
 	}

hive/gateway/ec2/volume/ModifyVolume.go

@@ -0,0 +1,47 @@
+package gateway_ec2_volume
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume"
+	"github.com/nats-io/nats.go"
+)
+
+// ValidateModifyVolumeInput validates the input parameters
+func ValidateModifyVolumeInput(input *ec2.ModifyVolumeInput) error {
+	if input == nil {
+		return errors.New("InvalidParameterValue")
+	}
+
+	if input.VolumeId == nil || !strings.HasPrefix(*input.VolumeId, "vol-") {
+		return errors.New("InvalidVolumeID.Malformed")
+	}
+
+	if input.Size != nil && *input.Size <= 0 {
+		return errors.New("InvalidParameterValue")
+	}
+
+	return nil
+}
+
+// ModifyVolume handles the ModifyVolume API call
+func ModifyVolume(input *ec2.ModifyVolumeInput, natsConn *nats.Conn) (ec2.ModifyVolumeOutput, error) {
+	var output ec2.ModifyVolumeOutput
+
+	err := ValidateModifyVolumeInput(input)
+	if err != nil {
+		return output, err
+	}
+
+	volumeService := handlers_ec2_volume.NewNATSVolumeService(natsConn)
+	result, err := volumeService.ModifyVolume(input)
+
+	if err != nil {
+		return output, err
+	}
+
+	output = *result
+	return output, nil
+}

hive/gateway/ec2/volume/ModifyVolume_test.go

@@ -0,0 +1,131 @@
+package gateway_ec2_volume
+
+import (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateModifyVolumeInput(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   *ec2.ModifyVolumeInput
+		wantErr bool
+		errMsg  string
+	}{
+		{
+			name:    "NilInput",
+			input:   nil,
+			wantErr: true,
+			errMsg:  "InvalidParameterValue",
+		},
+		{
+			name:    "EmptyInput_NoVolumeId",
+			input:   &ec2.ModifyVolumeInput{},
+			wantErr: true,
+			errMsg:  "InvalidVolumeID.Malformed",
+		},
+		{
+			name: "ValidVolumeId_WithSize",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("vol-0123456789abcdef0"),
+				Size:     aws.Int64(64),
+			},
+			wantErr: false,
+		},
+		{
+			name: "ValidVolumeId_NoSize",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("vol-abc123"),
+			},
+			wantErr: false,
+		},
+		{
+			name: "InvalidVolumeId_NoPrefix",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("invalid-id"),
+				Size:     aws.Int64(64),
+			},
+			wantErr: true,
+			errMsg:  "InvalidVolumeID.Malformed",
+		},
+		{
+			name: "InvalidVolumeId_WrongPrefix",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("ami-0123456789abcdef0"),
+				Size:     aws.Int64(64),
+			},
+			wantErr: true,
+			errMsg:  "InvalidVolumeID.Malformed",
+		},
+		{
+			name: "InvalidVolumeId_EmptyString",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String(""),
+				Size:     aws.Int64(64),
+			},
+			wantErr: true,
+			errMsg:  "InvalidVolumeID.Malformed",
+		},
+		{
+			name: "InvalidVolumeId_Nil",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: nil,
+				Size:     aws.Int64(64),
+			},
+			wantErr: true,
+			errMsg:  "InvalidVolumeID.Malformed",
+		},
+		{
+			name: "InvalidSize_Zero",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("vol-abc123"),
+				Size:     aws.Int64(0),
+			},
+			wantErr: true,
+			errMsg:  "InvalidParameterValue",
+		},
+		{
+			name: "InvalidSize_Negative",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("vol-abc123"),
+				Size:     aws.Int64(-10),
+			},
+			wantErr: true,
+			errMsg:  "InvalidParameterValue",
+		},
+		{
+			name: "ValidWithVolumeType",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId:   aws.String("vol-abc123"),
+				Size:       aws.Int64(100),
+				VolumeType: aws.String("gp3"),
+			},
+			wantErr: false,
+		},
+		{
+			name: "ValidWithIops",
+			input: &ec2.ModifyVolumeInput{
+				VolumeId: aws.String("vol-abc123"),
+				Size:     aws.Int64(100),
+				Iops:     aws.Int64(3000),
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateModifyVolumeInput(tt.input)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Equal(t, tt.errMsg, err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

hive/handlers/ec2/volume/service.go

@@ -5,4 +5,5 @@ import "github.com/aws/aws-sdk-go/service/ec2"
 // VolumeService defines the interface for EBS volume operations
 type VolumeService interface {
 	DescribeVolumes(input *ec2.DescribeVolumesInput) (*ec2.DescribeVolumesOutput, error)
+	ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.ModifyVolumeOutput, error)
 }

hive/handlers/ec2/volume/service_impl.go

@@ -1,14 +1,17 @@
 package handlers_ec2_volume
 
 import (
+	"bytes"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
@@ -111,7 +114,7 @@ func (s *VolumeServiceImpl) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e
 
 		vol, err := s.getVolumeByID(volumeID)
 		if err != nil {
-			slog.Debug("Failed to get volume", "volumeId", volumeID, "err", err)
+			slog.Error("Failed to get volume", "volumeId", volumeID, "err", err)
 			continue
 		}
 
@@ -160,40 +163,15 @@ func (s *VolumeServiceImpl) fetchVolumesByIDs(volumeIDs []*string) []*ec2.Volume
 
 // getVolumeByID fetches a single volume's config from S3 and builds an EC2 Volume
 func (s *VolumeServiceImpl) getVolumeByID(volumeID string) (*ec2.Volume, error) {
-	configKey := volumeID + "/config.json"
-
-	getResult, err := s.s3Client.GetObject(&s3.GetObjectInput{
-		Bucket: aws.String(s.config.Predastore.Bucket),
-		Key:    aws.String(configKey),
-	})
+	cfg, err := s.getVolumeConfig(volumeID)
 	if err != nil {
-		if aerr, ok := err.(awserr.Error); ok && (aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound") {
-			slog.Debug("Config file not found", "key", configKey)
-			return nil, errors.New("volume not found")
-		}
-		slog.Debug("Failed to get config file", "key", configKey, "err", err)
-		return nil, err
-	}
-	defer getResult.Body.Close()
-
-	body, err := io.ReadAll(getResult.Body)
-	if err != nil {
-		slog.Debug("Failed to read config body", "key", configKey, "err", err)
 		return nil, err
 	}
 
-	var vbConfig struct {
-		VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
-	}
-	if err := json.Unmarshal(body, &vbConfig); err != nil {
-		slog.Debug("Failed to unmarshal config", "key", configKey, "err", err)
-		return nil, err
-	}
-
-	volMeta := vbConfig.VolumeConfig.VolumeMetadata
+	volMeta := cfg.VolumeMetadata
 
 	if volMeta.VolumeID == "" {
-		slog.Debug("Volume ID is empty in config", "key", configKey)
+		slog.Debug("Volume ID is empty in config", "key", volumeID+"/config.json")
 		return nil, errors.New("volume ID is empty")
 	}
 
@@ -253,3 +231,202 @@ func (s *VolumeServiceImpl) getVolumeByID(volumeID string) (*ec2.Volume, error)
 
 	return volume, nil
 }
+
+// volumeConfigWrapper matches the JSON structure stored in S3 config.json files
+type volumeConfigWrapper struct {
+	VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
+}
+
+// getVolumeConfig reads the raw VolumeConfig from S3 for a given volume ID
+func (s *VolumeServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeConfig, error) {
+	configKey := volumeID + "/config.json"
+
+	getResult, err := s.s3Client.GetObject(&s3.GetObjectInput{
+		Bucket: aws.String(s.config.Predastore.Bucket),
+		Key:    aws.String(configKey),
+	})
+	if err != nil {
+		if aerr, ok := err.(awserr.Error); ok && (aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound") {
+			return nil, errors.New(awserrors.ErrorInvalidVolumeNotFound)
+		}
+		return nil, fmt.Errorf("failed to get config: %w", err)
+	}
+	defer getResult.Body.Close()
+
+	body, err := io.ReadAll(getResult.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read config body: %w", err)
+	}
+
+	var wrapper volumeConfigWrapper
+	if err := json.Unmarshal(body, &wrapper); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal config: %w", err)
+	}
+
+	return &wrapper.VolumeConfig, nil
+}
+
+// putVolumeConfig writes a VolumeConfig back to S3 as config.json.
+// It performs a read-modify-write to preserve full VBState if viperblock
+// has already written state (BlockSize, SeqNum, WALNum, etc.) to config.json.
+func (s *VolumeServiceImpl) putVolumeConfig(volumeID string, cfg *viperblock.VolumeConfig) error {
+	configKey := volumeID + "/config.json"
+
+	data, err := s.mergeVolumeConfig(configKey, cfg)
+	if err != nil {
+		return err
+	}
+
+	_, err = s.s3Client.PutObject(&s3.PutObjectInput{
+		Bucket: aws.String(s.config.Predastore.Bucket),
+		Key:    aws.String(configKey),
+		Body:   bytes.NewReader(data),
+	})
+	if err != nil {
+		return fmt.Errorf("failed to write config to S3: %w", err)
+	}
+
+	return nil
+}
+
+// mergeVolumeConfig reads existing config.json from S3 and merges the new
+// VolumeConfig into it, preserving full VBState when present. If no existing
+// VBState is found, it returns a plain volumeConfigWrapper.
+func (s *VolumeServiceImpl) mergeVolumeConfig(configKey string, cfg *viperblock.VolumeConfig) ([]byte, error) {
+	getResult, err := s.s3Client.GetObject(&s3.GetObjectInput{
+		Bucket: aws.String(s.config.Predastore.Bucket),
+		Key:    aws.String(configKey),
+	})
+	if err != nil {
+		if aerr, ok := err.(awserr.Error); ok && (aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound") {
+			// No existing config.json -- write wrapper for new volume
+			return json.Marshal(volumeConfigWrapper{VolumeConfig: *cfg})
+		}
+		return nil, fmt.Errorf("failed to read existing config for merge: %w", err)
+	}
+	defer getResult.Body.Close()
+
+	body, err := io.ReadAll(getResult.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read existing config: %w", err)
+	}
+
+	var state viperblock.VBState
+	if json.Unmarshal(body, &state) != nil || state.BlockSize == 0 {
+		// Not a full VBState (new volume or wrapper-only) -- write wrapper
+		return json.Marshal(volumeConfigWrapper{VolumeConfig: *cfg})
+	}
+
+	// Full VBState exists -- update VolumeConfig and reconcile VolumeSize
+	state.VolumeConfig = *cfg
+	configSizeBytes := uint64(cfg.VolumeMetadata.SizeGiB) * 1024 * 1024 * 1024
+	if configSizeBytes > 0 && configSizeBytes > state.VolumeSize {
+		state.VolumeSize = configSizeBytes
+	}
+
+	slog.Info("putVolumeConfig: preserved VBState", "volumeId", strings.TrimSuffix(configKey, "/config.json"),
+		"blockSize", state.BlockSize, "seqNum", state.SeqNum)
+
+	return json.Marshal(state)
+}
+
+// UpdateVolumeState updates the State and AttachedInstance fields for a volume in S3.
+func (s *VolumeServiceImpl) UpdateVolumeState(volumeID, state, attachedInstance string) error {
+	cfg, err := s.getVolumeConfig(volumeID)
+	if err != nil {
+		return fmt.Errorf("failed to get volume config for state update: %w", err)
+	}
+
+	cfg.VolumeMetadata.State = state
+	cfg.VolumeMetadata.AttachedInstance = attachedInstance
+
+	if err := s.putVolumeConfig(volumeID, cfg); err != nil {
+		return fmt.Errorf("failed to write volume config for state update: %w", err)
+	}
+
+	slog.Info("Updated volume state", "volumeId", volumeID, "state", state, "attachedInstance", attachedInstance)
+	return nil
+}
+
+// ModifyVolume modifies an EBS volume (grow-only, requires stopped instance)
+func (s *VolumeServiceImpl) ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.ModifyVolumeOutput, error) {
+	if input.VolumeId == nil || *input.VolumeId == "" {
+		return nil, errors.New(awserrors.ErrorInvalidVolumeIDMalformed)
+	}
+
+	volumeID := *input.VolumeId
+	slog.Info("ModifyVolume request", "volumeId", volumeID)
+
+	cfg, err := s.getVolumeConfig(volumeID)
+	if err != nil {
+		slog.Error("ModifyVolume failed to get volume config", "volumeId", volumeID, "err", err)
+		return nil, err
+	}
+
+	volMeta := &cfg.VolumeMetadata
+
+	// Record original values before modification
+	originalSize := utils.SafeUint64ToInt64(volMeta.SizeGiB)
+	originalType := volMeta.VolumeType
+	if originalType == "" {
+		originalType = "gp3"
+	}
+	originalIOPS := int64(volMeta.IOPS)
+
+	// Validate: grow only (new size must be greater than current)
+	if input.Size != nil && *input.Size <= originalSize {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+
+	// Validate: if volume is attached, instance must not be in-use (must be stopped)
+	if volMeta.AttachedInstance != "" && volMeta.State == "in-use" {
+		return nil, errors.New(awserrors.ErrorIncorrectState)
+	}
+
+	// Apply modifications
+	if input.Size != nil {
+		volMeta.SizeGiB = utils.SafeInt64ToUint64(*input.Size)
+	}
+	if input.VolumeType != nil {
+		volMeta.VolumeType = *input.VolumeType
+	}
+	if input.Iops != nil {
+		volMeta.IOPS = int(*input.Iops)
+	}
+
+	// Persist updated config
+	if err := s.putVolumeConfig(volumeID, cfg); err != nil {
+		slog.Error("ModifyVolume failed to write config", "volumeId", volumeID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// Build target values (after modification)
+	targetSize := utils.SafeUint64ToInt64(volMeta.SizeGiB)
+	targetType := volMeta.VolumeType
+	if targetType == "" {
+		targetType = "gp3"
+	}
+	targetIOPS := int64(volMeta.IOPS)
+
+	now := time.Now()
+	modification := &ec2.VolumeModification{
+		VolumeId:           aws.String(volumeID),
+		ModificationState:  aws.String("completed"),
+		Progress:           aws.Int64(100),
+		OriginalSize:       aws.Int64(originalSize),
+		OriginalVolumeType: aws.String(originalType),
+		OriginalIops:       aws.Int64(originalIOPS),
+		TargetSize:         aws.Int64(targetSize),
+		TargetVolumeType:   aws.String(targetType),
+		TargetIops:         aws.Int64(targetIOPS),
+		StartTime:          aws.Time(now),
+		EndTime:            aws.Time(now),
+	}
+
+	slog.Info("ModifyVolume completed", "volumeId", volumeID,
+		"originalSize", originalSize, "targetSize", targetSize)
+
+	return &ec2.ModifyVolumeOutput{
+		VolumeModification: modification,
+	}, nil
+}

hive/handlers/ec2/volume/service_nats.go

@@ -2,6 +2,7 @@ package handlers_ec2_volume
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
@@ -37,7 +38,7 @@ func (s *NATSVolumeService) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e
 	// Validate error response
 	responseError, err := utils.ValidateErrorPayload(msg.Data)
 	if err != nil {
-		return nil, fmt.Errorf("daemon returned error: %s", *responseError.Code)
+		return nil, errors.New(*responseError.Code)
 	}
 
 	// Unmarshal successful response
@@ -49,3 +50,29 @@ func (s *NATSVolumeService) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e
 
 	return &output, nil
 }
+
+// ModifyVolume sends a ModifyVolume request via NATS and waits for response
+func (s *NATSVolumeService) ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.ModifyVolumeOutput, error) {
+	jsonData, err := json.Marshal(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal input: %w", err)
+	}
+
+	msg, err := s.natsConn.Request("ec2.ModifyVolume", jsonData, 30*time.Second)
+	if err != nil {
+		return nil, fmt.Errorf("NATS request failed: %w", err)
+	}
+
+	responseError, err := utils.ValidateErrorPayload(msg.Data)
+	if err != nil {
+		return nil, errors.New(*responseError.Code)
+	}
+
+	var output ec2.ModifyVolumeOutput
+	err = json.Unmarshal(msg.Data, &output)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unmarshal response: %w", err)
+	}
+
+	return &output, nil
+}

hive/services/viperblockd/viperblockd.go

@@ -29,6 +29,7 @@ type MountedVolume struct {
 	Socket string // Unix socket path (when using socket transport)
 	NBDURI string // Full NBD URI (nbd:unix:/path.sock or nbd://host:port)
 	PID    int
+	VB     *viperblock.VB // Reference to viperblock instance for state sync/flush
 }
 
 type Config struct {
@@ -137,37 +138,47 @@ func launchService(cfg *Config) (err error) {
 			return
 		}
 
-		// Find the volume in the mounted volumes
+		// Find the volume and extract references while holding the lock,
+		// then release before calling VB.Close() (which does heavy S3 I/O).
 		var ebsResponse config.EBSUnMountResponse
-		var match bool
+		var matched MountedVolume
+		var matchIdx int = -1
 		cfg.mu.Lock()
-		for _, volume := range cfg.MountedVolumes {
-
-			// TODO: Confirm KVM/QEMU is not using the volume first.
+		for i, volume := range cfg.MountedVolumes {
 			if volume.Name == ebsRequest.Name {
-				ebsResponse = config.EBSUnMountResponse{
-					Volume:  volume.Name,
-					Mounted: false,
+				matched = volume
+				matchIdx = i
+				// Remove from slice while we hold the lock
+				cfg.MountedVolumes = append(cfg.MountedVolumes[:i], cfg.MountedVolumes[i+1:]...)
+				break
+			}
+		}
+		cfg.mu.Unlock()
+
+		if matchIdx >= 0 {
+			ebsResponse = config.EBSUnMountResponse{
+				Volume:  matched.Name,
+				Mounted: false,
+			}
+
+			// Clean up the VB instance's background goroutine.
+			// This VB is state-only (LoadState/sync) — actual I/O is in the nbdkit plugin process.
+			if matched.VB != nil {
+				matched.VB.StopWALSyncer()
+			}
+
+			utils.KillProcess(matched.PID)
+
+			// Remove the socket file if using socket transport
+			if matched.Socket != "" {
+				slog.Info("Removing socket file", "socket", matched.Socket)
+				if err := os.Remove(matched.Socket); err != nil && !os.IsNotExist(err) {
+					slog.Error("Failed to delete nbd socket", "err", err, "socket", matched.Socket)
 				}
-
-				utils.KillProcess(volume.PID)
-				match = true
-
-				// Remove the socket file if using socket transport
-				if volume.Socket != "" {
-					slog.Info("Removing socket file", "socket", volume.Socket)
-					err := os.Remove(volume.Socket)
-					if err != nil && !os.IsNotExist(err) {
-						slog.Error("Failed to delete nbd socket", "err", err, "socket", volume.Socket)
-					}
-				}
-
 			}
 		}
 
-		cfg.mu.Unlock()
-
-		if !match {
+		if matchIdx < 0 {
 			ebsResponse = config.EBSUnMountResponse{
 				Volume: ebsRequest.Name,
 				Error:  fmt.Sprintf("Volume %s not found", ebsRequest.Name),
@@ -185,6 +196,52 @@ func launchService(cfg *Config) (err error) {
 		nc.Publish("ebs.unmount.response", response)
 	})
 
+	nc.QueueSubscribe("ebs.sync", "hive-workers", func(msg *nats.Msg) {
+		slog.Info("Received ebs.sync message", "data", string(msg.Data))
+
+		var syncRequest config.EBSSyncRequest
+		if err := json.Unmarshal(msg.Data, &syncRequest); err != nil {
+			slog.Error("Failed to unmarshal ebs.sync message", "err", err)
+			errResp, _ := json.Marshal(config.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)})
+			msg.Respond(errResp)
+			return
+		}
+
+		syncResponse := config.EBSSyncResponse{Volume: syncRequest.Volume}
+
+		// Find the mounted volume and reload its state from the backend
+		cfg.mu.Lock()
+		var foundVB *viperblock.VB
+		for _, volume := range cfg.MountedVolumes {
+			if volume.Name == syncRequest.Volume && volume.VB != nil {
+				foundVB = volume.VB
+				break
+			}
+		}
+		cfg.mu.Unlock()
+
+		if foundVB == nil {
+			syncResponse.Error = fmt.Sprintf("volume %s not mounted or has no VB instance", syncRequest.Volume)
+			slog.Warn("ebs.sync: volume not found", "volume", syncRequest.Volume)
+		} else if err := foundVB.LoadState(); err != nil {
+			syncResponse.Error = fmt.Sprintf("failed to reload state: %v", err)
+			slog.Error("ebs.sync: LoadState failed", "volume", syncRequest.Volume, "err", err)
+		} else {
+			syncResponse.Synced = true
+			slog.Info("ebs.sync: state reloaded", "volume", syncRequest.Volume,
+				"volumeSize", foundVB.GetVolumeSize())
+		}
+
+		response, err := json.Marshal(syncResponse)
+		if err != nil {
+			slog.Error("Failed to marshal ebs.sync response", "err", err)
+			msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
+			return
+		}
+
+		msg.Respond(response)
+	})
+
 	nc.QueueSubscribe("ebs.mount", "hive-workers", func(msg *nats.Msg) {
 		slog.Info("Received message:", "data", string(msg.Data))
 
@@ -425,6 +482,7 @@ func launchService(cfg *Config) (err error) {
 			Socket: nbdSocket,
 			NBDURI: nbdURI,
 			PID:    pid,
+			VB:     vb,
 		})
 		cfg.mu.Unlock()
 
@@ -452,13 +510,21 @@ func launchService(cfg *Config) (err error) {
 
 	nc.Close()
 
-	// Close all nbdkit processes
+	// Snapshot mounted volumes and clear the list while holding the lock,
+	// then flush/kill outside the lock (VB.Close does heavy I/O).
 	cfg.mu.Lock()
-	for _, volume := range cfg.MountedVolumes {
+	volumes := make([]MountedVolume, len(cfg.MountedVolumes))
+	copy(volumes, cfg.MountedVolumes)
+	cfg.MountedVolumes = nil
+	cfg.mu.Unlock()
+
+	for _, volume := range volumes {
+		if volume.VB != nil {
+			volume.VB.StopWALSyncer()
+		}
 		slog.Info("Killing nbdkit process", "pid", volume.PID)
 		utils.KillProcess(volume.PID)
 	}
-	cfg.mu.Unlock()
 
 	return nil