parity: per-service AWS emulation deepening (DO NOT auto-merge — for review)

.github/workflows/ci.yml

@@ -18,6 +18,13 @@ permissions:
   security-events: write
   packages: write
 
+# Only the latest commit per PR (or ref) needs CI. A new push cancels any
+# still-running CI for the same PR so rapid commit bursts don't pile up runs;
+# CI settles on the last commit, which is all that matters.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.event.workflow_run.head_branch || github.ref }}
+  cancel-in-progress: true
+
 env:
   GH_CI_TOKEN: ${{ secrets.GH_TOKEN != '' && secrets.GH_TOKEN || github.token }}
 

cli.go

@@ -2693,6 +2693,10 @@ func initializeServices(appCtx *service.AppContext) ([]service.Registerable, err
 	// Wire Firehose → S3 and Lambda for actual record delivery and transformation.
 	wireFirehoseDelivery(byName["Firehose"], byName["S3"], byName["Lambda"])
 
+	// Wire DynamoDB → S3 so ImportTable reads source objects and
+	// ExportTableToPointInTime writes real export data.
+	wireDynamoDBS3(byName["DynamoDB"], byName["S3"])
+
 	// Wire Lambda invoker → SecretsManager rotation.
 	wireSecretsManagerLambda(byName["SecretsManager"], byName["Lambda"])
 
@@ -5080,6 +5084,30 @@ func wireFirehoseDelivery(firehoseReg, s3Reg, lambdaReg service.Registerable) {
 	}
 }
 
+// wireDynamoDBS3 connects the DynamoDB backend to the S3 backend so that
+// ImportTable can read source objects and ExportTableToPointInTime can write
+// export data to S3.
+func wireDynamoDBS3(ddbReg, s3Reg service.Registerable) {
+	ddbH, ok := ddbReg.(*ddbbackend.DynamoDBHandler)
+	if !ok {
+		return
+	}
+
+	s3H, s3Ok := s3Reg.(*s3backend.S3Handler)
+	if !s3Ok {
+		return
+	}
+
+	s3Bk, bkOk := s3H.Backend.(*s3backend.InMemoryBackend)
+	if !bkOk {
+		return
+	}
+
+	if ddbBk, ddbBkOk := ddbH.Backend.(*ddbbackend.InMemoryDB); ddbBkOk {
+		ddbBk.SetS3Backend(s3Bk)
+	}
+}
+
 // extractServiceName finds the service name for a given Echo context by checking
 // which service's route matcher matches the request.
 func extractServiceName(c *echo.Context, services []service.Registerable) string {

pkgs/awserr/awserr.go

@@ -3,6 +3,8 @@
 // to match any service error against a shared sentinel.
 package awserr
 
+import "fmt"
+
 // sentinelError is an unexported type used for constant sentinel errors.
 // Using a distinct type prevents reassignment and enables reliable [errors.Is] matching.
 type sentinelError string
@@ -28,6 +30,11 @@ func New(msg string, sentinel error) error {
 	return &wrappedError{msg: msg, cause: sentinel}
 }
 
+// Newf creates an error with a formatted message that wraps the given sentinel.
+func Newf(msg string, sentinel error, args ...any) error {
+	return &wrappedError{msg: fmt.Sprintf(msg, args...), cause: sentinel}
+}
+
 type wrappedError struct {
 	cause error
 	msg   string

pkgs/container/container.go

@@ -55,6 +55,8 @@ type Spec struct {
 	Mounts []string
 	// Cmd overrides the image's default CMD.
 	Cmd []string
+	// Entrypoint overrides the image's default ENTRYPOINT.
+	Entrypoint []string
 }
 
 // PooledContainer tracks a container managed by the warm pool.

pkgs/container/docker_runtime.go

@@ -200,10 +200,23 @@ func (r *DockerRuntime) CreateAndStart(ctx context.Context, spec Spec) (string,
 		cfg.Cmd = spec.Cmd
 	}
 
+	if len(spec.Entrypoint) > 0 {
+		cfg.Entrypoint = spec.Entrypoint
+	}
+
 	hostCfg := &dockercontainer.HostConfig{
 		Binds: spec.Mounts,
 	}
 
+	// Ensure the image is present before creating the container. Real AWS (and
+	// LocalStack) pull the runtime/base image on demand; without this a clean
+	// host fails container creation with "No such image".
+	if has, herr := r.HasImage(ctx, spec.Image); herr == nil && !has {
+		if perr := r.PullImage(ctx, spec.Image); perr != nil {
+			return "", fmt.Errorf("ensure image %q: %w", spec.Image, perr)
+		}
+	}
+
 	resp, err := r.docker.ContainerCreate(ctx, cfg, hostCfg, nil, nil, spec.Name)
 	if err != nil {
 		return "", fmt.Errorf("container create %q: %w", spec.Image, err)

pkgs/service/service.go

@@ -149,6 +149,7 @@ type FISActionDefinition struct {
 	ActionID    string // e.g., "aws:ec2:stop-instances"
 	Description string
 	TargetType  string // e.g., "aws:ec2:instance"; empty if action has no targets
+	TargetKey   string // key name used in the Targets map (e.g., "Instances", "Roles"); defaults to "Targets"
 	Parameters  []FISParamDef
 }
 

services/apigateway/backend.go

@@ -224,6 +224,10 @@ type StorageBackend interface {
 
 	// OpenAPI export.
 	GetExport(restAPIID, stageName, exportType string) (map[string]any, error)
+
+	// OpenAPI import.
+	ImportRestAPI(input ImportRestAPIInput) (*RestAPI, error)
+	PutRestAPI(input PutRestAPIInput) (*RestAPI, error)
 }
 
 const apiIDChars = "abcdefghijklmnopqrstuvwxyz0123456789"
@@ -265,7 +269,11 @@ const (
 	exportKeyBody        = "body"
 )
 
-const paramLocationHeader = "header"
+const (
+	paramLocationHeader = "header"
+	paramLocationPath   = "path"
+	paramLocationQuery  = "querystring"
+)
 
 // stageInvokeURL returns the gopherstack proxy path for a deployed stage.
 // The full URL is relative — clients prepend their gopherstack base URL.

services/apigateway/extra_coverage_test.go

@@ -492,6 +492,14 @@ func (n *noopBackend) GetExport(_ string, _ string, _ string) (map[string]any, e
 	return nil, errNoopNotImplemented
 }
 
+func (n *noopBackend) ImportRestAPI(_ apigateway.ImportRestAPIInput) (*apigateway.RestAPI, error) {
+	return nil, errNoopNotImplemented
+}
+
+func (n *noopBackend) PutRestAPI(_ apigateway.PutRestAPIInput) (*apigateway.RestAPI, error) {
+	return nil, errNoopNotImplemented
+}
+
 // restRequest sends a REST-style request (no X-Amz-Target header) to the handler.
 func restRequest(t *testing.T, handler *apigateway.Handler, method, path, body string) *httptest.ResponseRecorder {
 	t.Helper()

services/apigateway/handler.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"maps"
 	"net/http"
+	"net/url"
 	"strings"
 	"sync"
 	"time"
@@ -962,6 +963,17 @@ func (h *Handler) handleRESTAPI(c *echo.Context) error {
 		return c.String(http.StatusInternalServerError, "internal server error")
 	}
 
+	// OpenAPI import (ImportRestApi / PutRestApi) carries the raw spec document
+	// as the HTTP body. These are detected here because they share REST paths
+	// with CreateRestApi (POST /restapis) and UpdateRestApi (PUT /restapis/{id})
+	// but are distinguished by the request method/query, and the body must be
+	// passed through verbatim rather than treated as a flat field object.
+	if importAction, importBody, isImport := detectImportRESTAPI(
+		c.Request().Method, action, pathParams, c.Request().URL.Query(), body,
+	); isImport {
+		return h.dispatchAndRespond(ctx, c, importAction, importBody, contentTypeJSON)
+	}
+
 	// GET requests have no body; normalise to an empty JSON object so that
 	// json.Unmarshal calls in the action handlers don't fail with
 	// "unexpected end of JSON input".
@@ -986,19 +998,65 @@ func (h *Handler) handleRESTAPI(c *echo.Context) error {
 		}
 	}
 
+	return h.dispatchAndRespond(ctx, c, action, body, contentTypeJSON)
+}
+
+// dispatchAndRespond runs an action through the dispatch table and writes the
+// HTTP response, including correct handling of 204 No Content responses.
+func (h *Handler) dispatchAndRespond(
+	ctx context.Context, c *echo.Context, action string, body []byte, contentType string,
+) error {
 	statusCode, response, reqErr := h.dispatch(ctx, action, body)
 	if reqErr != nil {
 		return h.handleError(ctx, c, action, reqErr)
 	}
 
-	c.Response().Header().Set("Content-Type", contentTypeJSON)
+	c.Response().Header().Set("Content-Type", contentType)
 	if statusCode == http.StatusNoContent {
 		return c.NoContent(http.StatusNoContent)
 	}
 
 	return c.JSONBlob(statusCode, response)
 }
 
+// detectImportRESTAPI recognises ImportRestApi (POST /restapis?mode=import) and
+// PutRestApi (PUT /restapis/{id}) requests, returning the resolved action and a
+// JSON-encoded typed input whose Body field carries the raw spec document. The
+// AWS SDK sends the OpenAPI/Swagger document as the verbatim HTTP body, so it
+// must not be merged with path/query parameters like other operations.
+func detectImportRESTAPI(
+	method, action string, pathParams map[string]string, query url.Values, body []byte,
+) (string, []byte, bool) {
+	switch {
+	case action == opCreateRestAPI && method == http.MethodPost && query.Get("mode") == "import":
+		in := ImportRestAPIInput{
+			Body:           body,
+			FailOnWarnings: query.Get("failonwarnings") == litTrue,
+		}
+		encoded, err := json.Marshal(in)
+		if err != nil {
+			return "", nil, false
+		}
+
+		return opImportRestAPI, encoded, true
+	case action == opPutRestAPI && method == http.MethodPut && pathParams[keyRestAPIID] != "":
+		in := PutRestAPIInput{
+			RestAPIID:      pathParams[keyRestAPIID],
+			Mode:           query.Get("mode"),
+			FailOnWarnings: query.Get("failonwarnings") == litTrue,
+			Body:           body,
+		}
+		encoded, err := json.Marshal(in)
+		if err != nil {
+			return "", nil, false
+		}
+
+		return opPutRestAPI, encoded, true
+	}
+
+	return "", nil, false
+}
+
 // normalizePatchBody converts a JSON patch array (RFC 6902) to a flat JSON object.
 // AWS API Gateway REST PATCH endpoints accept patch operations like
 // [{"op":"replace","path":"/description","value":"foo"}].
@@ -1353,6 +1411,10 @@ func parseAPIGWRestAPIsDepth2(method, apiID string) (string, map[string]string,
 		return opDeleteRestAPI, params, true
 	case http.MethodPatch:
 		return opUpdateRestAPI, params, true
+	case http.MethodPut:
+		// PUT /restapis/{id} is PutRestApi (OpenAPI import into an existing
+		// API). The body is the raw spec; detectImportRESTAPI handles it.
+		return opPutRestAPI, params, true
 	}
 
 	return apiGWUnknownOp, nil, false

services/apigateway/handler_stubs.go

@@ -13,10 +13,6 @@ import (
 const (
 	// vpcLinkStatusAvailable is the status for an available VPC Link.
 	vpcLinkStatusAvailable = "AVAILABLE"
-	// stubImportedAPIName is the placeholder name for imported REST APIs.
-	stubImportedAPIName = "imported-api"
-	// stubImportedAPIID is the placeholder ID for imported REST APIs.
-	stubImportedAPIID = "stub0000"
 	// keyAPIName is the JSON key for API name in stub responses.
 	keyAPIName = "name"
 )
@@ -230,11 +226,31 @@ func (h *Handler) stubActions() map[string]actionFn {
 	actions[opImportDocumentationParts] = func(_ []byte) (int, any, error) {
 		return http.StatusOK, &documentationPartsImportStub{IDs: []string{}, Warnings: []string{}}, nil
 	}
-	actions[opImportRestAPI] = func(_ []byte) (int, any, error) {
-		return http.StatusCreated, map[string]any{"id": stubImportedAPIID, keyAPIName: stubImportedAPIName}, nil
+	actions[opImportRestAPI] = func(b []byte) (int, any, error) {
+		var input ImportRestAPIInput
+		if err := json.Unmarshal(b, &input); err != nil {
+			return 0, nil, err
+		}
+
+		api, err := h.Backend.ImportRestAPI(input)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		return http.StatusCreated, api, nil
 	}
-	actions[opPutRestAPI] = func(_ []byte) (int, any, error) {
-		return http.StatusOK, map[string]any{"id": stubImportedAPIID, keyAPIName: stubImportedAPIName}, nil
+	actions[opPutRestAPI] = func(b []byte) (int, any, error) {
+		var input PutRestAPIInput
+		if err := json.Unmarshal(b, &input); err != nil {
+			return 0, nil, err
+		}
+
+		api, err := h.Backend.PutRestAPI(input)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		return http.StatusOK, api, nil
 	}
 
 	// Usage update