orca/internal/transport/dispatch.go

// Package transport — dispatch.go implements the orca.v1.Dispatch
// service: a JSON-over-HTTP interface for cross-node job submission
// and status queries. Routes:
//
//	POST /orca.v1.Dispatch/Submit  ->  SubmitHandler
//	POST /orca.v1.Dispatch/Status  ->  StatusHandler
//
// mTLS is the v0.2 transport (P01). ConnectRPC is NOT used because
// it's not in go.mod (RESEARCH conclusion). The service is mounted on
// the orca daemon's mTLS listener (see internal/daemon/dispatch_handler.go).
package transport

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"time"
)

// SubmitRequest is the body of POST /orca.v1.Dispatch/Submit.
type SubmitRequest struct {
	Target         string          `json:"target"` // optional explicit node id; empty = bin-pack
	Spec           json.RawMessage `json:"spec"`   // HCL/YAML job spec, opaque to the dispatch service
	IdempotencyKey string          `json:"-"`      // set from X-Orca-Idempotency-Key header, not body
}

// SubmitResponse is the body of a Submit reply.
type SubmitResponse struct {
	JobID  string `json:"job_id"`
	NodeID string `json:"node_id"` // node that actually accepted the job (local or peer)
}

// StatusRequest is the body of POST /orca.v1.Dispatch/Status.
type StatusRequest struct {
	JobID string `json:"job_id"`
}

// StatusResponse is the body of a Status reply.
type StatusResponse struct {
	JobID  string `json:"job_id"`
	NodeID string `json:"node_id"`
	State  string `json:"state"` // "pending" | "running" | "complete" | "failed" | "stopped"
}

// Dispatcher is the contract the HTTP layer uses to actually run a
// job on a node. The engine layer implements this; the HTTP layer
// translates between JSON and Dispatcher calls.
type Dispatcher interface {
	LocalSubmit(ctx context.Context, spec []byte) (jobID string, err error)
	LocalStatus(ctx context.Context, jobID string) (state string, err error)
}

// SubmitHandler is an http.Handler that runs Submit on a local Dispatcher.
// It honors X-Orca-Idempotency-Key for dedupe. Errors are returned
// as JSON with an "error" field and an HTTP status code.
type SubmitHandler struct {
	Dispatcher Dispatcher
	Dedupe     *IdempotencyStore
}

// NewSubmitHandler builds a SubmitHandler.
func NewSubmitHandler(d Dispatcher, dedupe *IdempotencyStore) *SubmitHandler {
	if dedupe == nil {
		dedupe = NewIdempotencyStore()
	}
	return &SubmitHandler{Dispatcher: d, Dedupe: dedupe}
}

// ServeHTTP implements http.Handler.
func (h *SubmitHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
		return
	}
	defer r.Body.Close()
	var req SubmitRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "decode body: "+err.Error())
		return
	}
	if len(req.Spec) == 0 {
		writeError(w, http.StatusBadRequest, "spec is required")
		return
	}
	req.IdempotencyKey = r.Header.Get(IdempotencyHeader)

	// Idempotency check.
	if req.IdempotencyKey != "" {
		if jobID, ok := h.Dedupe.Get(req.IdempotencyKey); ok {
			// Replay the previous response.
			writeJSON(w, http.StatusOK, SubmitResponse{JobID: jobID, NodeID: ""})
			return
		}
	}

	jobID, err := h.Dispatcher.LocalSubmit(r.Context(), req.Spec)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	if req.IdempotencyKey != "" {
		h.Dedupe.Put(req.IdempotencyKey, jobID)
	}
	writeJSON(w, http.StatusOK, SubmitResponse{JobID: jobID, NodeID: "self"})
}

// StatusHandler is an http.Handler that runs Status on a local Dispatcher.
type StatusHandler struct {
	Dispatcher Dispatcher
}

// NewStatusHandler builds a StatusHandler.
func NewStatusHandler(d Dispatcher) *StatusHandler {
	return &StatusHandler{Dispatcher: d}
}

// ServeHTTP implements http.Handler.
func (h *StatusHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
		return
	}
	defer r.Body.Close()
	var req StatusRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "decode body: "+err.Error())
		return
	}
	if req.JobID == "" {
		writeError(w, http.StatusBadRequest, "job_id is required")
		return
	}
	state, err := h.Dispatcher.LocalStatus(r.Context(), req.JobID)
	if err != nil {
		writeError(w, http.StatusNotFound, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, StatusResponse{JobID: req.JobID, NodeID: "self", State: state})
}

// DispatchClient is the client-side wrapper that calls Submit/Status
// on a remote peer. It uses mTLS (REQ-011) and the retry helper
// (REQ-037).
type DispatchClient struct {
	HTTP     *MTLSClient
	PeerAddr string // http://host:port or https://host:port
}

// NewDispatchClient builds a DispatchClient for a peer.
func NewDispatchClient(caPath, serverName, peerAddr string) (*DispatchClient, error) {
	c, err := NewMTLSClient(caPath, serverName, "", "")
	if err != nil {
		return nil, fmt.Errorf("NewDispatchClient: %w", err)
	}
	return &DispatchClient{HTTP: c, PeerAddr: peerAddr}, nil
}

// Submit calls POST /orca.v1.Dispatch/Submit on the peer with the
// given spec and idempotency key. Retries per the default policy.
func (c *DispatchClient) Submit(ctx context.Context, spec []byte, idempotencyKey string) (*SubmitResponse, error) {
	if idempotencyKey != "" {
		ctx = WithIdempotencyKey(ctx, idempotencyKey)
	}
	body, _ := json.Marshal(SubmitRequest{Spec: spec})
	policy := DefaultRetryPolicy()
	for attempt := 1; attempt <= policy.MaxAttempts; attempt++ {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		req, _ := http.NewRequestWithContext(ctx, http.MethodPost, c.PeerAddr+"/orca.v1.Dispatch/Submit", bytesReader(body))
		req.Header.Set("Content-Type", "application/json")
		if k := IdempotencyKeyFromContext(ctx); k != "" {
			req.Header.Set(IdempotencyHeader, k)
		}
		r, err := c.HTTP.Do(req)
		if err == nil {
			defer r.Body.Close()
			if r.StatusCode == http.StatusOK {
				var resp SubmitResponse
				if derr := json.NewDecoder(r.Body).Decode(&resp); derr == nil {
					return &resp, nil
				} else {
					return nil, fmt.Errorf("DispatchClient.Submit: decode: %w", derr)
				}
			}
			err = fmt.Errorf("status %d", r.StatusCode)
			err = fmt.Errorf("%w: %v", ErrTransient, err)
		} else {
			err = fmt.Errorf("%w: %v", ErrTransient, err)
		}
		// No key, not idempotent: bail on first transient error.
		if IdempotencyKeyFromContext(ctx) == "" {
			return nil, err
		}
		if attempt == policy.MaxAttempts {
			return nil, err
		}
		// Wait with backoff, respecting ctx.
		wait := backoff(policy.Initial, policy.Max, attempt)
		t := time.NewTimer(wait)
		select {
		case <-ctx.Done():
			t.Stop()
			return nil, ctx.Err()
		case <-t.C:
		}
	}
	return nil, fmt.Errorf("DispatchClient.Submit: exhausted attempts")
}

// Status calls POST /orca.v1.Dispatch/Status on the peer. Status is
// idempotent at the verb level, so retries are always safe.
func (c *DispatchClient) Status(ctx context.Context, jobID string) (*StatusResponse, error) {
	body, _ := json.Marshal(StatusRequest{JobID: jobID})
	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, c.PeerAddr+"/orca.v1.Dispatch/Status", bytesReader(body))
	req.Header.Set("Content-Type", "application/json")
	r, err := c.HTTP.Do(req)
	if err != nil {
		return nil, fmt.Errorf("DispatchClient.Status: %w", err)
	}
	defer r.Body.Close()
	if r.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("DispatchClient.Status: status %d", r.StatusCode)
	}
	var resp StatusResponse
	if err := json.NewDecoder(r.Body).Decode(&resp); err != nil {
		return nil, fmt.Errorf("DispatchClient.Status: decode: %w", err)
	}
	return &resp, nil
}

// writeJSON encodes v as JSON and writes it with the given status.
func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}

// writeError writes a JSON error response.
func writeError(w http.ResponseWriter, status int, msg string) {
	writeJSON(w, status, map[string]string{"error": msg})
}

// bytesReader is a small helper to keep this file self-contained.
type bytesReadCloser struct {
	b   []byte
	pos int
}

func bytesReader(b []byte) *bytesReadCloser { return &bytesReadCloser{b: b} }

func (r *bytesReadCloser) Read(p []byte) (int, error) {
	if r.pos >= len(r.b) {
		return 0, io.EOF
	}
	n := copy(p, r.b[r.pos:])
	r.pos += n
	return n, nil
}

func (r *bytesReadCloser) Close() error { return nil }