diff --git a/contracts/viperblockd/legacy/v1/README.md b/contracts/viperblockd/legacy/v1/README.md index 44ec05a69..2c0eeed5b 100644 --- a/contracts/viperblockd/legacy/v1/README.md +++ b/contracts/viperblockd/legacy/v1/README.md @@ -1,10 +1,10 @@ # Viperblockd legacy contracts, version 1 -This package owns the deployed, transitional request/reply protocol between -Spinifex's legacy control-plane EBS routes and `viperblockd`. It makes the -existing compatibility boundary explicit while the generic `providers/ebs` -contract is adopted; it is not the EBS provider abstraction and new provider -integrations must not depend on it. +This package owns the deployed, transitional configuration-update and +state-reload request/reply protocols between Spinifex's legacy control plane +and `viperblockd`. It makes those existing compatibility boundaries explicit +while the generic `providers/ebs` contract is adopted; it is not the EBS +provider abstraction and new provider integrations must not depend on it. ## Configuration update routes @@ -22,6 +22,19 @@ Both routes carry `EBSConfigUpdateRequest` and respond with request/reply semantics are deployed compatibility behaviour. Incompatible changes require a new contract version or an explicit migration plan. +## State reload route + +`SyncSubject` is `ebs.sync`. After a successful EC2 `ModifyVolume`, the daemon +uses request/reply to ask a `spinifex-workers` queue-group member to reload the +volume's state from its backend. The selected Viperblockd worker responds with +`EBSSyncResponse`; it returns an error when it does not own a live mount of the +volume. The daemon records that error as a warning after the API response has +already been returned. This is deployed legacy behaviour, not a guarantee that +every modification has reached its owning mount. + +The route carries `EBSSyncRequest` and `EBSSyncResponse`. Their JSON shape and +the queue-group route are compatibility behaviour. + This package does not authorize requests, validate a volume ID, store volume state, or define the generic storage-provider API. Those concerns remain with the consuming daemon, the EC2 domain, and `providers/ebs` respectively. diff --git a/contracts/viperblockd/legacy/v1/sync.go b/contracts/viperblockd/legacy/v1/sync.go new file mode 100644 index 000000000..867791510 --- /dev/null +++ b/contracts/viperblockd/legacy/v1/sync.go @@ -0,0 +1,22 @@ +package viperblocklegacyv1 + +const ( + // SyncSubject is the legacy queue-group route that asks a viperblockd + // worker to reload a mounted volume's state from its backend. + SyncSubject = "ebs.sync" +) + +// EBSSyncRequest identifies the volume whose mounted Viperblock state should +// be reloaded by the selected legacy worker. +type EBSSyncRequest struct { + Volume string `json:"Volume"` +} + +// EBSSyncResponse reports the selected worker's reload result. A worker that +// does not own the mounted volume returns an error rather than forwarding the +// request. +type EBSSyncResponse struct { + Volume string `json:"Volume"` + Synced bool `json:"Synced"` + Error string `json:"Error"` +} diff --git a/contracts/viperblockd/legacy/v1/sync_test.go b/contracts/viperblockd/legacy/v1/sync_test.go new file mode 100644 index 000000000..2bf4142be --- /dev/null +++ b/contracts/viperblockd/legacy/v1/sync_test.go @@ -0,0 +1,28 @@ +package viperblocklegacyv1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestSyncSubject(t *testing.T) { + require.Equal(t, "ebs.sync", SyncSubject) +} + +func TestSyncRequestJSONContract(t *testing.T) { + data, err := json.Marshal(EBSSyncRequest{Volume: "vol-123"}) + require.NoError(t, err) + require.JSONEq(t, `{"Volume":"vol-123"}`, string(data)) +} + +func TestSyncResponseJSONContract(t *testing.T) { + data, err := json.Marshal(EBSSyncResponse{ + Volume: "vol-123", + Synced: true, + Error: "", + }) + require.NoError(t, err) + require.JSONEq(t, `{"Volume":"vol-123","Synced":true,"Error":""}`, string(data)) +} diff --git a/docs/PACKAGE_BOUNDARY_MIGRATION.md b/docs/PACKAGE_BOUNDARY_MIGRATION.md index cbb5d41f2..9fb4270a7 100644 --- a/docs/PACKAGE_BOUNDARY_MIGRATION.md +++ b/docs/PACKAGE_BOUNDARY_MIGRATION.md @@ -191,6 +191,7 @@ while the branch is in flight; they do not describe the umbrella repository's | `08bd335f0` | `spinifex/types/node.go` (`NodeStatusResponse`, capacity and GPU inventory) | `contracts/cluster/v1/status.go` | Node status is a versioned fan-out observation contract for operator inspection and EC2/EKS scheduling consumers; `status_test.go` pins its subject and JSON shape. | | `f4cb550c9` | `spinifex/types/ebs.go` (GP3 policy constants) | `spinifex/domains/ec2/ebs/policy/gp3.go` | The supported GP3 product envelope is EC2 EBS domain policy, not generic runtime or Viperblockd wire state; `gp3_test.go` pins the deployed values. | | `adb55de33` | `spinifex/types/ebs.go` (`NBDTransport`) | `spinifex/services/viperblockd/nbd_transport.go` | NBD endpoint transport selection is Viperblockd-local delivery configuration, not generic EBS state or a provider contract; `nbd_transport_test.go` pins the socket and TCP forms. | +| `94a6e7af2` | `spinifex/types/ebs.go` (`EBSConfigUpdateRequest`, `EBSConfigUpdateResponse`) | `contracts/viperblockd/legacy/v1/config_update.go` | Legacy configuration updates have deployed queue and volume-addressed NATS routes; the versioned compatibility contract is expressly transitional and separate from `providers/ebs`. | The EC2 contract row is intentionally different from the directory moves: it creates a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS diff --git a/docs/service-interfaces.yaml b/docs/service-interfaces.yaml index d411a2402..69eb8b87e 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -169,9 +169,10 @@ services: - "vpc.dhcp.acquire" - "vpc.dhcp.release" # → viperblockd - # Legacy configuration-update payloads and routes are owned by - # contracts/viperblockd/legacy/v1; they are transitional and distinct - # from the generic providers/ebs protocol. + # The migrated legacy ebs.config and ebs.sync payloads and routes are + # owned by contracts/viperblockd/legacy/v1. They are transitional and + # distinct from the generic providers/ebs protocol; other legacy ebs.* + # routes remain to be migrated separately. - "ebs.mount" - "ebs.unmount" - "ebs.delete" @@ -227,8 +228,9 @@ services: # mulga/viperblock submodule. Submodule pointer bump → selector forces # "run all" (contract changes can be invisible at the spinifex source # level). - # The legacy ebs.config routes are versioned by - # contracts/viperblockd/legacy/v1, not by providers/ebs. + # The migrated legacy ebs.config and ebs.sync routes are versioned by + # contracts/viperblockd/legacy/v1, not by providers/ebs. Other legacy + # ebs.* routes remain to be migrated separately. subscribes: - "ebs.mount" - "ebs.unmount" diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index b321f9522..b6194299e 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -19,6 +19,7 @@ import ( clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/contracts/ec2/v1" + viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1" "github.com/mulgadc/spinifex/internal/testkit" "github.com/mulgadc/spinifex/internal/testkit/ebsfake" "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata" @@ -2675,7 +2676,7 @@ func TestHandleEC2ModifyVolume_Success(t *testing.T) { }) // Subscribe a dummy ebs.sync handler so the NATS Request doesn't time out - syncSub, err := daemon.natsConn.Subscribe("ebs.sync", func(msg *nats.Msg) { + syncSub, err := daemon.natsConn.Subscribe(viperblocklegacyv1.SyncSubject, func(msg *nats.Msg) { _ = msg.Respond([]byte(`{}`)) }) require.NoError(t, err) diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index ab561073a..6d9a15b1e 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -9,10 +9,10 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/contracts/ec2/v1" + viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" handlers_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/handlers/ec2/snapshot" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -241,11 +241,11 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) string { // Notify viperblockd to reload state after volume modification (e.g. resize) if modifyVolumeInput.VolumeId != nil { - syncData, err := json.Marshal(types.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId}) + syncData, err := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId}) if err != nil { slog.ErrorContext(ctx, "failed to marshal ebs.sync request", "volumeId", *modifyVolumeInput.VolumeId, "err", err) } else { - _, syncErr := d.natsConn.Request("ebs.sync", syncData, 5*time.Second) + _, syncErr := d.natsConn.Request(viperblocklegacyv1.SyncSubject, syncData, 5*time.Second) if syncErr != nil { slog.WarnContext(ctx, "ebs.sync notification failed (volume may not be mounted)", "volumeId", *modifyVolumeInput.VolumeId, "err", syncErr) diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 012292d47..729f80f53 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -895,20 +895,20 @@ func launchService(cfg *Config) (err error) { return fmt.Errorf("failed to subscribe to %s: %w", abandonTopic, err) } - if _, err := nc.QueueSubscribe("ebs.sync", "spinifex-workers", func(msg *nats.Msg) { + if _, err := nc.QueueSubscribe(viperblocklegacyv1.SyncSubject, "spinifex-workers", func(msg *nats.Msg) { ctx, span := utils.StartConsumerSpan(msg) defer span.End() slog.InfoContext(ctx, "Received ebs.sync message") - var syncRequest types.EBSSyncRequest + var syncRequest viperblocklegacyv1.EBSSyncRequest if err := json.Unmarshal(msg.Data, &syncRequest); err != nil { slog.ErrorContext(ctx, "Failed to unmarshal ebs.sync message", "err", err) utils.MarkSpanError(span, err) - respondJSON(msg, types.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)}) + respondJSON(msg, viperblocklegacyv1.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)}) return } - syncResponse := types.EBSSyncResponse{Volume: syncRequest.Volume} + syncResponse := viperblocklegacyv1.EBSSyncResponse{Volume: syncRequest.Volume} defer func() { if syncResponse.Error != "" { utils.MarkSpanError(span, errors.New(syncResponse.Error)) diff --git a/spinifex/services/viperblockd/viperblockd_handlers_test.go b/spinifex/services/viperblockd/viperblockd_handlers_test.go index 102eabe43..7d2625886 100644 --- a/spinifex/services/viperblockd/viperblockd_handlers_test.go +++ b/spinifex/services/viperblockd/viperblockd_handlers_test.go @@ -324,11 +324,11 @@ func TestIntegration_EBSSyncVolumeNotMounted(t *testing.T) { require.NoError(t, err) defer nc.Close() - reqData, _ := json.Marshal(types.EBSSyncRequest{Volume: "vol-not-here"}) - msg, err := nc.Request("ebs.sync", reqData, 3*time.Second) + reqData, _ := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: "vol-not-here"}) + msg, err := nc.Request(viperblocklegacyv1.SyncSubject, reqData, 3*time.Second) require.NoError(t, err) - var resp types.EBSSyncResponse + var resp viperblocklegacyv1.EBSSyncResponse require.NoError(t, json.Unmarshal(msg.Data, &resp)) assert.False(t, resp.Synced) assert.Contains(t, resp.Error, "not mounted") @@ -354,11 +354,11 @@ func TestIntegration_EBSSyncVolumeNoVBInstance(t *testing.T) { require.NoError(t, err) defer nc.Close() - reqData, _ := json.Marshal(types.EBSSyncRequest{Volume: "vol-no-vb"}) - msg, err := nc.Request("ebs.sync", reqData, 3*time.Second) + reqData, _ := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: "vol-no-vb"}) + msg, err := nc.Request(viperblocklegacyv1.SyncSubject, reqData, 3*time.Second) require.NoError(t, err) - var resp types.EBSSyncResponse + var resp viperblocklegacyv1.EBSSyncResponse require.NoError(t, json.Unmarshal(msg.Data, &resp)) assert.False(t, resp.Synced) assert.Contains(t, resp.Error, "not mounted") @@ -381,10 +381,10 @@ func TestIntegration_EBSSyncInvalidJSON(t *testing.T) { require.NoError(t, err) defer nc.Close() - msg, err := nc.Request("ebs.sync", []byte("garbage"), 3*time.Second) + msg, err := nc.Request(viperblocklegacyv1.SyncSubject, []byte("garbage"), 3*time.Second) require.NoError(t, err) - var resp types.EBSSyncResponse + var resp viperblocklegacyv1.EBSSyncResponse require.NoError(t, json.Unmarshal(msg.Data, &resp)) assert.Contains(t, resp.Error, "bad request:") } @@ -491,11 +491,11 @@ func TestIntegration_EBSSyncWithVBInstance(t *testing.T) { require.NoError(t, err) defer nc.Close() - reqData, _ := json.Marshal(types.EBSSyncRequest{Volume: "vol-sync-vb"}) - msg, err := nc.Request("ebs.sync", reqData, 3*time.Second) + reqData, _ := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: "vol-sync-vb"}) + msg, err := nc.Request(viperblocklegacyv1.SyncSubject, reqData, 3*time.Second) require.NoError(t, err) - var resp types.EBSSyncResponse + var resp viperblocklegacyv1.EBSSyncResponse require.NoError(t, json.Unmarshal(msg.Data, &resp)) assert.True(t, resp.Synced) assert.Empty(t, resp.Error) diff --git a/spinifex/types/ebs.go b/spinifex/types/ebs.go index 03e06c6a2..af5e10956 100644 --- a/spinifex/types/ebs.go +++ b/spinifex/types/ebs.go @@ -44,16 +44,6 @@ type EBSUnMountResponse struct { Reaped bool `json:"Reaped,omitempty"` } -type EBSSyncRequest struct { - Volume string `json:"Volume"` -} - -type EBSSyncResponse struct { - Volume string `json:"Volume"` - Synced bool `json:"Synced"` - Error string `json:"Error"` -} - type EBSDeleteRequest struct { Volume string `json:"Volume"` }