diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 4cbe05d1b..080cdc820 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -30,7 +30,7 @@ make test # Run all unit tests (sets LOG_IGNORE=1) make check-format # Check gofmt (fails on diff, same as CI) make format # Fix gofmt in place make vet # Run go vet (fails on issues) -make security # Run go vet + govulncheck, gosec, staticcheck (fails on issues) +make security-check # Run govulncheck, gosec, staticcheck (fails on issues) make bench # Run benchmarks make clean # Remove build artifacts make test-docker # Run single + multi-node E2E in Docker (requires /dev/kvm) diff --git a/Makefile b/Makefile index 1ab9f96d4..937cf950d 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,5 @@ GO_PROJECT_NAME := hive +SHELL := /bin/bash # Detect architecture for cross-platform support ARCH := $(shell uname -m) @@ -37,30 +38,30 @@ build: # Build hive-ui frontend (requires pnpm) build-ui: - @echo "\n....Building hive-ui frontend...." + @echo -e "\n....Building hive-ui frontend...." cd hive/services/hiveui/frontend && pnpm build # GO commands go_build: - @echo "\n....Building $(GO_PROJECT_NAME)" + @echo -e "\n....Building $(GO_PROJECT_NAME)" go build $(GO_BUILD_MOD) -ldflags "-s -w" -o ./bin/$(GO_PROJECT_NAME) cmd/hive/main.go go_run: - @echo "\n....Running $(GO_PROJECT_NAME)...." + @echo -e "\n....Running $(GO_PROJECT_NAME)...." $(GOPATH)/bin/$(GO_PROJECT_NAME) # Preflight — runs the same checks as GitHub Actions (format + lint + security + tests). # Use this before committing to catch CI failures locally. preflight: check-format vet security-check test - @echo "\n✅ Preflight passed — safe to commit." + @echo -e "\n ✅ Preflight passed — safe to commit." # Run unit tests test: - @echo "\n....Running tests for $(GO_PROJECT_NAME)...." + @echo -e "\n....Running tests for $(GO_PROJECT_NAME)...." LOG_IGNORE=1 go test -v -timeout 120s ./... bench: - @echo "\n....Running benchmarks for $(GO_PROJECT_NAME)...." + @echo -e "\n....Running benchmarks for $(GO_PROJECT_NAME)...." $(MAKE) easyjson LOG_IGNORE=1 go test -benchmem -run=. -bench=. ./... @@ -73,7 +74,7 @@ clean: rm -rf hive/services/hiveui/frontend/dist install-system: - @echo "\n....Installing system dependencies for $(ARCH)...." + @echo -e "\n....Installing system dependencies for $(ARCH)...." @echo "QEMU packages: $(QEMU_PACKAGES)" apt-get update && sudo apt-get install -y \ nbdkit nbdkit-plugin-dev pkg-config $(QEMU_PACKAGES) qemu-utils qemu-kvm \ @@ -81,7 +82,7 @@ install-system: iproute2 netcat-openbsd openssh-client wget git unzip sudo xz-utils file install-go: - @echo "\n....Installing Go 1.26.0 for $(ARCH) ($(GO_ARCH))...." + @echo -e "\n....Installing Go 1.26.0 for $(ARCH) ($(GO_ARCH))...." @if [ ! -d "/usr/local/go" ]; then \ curl -L https://go.dev/dl/go1.26.0.linux-$(GO_ARCH).tar.gz | tar -C /usr/local -xz; \ else \ @@ -90,7 +91,7 @@ install-go: @echo "Go version: $$(go version)" install-aws: - @echo "\n....Installing AWS CLI v2 for $(ARCH) ($(AWS_ARCH))...." + @echo -e "\n....Installing AWS CLI v2 for $(ARCH) ($(AWS_ARCH))...." @if ! command -v aws >/dev/null 2>&1; then \ curl "https://awscli.amazonaws.com/awscli-exe-linux-$(AWS_ARCH).zip" -o "awscliv2.zip"; \ unzip -o awscliv2.zip; \ @@ -101,7 +102,7 @@ install-aws: fi quickinstall: install-system install-go install-aws - @echo "\n✅ Quickinstall complete for $(ARCH)." + @echo -e "\n✅ Quickinstall complete for $(ARCH)." @echo " Please ensure /usr/local/go/bin is in your PATH." @echo " Installed: Go ($(GO_ARCH)), AWS CLI ($(AWS_ARCH)), QEMU ($(QEMU_PACKAGES))" @@ -130,12 +131,12 @@ vet: # Security checks — each tool fails the build on findings (matches CI). # Reports are also saved to tests/ for review. security-check: - @echo "\n....Running security checks for $(GO_PROJECT_NAME)...." - go tool govulncheck ./... 2>&1 | tee tests/govulncheck-report.txt + @echo -e "\n....Running security checks for $(GO_PROJECT_NAME)...." + set -o pipefail && go tool govulncheck ./... 2>&1 | tee tests/govulncheck-report.txt @echo " govulncheck ok" - go tool gosec -exclude=G204,G304,G402 -exclude-generated -exclude-dir=cmd ./... 2>&1 | tee tests/gosec-report.txt + set -o pipefail && go tool gosec -exclude=G204,G304,G402 -exclude-generated -exclude-dir=cmd ./... 2>&1 | tee tests/gosec-report.txt @echo " gosec ok" - go tool staticcheck -checks="all,-ST1000,-ST1003,-ST1016,-ST1020,-ST1021,-ST1022,-SA1019,-SA9005" ./... 2>&1 | tee tests/staticcheck-report.txt + set -o pipefail && go tool staticcheck -checks="all,-ST1000,-ST1003,-ST1016,-ST1020,-ST1021,-ST1022,-SA1019,-SA9005" ./... 2>&1 | tee tests/staticcheck-report.txt @echo " staticcheck ok" # Legacy alias — now runs the strict checks (same as security-check) @@ -149,7 +150,7 @@ PARENT_DIR := $(shell cd .. && pwd) E2E_IMAGE := hive-e2e:latest test-docker-build: - @echo "\n....Building E2E Docker image...." + @echo -e "\n....Building E2E Docker image...." @for dep in viperblock predastore; do \ if [ ! -d "$(PARENT_DIR)/$$dep" ]; then \ echo "Missing sibling repo $$dep — running clone-deps.sh"; \ @@ -160,17 +161,17 @@ test-docker-build: docker build -t $(E2E_IMAGE) -f tests/e2e/Dockerfile.e2e $(PARENT_DIR) test-docker-single: test-docker-build - @echo "\n....Running Single-Node E2E Tests...." + @echo -e "\n....Running Single-Node E2E Tests...." docker run --privileged --rm -v /dev/kvm:/dev/kvm $(E2E_IMAGE) test-docker-multi: test-docker-build - @echo "\n....Running Multi-Node E2E Tests...." + @echo -e "\n....Running Multi-Node E2E Tests...." docker run --privileged --rm -v /dev/kvm:/dev/kvm --cap-add=NET_ADMIN $(E2E_IMAGE) ./tests/e2e/run-multinode-e2e.sh test-docker: test-docker-build - @echo "\n....Running Single-Node E2E Tests...." + @echo -e "\n....Running Single-Node E2E Tests...." docker run --privileged --rm -v /dev/kvm:/dev/kvm $(E2E_IMAGE) - @echo "\n....Running Multi-Node E2E Tests...." + @echo -e "\n....Running Multi-Node E2E Tests...." docker run --privileged --rm -v /dev/kvm:/dev/kvm --cap-add=NET_ADMIN $(E2E_IMAGE) ./tests/e2e/run-multinode-e2e.sh .PHONY: build build-ui go_build go_run preflight test bench run clean \ diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 6ca22ccd3..c1d9a153d 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -246,10 +246,12 @@ func generateInstanceTypes(family, arch string) map[string]*ec2.InstanceTypeInfo {"2xlarge", 8, 32.0}, } - // Determine the burstable family based on architecture + // Determine the burstable family based on architecture and CPU vendor burstableFamily := "t3" if arch == "arm64" { burstableFamily = "t4g" + } else if strings.HasSuffix(family, "a") { + burstableFamily = "t3a" } // Build list of families to generate: CPU-specific + burstable (if different) @@ -328,6 +330,8 @@ func NewResourceManager() *ResourceManager { burstableFamily := "t3" if runtime.GOARCH == "arm64" { burstableFamily = "t4g" + } else if strings.HasSuffix(instanceFamily, "a") { + burstableFamily = "t3a" } slog.Info("System resources detected", "vCPUs", numCPU, "memGB", totalMemGB, "cpu", cpuModel, @@ -592,8 +596,12 @@ func (d *Daemon) Start() error { d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store) d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn) - d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn) - d.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImpl(d.config, d.natsConn) + snapSvc, snapshotKV, err := handlers_ec2_snapshot.NewSnapshotServiceImplWithNATS(d.config, d.natsConn) + if err != nil { + return fmt.Errorf("failed to initialize snapshot service with NATS KV: %w", err) + } + d.snapshotService = snapSvc + d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn, snapshotKV) d.tagsService = handlers_ec2_tags.NewTagsServiceImpl(d.config) if eigwSvc, err := handlers_ec2_eigw.NewEgressOnlyIGWServiceImplWithNATS(d.config, d.natsConn); err != nil { @@ -746,7 +754,7 @@ func (d *Daemon) checkPredastoreReady() bool { if err != nil { return false } - conn.Close() + _ = conn.Close() return true } diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 9bee86b6b..2a5d81118 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -76,7 +76,7 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { // Initialize services (needed for handler tests) daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore()) - daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(cfg, nc) + daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(cfg, nc, nil) t.Cleanup(func() { if daemon.natsConn != nil { @@ -3256,6 +3256,28 @@ func TestGenerateInstanceTypes_SingleFamily(t *testing.T) { } } +func TestGenerateInstanceTypes_AMDBurstable(t *testing.T) { + types := generateInstanceTypes("m8a", "x86_64") + assert.Len(t, types, 14) // 7 m8a + 7 t3a + + m8aCount, t3aCount := 0, 0 + for name := range types { + if strings.HasPrefix(name, "m8a.") { + m8aCount++ + } + if strings.HasPrefix(name, "t3a.") { + t3aCount++ + } + } + assert.Equal(t, 7, m8aCount) + assert.Equal(t, 7, t3aCount) + + // Verify t3 (Intel burstable) is NOT present + for name := range types { + assert.False(t, strings.HasPrefix(name, "t3."), "AMD should get t3a, not t3: %s", name) + } +} + func TestGenerateInstanceTypes_ARM64(t *testing.T) { types := generateInstanceTypes("m8g", "arm64") // m8g is not t4g, so we should get both families diff --git a/hive/handlers/ec2/snapshot/service_impl.go b/hive/handlers/ec2/snapshot/service_impl.go index f067d3ea8..1f6312274 100644 --- a/hive/handlers/ec2/snapshot/service_impl.go +++ b/hive/handlers/ec2/snapshot/service_impl.go @@ -24,11 +24,14 @@ import ( // Ensure SnapshotServiceImpl implements SnapshotService var _ SnapshotService = (*SnapshotServiceImpl)(nil) +const KVBucketVolumeSnapshots = "hive-volume-snapshots" + // SnapshotServiceImpl implements SnapshotService with S3-backed storage type SnapshotServiceImpl struct { config *config.Config store objectstore.ObjectStore natsConn *nats.Conn + snapKV nats.KeyValue mutex sync.RWMutex } @@ -47,8 +50,8 @@ type SnapshotConfig struct { Tags map[string]string `json:"tags"` } -// NewSnapshotServiceImpl creates a new snapshot service implementation -func NewSnapshotServiceImpl(cfg *config.Config, natsConn *nats.Conn) *SnapshotServiceImpl { +// NewSnapshotServiceImplWithNATS creates a snapshot service with JetStream KV for volume-snapshot tracking +func NewSnapshotServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*SnapshotServiceImpl, nats.KeyValue, error) { store := objectstore.NewS3ObjectStoreFromConfig( cfg.Predastore.Host, cfg.Predastore.Region, @@ -56,20 +59,52 @@ func NewSnapshotServiceImpl(cfg *config.Config, natsConn *nats.Conn) *SnapshotSe cfg.Predastore.SecretKey, ) + js, err := natsConn.JetStream() + if err != nil { + return nil, nil, fmt.Errorf("failed to get JetStream context: %w", err) + } + + kv, err := getOrCreateKVBucket(js, KVBucketVolumeSnapshots, 10) + if err != nil { + return nil, nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVolumeSnapshots, err) + } + + slog.Info("Snapshot service initialized with JetStream KV", "bucket", KVBucketVolumeSnapshots) + return &SnapshotServiceImpl{ config: cfg, store: store, natsConn: natsConn, - } + snapKV: kv, + }, kv, nil } -// NewSnapshotServiceImplWithStore creates a snapshot service with a custom ObjectStore (for testing) -func NewSnapshotServiceImplWithStore(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn) *SnapshotServiceImpl { - return &SnapshotServiceImpl{ +// NewSnapshotServiceImplWithStore creates a snapshot service with a custom ObjectStore (for testing). +// An optional snapshotKV can be provided for KV-backed volume-snapshot tracking. +func NewSnapshotServiceImplWithStore(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn, snapshotKV ...nats.KeyValue) *SnapshotServiceImpl { + svc := &SnapshotServiceImpl{ config: cfg, store: store, natsConn: natsConn, } + if len(snapshotKV) > 0 { + svc.snapKV = snapshotKV[0] + } + return svc +} + +func getOrCreateKVBucket(js nats.JetStreamContext, bucketName string, history int) (nats.KeyValue, error) { + kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: bucketName, + History: utils.SafeIntToUint8(history), + }) + if err != nil { + kv, err = js.KeyValue(bucketName) + if err != nil { + return nil, err + } + } + return kv, nil } // getSnapshotKey returns the S3 key for storing snapshot metadata. @@ -197,22 +232,25 @@ func (s *SnapshotServiceImpl) CreateSnapshot(input *ec2.CreateSnapshotInput) (*e } msg, err := s.natsConn.Request(fmt.Sprintf("ebs.snapshot.%s", volumeID), snapData, 30*time.Second) - if err != nil { + if errors.Is(err, nats.ErrNoResponders) { + // Volume is not mounted — data is already persisted to S3, proceed with metadata-only snapshot. + slog.Info("CreateSnapshot: volume not mounted, creating metadata-only snapshot", "volumeId", volumeID, "snapshotId", snapshotID) + } else if err != nil { slog.Error("CreateSnapshot: ebs.snapshot NATS request failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) return nil, errors.New(awserrors.ErrorServerInternal) - } + } else { + var snapResp config.EBSSnapshotResponse + if err := json.Unmarshal(msg.Data, &snapResp); err != nil { + slog.Error("CreateSnapshot: failed to unmarshal ebs.snapshot response", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if !snapResp.Success || snapResp.Error != "" { + slog.Error("CreateSnapshot: viperblock snapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", snapResp.Error) + return nil, errors.New(awserrors.ErrorServerInternal) + } - var snapResp config.EBSSnapshotResponse - if err := json.Unmarshal(msg.Data, &snapResp); err != nil { - slog.Error("CreateSnapshot: failed to unmarshal ebs.snapshot response", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) - return nil, errors.New(awserrors.ErrorServerInternal) + slog.Info("CreateSnapshot: viperblock snapshot created", "volumeId", volumeID, "snapshotId", snapshotID) } - if !snapResp.Success || snapResp.Error != "" { - slog.Error("CreateSnapshot: viperblock snapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", snapResp.Error) - return nil, errors.New(awserrors.ErrorServerInternal) - } - - slog.Info("CreateSnapshot: viperblock snapshot created", "volumeId", volumeID, "snapshotId", snapshotID) } else { slog.Warn("CreateSnapshot: natsConn is nil, skipping viperblock snapshot (metadata-only)", "volumeId", volumeID) } @@ -246,8 +284,16 @@ func (s *SnapshotServiceImpl) CreateSnapshot(input *ec2.CreateSnapshotInput) (*e } } + // Track the volume→snapshot dependency in KV before persisting to S3. + // This ensures we never have an untracked snapshot in S3. + if err := s.addSnapshotRef(volumeID, snapshotID); err != nil { + slog.Error("CreateSnapshot failed to add snapshot ref to KV", "snapshotId", snapshotID, "volumeId", volumeID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if err := s.putSnapshotConfig(snapshotID, snapshotCfg); err != nil { slog.Error("CreateSnapshot failed to write config", "snapshotId", snapshotID, "err", err) + _ = s.removeSnapshotRef(volumeID, snapshotID) // best-effort cleanup return nil, errors.New(awserrors.ErrorServerInternal) } @@ -318,7 +364,7 @@ func (s *SnapshotServiceImpl) DeleteSnapshot(input *ec2.DeleteSnapshotInput) (*e slog.Info("DeleteSnapshot request", "snapshotId", snapshotID) - _, err := s.getSnapshotConfig(snapshotID) + cfg, err := s.getSnapshotConfig(snapshotID) if err != nil { slog.Error("DeleteSnapshot snapshot not found", "snapshotId", snapshotID, "err", err) return nil, err @@ -346,6 +392,12 @@ func (s *SnapshotServiceImpl) DeleteSnapshot(input *ec2.DeleteSnapshotInput) (*e } } + // Remove from KV after S3 cleanup. Failure is logged but not fatal — + // a phantom entry safely blocks volume deletion rather than allowing it. + if err := s.removeSnapshotRef(cfg.VolumeID, snapshotID); err != nil { + slog.Warn("DeleteSnapshot failed to remove snapshot ref from KV", "snapshotId", snapshotID, "volumeId", cfg.VolumeID, "err", err) + } + slog.Info("DeleteSnapshot completed", "snapshotId", snapshotID) return &ec2.DeleteSnapshotOutput{}, nil @@ -391,8 +443,15 @@ func (s *SnapshotServiceImpl) CopySnapshot(input *ec2.CopySnapshotInput) (*ec2.C newCfg.Tags[k] = v } + // Track the volume→snapshot dependency in KV before persisting to S3. + if err := s.addSnapshotRef(sourceCfg.VolumeID, newSnapshotID); err != nil { + slog.Error("CopySnapshot failed to add snapshot ref to KV", "snapshotId", newSnapshotID, "volumeId", sourceCfg.VolumeID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if err := s.putSnapshotConfig(newSnapshotID, newCfg); err != nil { slog.Error("CopySnapshot failed to write config", "snapshotId", newSnapshotID, "err", err) + _ = s.removeSnapshotRef(sourceCfg.VolumeID, newSnapshotID) // best-effort cleanup return nil, errors.New(awserrors.ErrorServerInternal) } @@ -402,3 +461,137 @@ func (s *SnapshotServiceImpl) CopySnapshot(input *ec2.CopySnapshotInput) (*ec2.C SnapshotId: aws.String(newSnapshotID), }, nil } + +// addSnapshotRef adds snapshotID to the volume's snapshot list in KV. +// Uses CAS (Create/Update with revision) to prevent lost updates under concurrency. +func (s *SnapshotServiceImpl) addSnapshotRef(volumeID, snapshotID string) error { + if s.snapKV == nil { + slog.Debug("addSnapshotRef: snapshotKV is nil, skipping", "volumeId", volumeID, "snapshotId", snapshotID) + return nil + } + + const maxRetries = 5 + for attempt := range maxRetries { + entry, err := s.snapKV.Get(volumeID) + var snapshots []string + + if err != nil { + if !errors.Is(err, nats.ErrKeyNotFound) { + return fmt.Errorf("addSnapshotRef: failed to get KV key %s: %w", volumeID, err) + } + // Key doesn't exist yet — create with just this snapshot + data, err := json.Marshal([]string{snapshotID}) + if err != nil { + return fmt.Errorf("addSnapshotRef: failed to marshal snapshot list: %w", err) + } + if _, err := s.snapKV.Create(volumeID, data); err != nil { + if attempt < maxRetries-1 { + continue // concurrent Create/Update — retry + } + return fmt.Errorf("addSnapshotRef: failed to create KV key %s: %w", volumeID, err) + } + slog.Info("addSnapshotRef: added snapshot ref", "volumeId", volumeID, "snapshotId", snapshotID) + return nil + } + + if err := json.Unmarshal(entry.Value(), &snapshots); err != nil { + return fmt.Errorf("addSnapshotRef: failed to unmarshal KV value for %s: %w", volumeID, err) + } + + snapshots = append(snapshots, snapshotID) + + data, err := json.Marshal(snapshots) + if err != nil { + return fmt.Errorf("addSnapshotRef: failed to marshal snapshot list: %w", err) + } + + if _, err := s.snapKV.Update(volumeID, data, entry.Revision()); err != nil { + if attempt < maxRetries-1 { + continue // concurrent update — retry + } + return fmt.Errorf("addSnapshotRef: failed to update KV key %s: %w", volumeID, err) + } + + slog.Info("addSnapshotRef: added snapshot ref", "volumeId", volumeID, "snapshotId", snapshotID) + return nil + } + + return fmt.Errorf("addSnapshotRef: exhausted retries for KV key %s", volumeID) +} + +// removeSnapshotRef removes snapshotID from the volume's snapshot list in KV. +// Deletes the key if the list becomes empty. +// Uses CAS (Update with revision) to prevent lost updates under concurrency. +func (s *SnapshotServiceImpl) removeSnapshotRef(volumeID, snapshotID string) error { + if s.snapKV == nil { + slog.Debug("removeSnapshotRef: snapshotKV is nil, skipping", "volumeId", volumeID, "snapshotId", snapshotID) + return nil + } + + const maxRetries = 5 + for attempt := range maxRetries { + entry, err := s.snapKV.Get(volumeID) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil + } + return fmt.Errorf("removeSnapshotRef: failed to get KV key %s: %w", volumeID, err) + } + + var snapshots []string + if err := json.Unmarshal(entry.Value(), &snapshots); err != nil { + return fmt.Errorf("removeSnapshotRef: failed to unmarshal KV value for %s: %w", volumeID, err) + } + + filtered := snapshots[:0] + for _, snap := range snapshots { + if snap != snapshotID { + filtered = append(filtered, snap) + } + } + + if len(filtered) == 0 { + if err := s.snapKV.Delete(volumeID); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + return fmt.Errorf("removeSnapshotRef: failed to delete KV key %s: %w", volumeID, err) + } + } else { + data, err := json.Marshal(filtered) + if err != nil { + return fmt.Errorf("removeSnapshotRef: failed to marshal snapshot list: %w", err) + } + if _, err := s.snapKV.Update(volumeID, data, entry.Revision()); err != nil { + if attempt < maxRetries-1 { + continue // concurrent update — retry + } + return fmt.Errorf("removeSnapshotRef: failed to update KV key %s: %w", volumeID, err) + } + } + + slog.Info("removeSnapshotRef: removed snapshot ref", "volumeId", volumeID, "snapshotId", snapshotID) + return nil + } + + return fmt.Errorf("removeSnapshotRef: exhausted retries for KV key %s", volumeID) +} + +// volumeHasSnapshots returns true if the volume has any snapshots in KV. +func (s *SnapshotServiceImpl) volumeHasSnapshots(volumeID string) (bool, error) { + if s.snapKV == nil { + return false, nil + } + + entry, err := s.snapKV.Get(volumeID) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return false, nil + } + return false, fmt.Errorf("volumeHasSnapshots: failed to get KV key %s: %w", volumeID, err) + } + + var snapshots []string + if err := json.Unmarshal(entry.Value(), &snapshots); err != nil { + return false, fmt.Errorf("volumeHasSnapshots: failed to unmarshal KV value for %s: %w", volumeID, err) + } + + return len(snapshots) > 0, nil +} diff --git a/hive/handlers/ec2/snapshot/service_impl_test.go b/hive/handlers/ec2/snapshot/service_impl_test.go index 81b52fff6..07e00426f 100644 --- a/hive/handlers/ec2/snapshot/service_impl_test.go +++ b/hive/handlers/ec2/snapshot/service_impl_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "testing" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" @@ -12,6 +13,8 @@ import ( "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/viperblock/viperblock" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -349,3 +352,209 @@ func TestCopySnapshot_PreservesTags(t *testing.T) { assert.Equal(t, "Environment", *result.Snapshots[0].Tags[0].Key) assert.Equal(t, "test", *result.Snapshots[0].Tags[0].Value) } + +// setupTestNATSKV creates a NATS JetStream test server and returns a KV bucket for testing. +func setupTestNATSKV(t *testing.T) nats.KeyValue { + t.Helper() + opts := &server.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: true, + StoreDir: t.TempDir(), + NoLog: true, + NoSigs: true, + } + ns, err := server.NewServer(opts) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(5*time.Second)) + t.Cleanup(func() { ns.Shutdown() }) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(func() { nc.Close() }) + + js, err := nc.JetStream() + require.NoError(t, err) + + kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: KVBucketVolumeSnapshots, + }) + require.NoError(t, err) + return kv +} + +func TestAddSnapshotRef(t *testing.T) { + kv := setupTestNATSKV(t) + svc := &SnapshotServiceImpl{snapKV: kv} + + require.NoError(t, svc.addSnapshotRef("vol-1", "snap-a")) + require.NoError(t, svc.addSnapshotRef("vol-1", "snap-b")) + + entry, err := kv.Get("vol-1") + require.NoError(t, err) + var snapshots []string + require.NoError(t, json.Unmarshal(entry.Value(), &snapshots)) + assert.Equal(t, []string{"snap-a", "snap-b"}, snapshots) +} + +func TestRemoveSnapshotRef(t *testing.T) { + kv := setupTestNATSKV(t) + svc := &SnapshotServiceImpl{snapKV: kv} + + require.NoError(t, svc.addSnapshotRef("vol-1", "snap-a")) + require.NoError(t, svc.addSnapshotRef("vol-1", "snap-b")) + + // Remove one + require.NoError(t, svc.removeSnapshotRef("vol-1", "snap-a")) + + entry, err := kv.Get("vol-1") + require.NoError(t, err) + var snapshots []string + require.NoError(t, json.Unmarshal(entry.Value(), &snapshots)) + assert.Equal(t, []string{"snap-b"}, snapshots) + + // Remove last — key should be deleted + require.NoError(t, svc.removeSnapshotRef("vol-1", "snap-b")) + + _, err = kv.Get("vol-1") + assert.ErrorIs(t, err, nats.ErrKeyNotFound) +} + +func TestRemoveSnapshotRef_NonExistentKey(t *testing.T) { + kv := setupTestNATSKV(t) + svc := &SnapshotServiceImpl{snapKV: kv} + + // Should not error on non-existent key + require.NoError(t, svc.removeSnapshotRef("vol-nonexistent", "snap-x")) +} + +func TestVolumeHasSnapshots(t *testing.T) { + kv := setupTestNATSKV(t) + svc := &SnapshotServiceImpl{snapKV: kv} + + // No entry → false + has, err := svc.volumeHasSnapshots("vol-1") + require.NoError(t, err) + assert.False(t, has) + + // Add one → true + require.NoError(t, svc.addSnapshotRef("vol-1", "snap-a")) + has, err = svc.volumeHasSnapshots("vol-1") + require.NoError(t, err) + assert.True(t, has) + + // Remove it → false + require.NoError(t, svc.removeSnapshotRef("vol-1", "snap-a")) + has, err = svc.volumeHasSnapshots("vol-1") + require.NoError(t, err) + assert.False(t, has) +} + +func TestKVNilFallback(t *testing.T) { + svc := &SnapshotServiceImpl{snapKV: nil} + + // All methods should be no-ops when KV is nil + require.NoError(t, svc.addSnapshotRef("vol-1", "snap-a")) + require.NoError(t, svc.removeSnapshotRef("vol-1", "snap-a")) + has, err := svc.volumeHasSnapshots("vol-1") + require.NoError(t, err) + assert.False(t, has) +} + +func TestCreateSnapshot_WritesKVEntry(t *testing.T) { + kv := setupTestNATSKV(t) + store := objectstore.NewMemoryObjectStore() + cfg := &config.Config{ + Predastore: config.PredastoreConfig{ + Bucket: "test-bucket", + AccessKey: "test-owner-123", + }, + } + svc := NewSnapshotServiceImplWithStore(cfg, store, nil, kv) + + volumeID := "vol-kvtest" + createTestVolume(t, store, volumeID, 10) + + snap, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{ + VolumeId: aws.String(volumeID), + }) + require.NoError(t, err) + + // Verify KV entry exists + has, err := svc.volumeHasSnapshots(volumeID) + require.NoError(t, err) + assert.True(t, has) + + // Verify snapshot ID is in the list + entry, err := kv.Get(volumeID) + require.NoError(t, err) + var snapshots []string + require.NoError(t, json.Unmarshal(entry.Value(), &snapshots)) + assert.Contains(t, snapshots, *snap.SnapshotId) +} + +func TestDeleteSnapshot_RemovesKVEntry(t *testing.T) { + kv := setupTestNATSKV(t) + store := objectstore.NewMemoryObjectStore() + cfg := &config.Config{ + Predastore: config.PredastoreConfig{ + Bucket: "test-bucket", + AccessKey: "test-owner-123", + }, + } + svc := NewSnapshotServiceImplWithStore(cfg, store, nil, kv) + + volumeID := "vol-kvdelete" + createTestVolume(t, store, volumeID, 10) + + snap, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{ + VolumeId: aws.String(volumeID), + }) + require.NoError(t, err) + + // Delete the snapshot + _, err = svc.DeleteSnapshot(&ec2.DeleteSnapshotInput{ + SnapshotId: snap.SnapshotId, + }) + require.NoError(t, err) + + // KV should now be empty for this volume + has, err := svc.volumeHasSnapshots(volumeID) + require.NoError(t, err) + assert.False(t, has) +} + +func TestCopySnapshot_AddsKVEntry(t *testing.T) { + kv := setupTestNATSKV(t) + store := objectstore.NewMemoryObjectStore() + cfg := &config.Config{ + Predastore: config.PredastoreConfig{ + Bucket: "test-bucket", + AccessKey: "test-owner-123", + }, + } + svc := NewSnapshotServiceImplWithStore(cfg, store, nil, kv) + + volumeID := "vol-kvcopy" + createTestVolume(t, store, volumeID, 10) + + snap, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{ + VolumeId: aws.String(volumeID), + }) + require.NoError(t, err) + + copyResult, err := svc.CopySnapshot(&ec2.CopySnapshotInput{ + SourceSnapshotId: snap.SnapshotId, + }) + require.NoError(t, err) + + // Both snapshot IDs should be in KV + entry, err := kv.Get(volumeID) + require.NoError(t, err) + var snapshots []string + require.NoError(t, json.Unmarshal(entry.Value(), &snapshots)) + assert.Contains(t, snapshots, *snap.SnapshotId) + assert.Contains(t, snapshots, *copyResult.SnapshotId) + assert.Len(t, snapshots, 2) +} diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go index 6704dd5d2..47b09e001 100644 --- a/hive/handlers/ec2/volume/service_impl.go +++ b/hive/handlers/ec2/volume/service_impl.go @@ -34,10 +34,13 @@ type VolumeServiceImpl struct { store objectstore.ObjectStore bucketName string natsConn *nats.Conn + snapshotKV nats.KeyValue } -// NewVolumeServiceImpl creates a new daemon-side volume service -func NewVolumeServiceImpl(cfg *config.Config, natsConn *nats.Conn) *VolumeServiceImpl { +// NewVolumeServiceImpl creates a new daemon-side volume service. +// snapshotKV is optional — when non-nil, DeleteVolume uses O(1) KV lookup +// instead of scanning all snapshots in S3. +func NewVolumeServiceImpl(cfg *config.Config, natsConn *nats.Conn, snapshotKV nats.KeyValue) *VolumeServiceImpl { store := objectstore.NewS3ObjectStoreFromConfig( cfg.Predastore.Host, cfg.Predastore.Region, @@ -50,21 +53,26 @@ func NewVolumeServiceImpl(cfg *config.Config, natsConn *nats.Conn) *VolumeServic store: store, bucketName: cfg.Predastore.Bucket, natsConn: natsConn, + snapshotKV: snapshotKV, } } // NewVolumeServiceImplWithStore creates a volume service with a custom ObjectStore (for testing) -func NewVolumeServiceImplWithStore(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn) *VolumeServiceImpl { +func NewVolumeServiceImplWithStore(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn, snapshotKV ...nats.KeyValue) *VolumeServiceImpl { bucketName := "" if cfg != nil { bucketName = cfg.Predastore.Bucket } - return &VolumeServiceImpl{ + svc := &VolumeServiceImpl{ config: cfg, store: store, bucketName: bucketName, natsConn: natsConn, } + if len(snapshotKV) > 0 { + svc.snapshotKV = snapshotKV[0] + } + return svc } // CreateVolume creates a new EBS volume via viperblock and persists its config to S3 @@ -705,9 +713,9 @@ func (s *VolumeServiceImpl) DeleteVolume(input *ec2.DeleteVolumeInput) (*ec2.Del return nil, errors.New(awserrors.ErrorVolumeInUse) } - // Check if any snapshots reference this volume. Snapshot-backed clones - // read chunk files from the source volume's S3 prefix via ReadFrom(). - // Deleting the source volume would silently break all clones. + // Check if any snapshots reference this volume via JetStream KV. + // Snapshot-backed clones read chunk files from the source volume's + // S3 prefix via ReadFrom(). Deleting the source would break all clones. if err := s.checkVolumeHasNoSnapshots(volumeID); err != nil { return nil, err } @@ -831,90 +839,40 @@ func (s *VolumeServiceImpl) getSnapshotMetadata(snapshotID string) (*snapshotMet return &meta, nil } -// snapshotVolumeRef reads VolumeID from hive metadata.json (volume_id) -// or viperblock config.json (SourceVolumeName). -type snapshotVolumeRef struct { - VolumeID string `json:"volume_id"` - SourceVolumeName string `json:"SourceVolumeName"` -} - -func (r snapshotVolumeRef) referencesVolume(volumeID string) bool { - return r.VolumeID == volumeID || r.SourceVolumeName == volumeID -} - -// checkVolumeHasNoSnapshots scans all snap-* prefixes in S3 and returns an -// error if any snapshot references the given volume. Checks both hive's -// metadata.json and viperblock's config.json to handle either format. +// checkVolumeHasNoSnapshots checks if a volume has dependent snapshots +// using the JetStream KV index. func (s *VolumeServiceImpl) checkVolumeHasNoSnapshots(volumeID string) error { - listResult, err := s.store.ListObjects(&s3.ListObjectsInput{ - Bucket: aws.String(s.bucketName), - Prefix: aws.String("snap-"), - Delimiter: aws.String("/"), - }) - if err != nil { - slog.Error("checkVolumeHasNoSnapshots: failed to list snapshots", "volumeId", volumeID, "err", err) + if s.snapshotKV == nil { + slog.Error("checkVolumeHasNoSnapshots: snapshotKV is nil", "volumeId", volumeID) return errors.New(awserrors.ErrorServerInternal) } - for _, prefix := range listResult.CommonPrefixes { - if prefix.Prefix == nil { - continue - } - - snapshotID := strings.TrimSuffix(*prefix.Prefix, "/") - - if err := s.snapshotReferencesVolume(snapshotID, volumeID); err != nil { - return err - } + has, err := s.volumeHasSnapshotsKV(volumeID) + if err != nil { + slog.Error("checkVolumeHasNoSnapshots: KV lookup failed", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + if has { + slog.Error("DeleteVolume blocked: volume has snapshots", "volumeId", volumeID) + return errors.New(awserrors.ErrorVolumeInUse) } - return nil } -// snapshotReferencesVolume checks if a snapshot references the given volume. -// Returns an error if the snapshot references the volume or if a non-recoverable -// S3/IO error occurs. Returns nil if the snapshot does not reference the volume. -func (s *VolumeServiceImpl) snapshotReferencesVolume(snapshotID, volumeID string) error { - // Try metadata.json first (hive format), then config.json (viperblock format) - for _, filename := range []string{"metadata.json", "config.json"} { - key := snapshotID + "/" + filename - - getResult, err := s.store.GetObject(&s3.GetObjectInput{ - Bucket: aws.String(s.bucketName), - Key: aws.String(key), - }) - if err != nil { - if objectstore.IsNoSuchKeyError(err) { - continue // Expected: this format file doesn't exist for this snapshot - } - slog.Error("checkVolumeHasNoSnapshots: failed to read snapshot metadata", - "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - body, err := io.ReadAll(getResult.Body) - if closeErr := getResult.Body.Close(); closeErr != nil { - slog.Error("checkVolumeHasNoSnapshots: failed to close response body", - "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", closeErr) - } - if err != nil { - slog.Error("checkVolumeHasNoSnapshots: failed to read snapshot body", - "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - var ref snapshotVolumeRef - if err := json.Unmarshal(body, &ref); err != nil { - slog.Warn("checkVolumeHasNoSnapshots: failed to parse snapshot metadata, skipping file", - "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", err) - continue // JSON parse errors for individual files are not fatal -- try next file - } - - if ref.referencesVolume(volumeID) { - slog.Error("DeleteVolume blocked: volume has snapshots", "volumeId", volumeID, "snapshotId", snapshotID) - return fmt.Errorf("%s: volume %s has existing snapshot %s. Delete snapshots first", awserrors.ErrorVolumeInUse, volumeID, snapshotID) +// volumeHasSnapshotsKV checks the JetStream KV index for snapshot references. +func (s *VolumeServiceImpl) volumeHasSnapshotsKV(volumeID string) (bool, error) { + entry, err := s.snapshotKV.Get(volumeID) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return false, nil } + return false, err } - return nil + var snapshots []string + if err := json.Unmarshal(entry.Value(), &snapshots); err != nil { + return false, err + } + + return len(snapshots) > 0, nil } diff --git a/hive/handlers/ec2/volume/service_impl_test.go b/hive/handlers/ec2/volume/service_impl_test.go index 14a6833a8..09d769c03 100644 --- a/hive/handlers/ec2/volume/service_impl_test.go +++ b/hive/handlers/ec2/volume/service_impl_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "testing" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" @@ -12,6 +13,8 @@ import ( "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/viperblock/viperblock" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -271,98 +274,6 @@ func newTestVolumeServiceWithStore(az string, store *objectstore.MemoryObjectSto return NewVolumeServiceImplWithStore(cfg, store, nil) } -func TestDeleteVolume_BlockedBySnapshot(t *testing.T) { - store := objectstore.NewMemoryObjectStore() - svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) - - volumeID := "vol-test123" - - // Create volume config in store - volumeState := viperblock.VBState{ - VolumeConfig: viperblock.VolumeConfig{ - VolumeMetadata: viperblock.VolumeMetadata{ - VolumeID: volumeID, - SizeGiB: 10, - State: "available", - }, - }, - } - data, err := json.Marshal(volumeState) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String(volumeID + "/config.json"), - Body: strings.NewReader(string(data)), - }) - require.NoError(t, err) - - // Create a snapshot referencing this volume - snapCfg := snapshotVolumeRef{VolumeID: volumeID} - snapData, err := json.Marshal(snapCfg) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String("snap-abc123/config.json"), - Body: strings.NewReader(string(snapData)), - }) - require.NoError(t, err) - - // DeleteVolume should be blocked - _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ - VolumeId: aws.String(volumeID), - }) - require.Error(t, err) - assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) - assert.Contains(t, err.Error(), "snap-abc123") -} - -func TestDeleteVolume_AllowedWithoutSnapshots(t *testing.T) { - store := objectstore.NewMemoryObjectStore() - svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) - - volumeID := "vol-test456" - - // Create volume config in store - volumeState := viperblock.VBState{ - VolumeConfig: viperblock.VolumeConfig{ - VolumeMetadata: viperblock.VolumeMetadata{ - VolumeID: volumeID, - SizeGiB: 10, - State: "available", - }, - }, - } - data, err := json.Marshal(volumeState) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String(volumeID + "/config.json"), - Body: strings.NewReader(string(data)), - }) - require.NoError(t, err) - - // Create a snapshot referencing a DIFFERENT volume - snapCfg := snapshotVolumeRef{VolumeID: "vol-other"} - snapData, err := json.Marshal(snapCfg) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String("snap-xyz789/config.json"), - Body: strings.NewReader(string(snapData)), - }) - require.NoError(t, err) - - // DeleteVolume should succeed (no snapshots reference this volume) - _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ - VolumeId: aws.String(volumeID), - }) - require.NoError(t, err) -} - func TestCreateVolume_FromSnapshot_PassesValidation(t *testing.T) { store := objectstore.NewMemoryObjectStore() svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) @@ -502,100 +413,6 @@ func TestCreateVolume_FromSnapshot_SizeEqualToSnapshot(t *testing.T) { } } -func TestDeleteVolume_BlockedBySourceVolumeName(t *testing.T) { - store := objectstore.NewMemoryObjectStore() - svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) - - volumeID := "vol-srcname" - - // Create volume config - volumeState := viperblock.VBState{ - VolumeConfig: viperblock.VolumeConfig{ - VolumeMetadata: viperblock.VolumeMetadata{ - VolumeID: volumeID, - SizeGiB: 10, - State: "available", - }, - }, - } - data, err := json.Marshal(volumeState) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String(volumeID + "/config.json"), - Body: strings.NewReader(string(data)), - }) - require.NoError(t, err) - - // Snapshot references volume via SourceVolumeName (viperblock config.json format) - snapCfg := snapshotVolumeRef{SourceVolumeName: volumeID} - snapData, err := json.Marshal(snapCfg) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String("snap-vb001/config.json"), - Body: strings.NewReader(string(snapData)), - }) - require.NoError(t, err) - - // DeleteVolume should be blocked via SourceVolumeName match - _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ - VolumeId: aws.String(volumeID), - }) - require.Error(t, err) - assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) - assert.Contains(t, err.Error(), "snap-vb001") -} - -func TestDeleteVolume_BlockedByMetadataJson(t *testing.T) { - store := objectstore.NewMemoryObjectStore() - svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) - - volumeID := "vol-metablock" - - // Create volume config - volumeState := viperblock.VBState{ - VolumeConfig: viperblock.VolumeConfig{ - VolumeMetadata: viperblock.VolumeMetadata{ - VolumeID: volumeID, - SizeGiB: 10, - State: "available", - }, - }, - } - data, err := json.Marshal(volumeState) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String(volumeID + "/config.json"), - Body: strings.NewReader(string(data)), - }) - require.NoError(t, err) - - // Snapshot references volume via metadata.json (hive format, volume_id field) - snapCfg := snapshotVolumeRef{VolumeID: volumeID} - snapData, err := json.Marshal(snapCfg) - require.NoError(t, err) - - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String("snap-meta001/metadata.json"), - Body: strings.NewReader(string(snapData)), - }) - require.NoError(t, err) - - // DeleteVolume should be blocked via metadata.json match - _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ - VolumeId: aws.String(volumeID), - }) - require.Error(t, err) - assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) - assert.Contains(t, err.Error(), "snap-meta001") -} - func TestCreateVolume_FromSnapshot_CorruptMetadata(t *testing.T) { store := objectstore.NewMemoryObjectStore() svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) @@ -617,13 +434,39 @@ func TestCreateVolume_FromSnapshot_CorruptMetadata(t *testing.T) { require.Error(t, err) } -func TestDeleteVolume_CorruptSnapshotMetadata_SkipsFile(t *testing.T) { - store := objectstore.NewMemoryObjectStore() - svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) +// setupTestVolumeKV creates a NATS JetStream test server and returns a KV bucket. +func setupTestVolumeKV(t *testing.T) nats.KeyValue { + t.Helper() + opts := &server.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: true, + StoreDir: t.TempDir(), + NoLog: true, + NoSigs: true, + } + ns, err := server.NewServer(opts) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(5*time.Second)) + t.Cleanup(func() { ns.Shutdown() }) - volumeID := "vol-corruptsnap" + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(func() { nc.Close() }) - // Create volume config + js, err := nc.JetStream() + require.NoError(t, err) + + kv, err := js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: "hive-volume-snapshots", + }) + require.NoError(t, err) + return kv +} + +func createVolumeInStore(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string) { + t.Helper() volumeState := viperblock.VBState{ VolumeConfig: viperblock.VolumeConfig{ VolumeMetadata: viperblock.VolumeMetadata{ @@ -642,19 +485,59 @@ func TestDeleteVolume_CorruptSnapshotMetadata_SkipsFile(t *testing.T) { Body: strings.NewReader(string(data)), }) require.NoError(t, err) +} - // Put corrupt metadata.json (invalid JSON) -- should be skipped - _, err = store.PutObject(&s3.PutObjectInput{ - Bucket: aws.String("test-bucket"), - Key: aws.String("snap-corrupt/metadata.json"), - Body: strings.NewReader("not json{{{"), - }) +func TestDeleteVolume_BlockedByKV(t *testing.T) { + kv := setupTestVolumeKV(t) + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + svc.snapshotKV = kv + + volumeID := "vol-kvblocked" + createVolumeInStore(t, store, volumeID) + + // Put a snapshot ref in KV + data, err := json.Marshal([]string{"snap-001"}) + require.NoError(t, err) + _, err = kv.Put(volumeID, data) require.NoError(t, err) - // No valid config.json either, so no snapshot blocks this volume. - // DeleteVolume should succeed since corrupt metadata is skipped. + // DeleteVolume should be blocked _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ VolumeId: aws.String(volumeID), }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) +} + +func TestDeleteVolume_AllowedByKV(t *testing.T) { + kv := setupTestVolumeKV(t) + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + svc.snapshotKV = kv + + volumeID := "vol-kvallowed" + createVolumeInStore(t, store, volumeID) + + // No KV entry → delete allowed + _, err := svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) require.NoError(t, err) } + +func TestDeleteVolume_ErrorWhenKVNil(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + // snapshotKV is nil by default + + volumeID := "vol-nokvtest" + createVolumeInStore(t, store, volumeID) + + // Should fail because snapshotKV is nil + _, err := svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorServerInternal) +}