feat(P10a): transactional plane (REQ-075, REQ-079; C-09, C-23)
internal/txn/txn.go: Bundle (desired-state + apply/verify/rollback scripts + signed manifest), RenderBundle (content-addressed txn-id), Stage (SCP to lead), Apply (idempotent + rollback on failure). scripts/orca-pull.sh: C-09 failure contract (idempotent, bounded retry, deterministic, structured syslog) + C-23 (cluster-wide vs ns-scoped --force distinction). internal/cli/txn.go: orca txn apply/list/show/rollback CLI. ---ci--- project: orca phase: 10a milestone: v0.11 status: execute ---/ci---
This commit is contained in:
Jon Chery
@@ -45,6 +45,10 @@ func resetCommandFlags() {
|
||||
restoreDryRun = false
|
||||
collectorRoot = "/"
|
||||
collectorDryRun = false
|
||||
txnApplyForce, txnApplyAckRisk, txnApplyYes = false, false, false
|
||||
txnApplyNamespace = ""
|
||||
txnApplyTimeout = 5 * time.Minute
|
||||
txnApplyLead, txnRollbackLead = "", ""
|
||||
resetNSFlags()
|
||||
// Reset per-command output writers so tests that polluted them
|
||||
// (e.g. daemon tests calling cmd.SetOut(&buf)) don't leak into
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
// Package cli: txn.go implements the `orca txn` subcommand family
|
||||
// (P10a, v0.11; REQ-075, REQ-079; gates C-09, C-23). Subcommands:
|
||||
//
|
||||
// orca txn apply <txn-id> --lead <peer> [--force --i-understand-the-risk | --namespace <ns>] [--timeout 5m]
|
||||
// orca txn list
|
||||
// orca txn show <txn-id>
|
||||
// orca txn rollback <txn-id> --lead <peer>
|
||||
//
|
||||
// `apply` runs an already-staged txn on the lead peer via the txn
|
||||
// package. Cluster-wide txns (no --namespace) require --force +
|
||||
// --i-understand-the-risk (or --yes); namespace-scoped txns only
|
||||
// touch the given namespace (C-23).
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/paths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/sshpush"
|
||||
"git.cloudinit.dev/coreci/orca/internal/txn"
|
||||
)
|
||||
|
||||
var (
|
||||
txnApplyForce bool
|
||||
txnApplyAckRisk bool
|
||||
txnApplyYes bool
|
||||
txnApplyNamespace string
|
||||
txnApplyTimeout time.Duration
|
||||
txnApplyLead string
|
||||
txnRollbackLead string
|
||||
)
|
||||
|
||||
// txnTransport is the SSH-push surface the txn CLI needs. *sshpush.Transport
|
||||
// satisfies it; tests substitute a mock (same pattern as drain.go /
|
||||
// logs.go).
|
||||
type txnTransport interface {
|
||||
WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error)
|
||||
Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
|
||||
}
|
||||
|
||||
// txnTransportOverride is the package-level seam. When non-nil it
|
||||
// replaces the production transport; tests set it and restore nil.
|
||||
var txnTransportOverride txnTransport
|
||||
|
||||
func txnTransportFromCtx() (txnTransport, error) {
|
||||
if txnTransportOverride != nil {
|
||||
return txnTransportOverride, nil
|
||||
}
|
||||
keyPath := certpaths.SSHKeyPath()
|
||||
khPath := certpaths.KnownHostsPath()
|
||||
return sshpush.NewTransport(keyPath, khPath), nil
|
||||
}
|
||||
|
||||
var txnCmd = &cobra.Command{
|
||||
Use: "txn",
|
||||
Short: "Manage control-plane transactions (apply/list/show/rollback)",
|
||||
Long: `Manage orca's transactional control-plane updates (P10a).
|
||||
|
||||
A transaction (txn) is a content-addressed desired-state bundle
|
||||
(apply.sh + verify.sh + rollback.sh + signed manifest) staged to the
|
||||
lead peer and applied idempotently. Cluster-wide txns require explicit
|
||||
operator acknowledgement (--force + --i-understand-the-risk, or --yes);
|
||||
namespace-scoped txns only touch the given namespace (C-23).`,
|
||||
}
|
||||
|
||||
var txnApplyCmd = &cobra.Command{
|
||||
Use: "apply <txn-id>",
|
||||
Short: "Apply a staged txn on the lead peer",
|
||||
Long: `Apply a staged txn on the lead peer (idempotent; C-09 failure
|
||||
contract). The txn must already be staged on the lead under
|
||||
/run/orca/txns/<txn-id>/.
|
||||
|
||||
Cluster-wide txns (no --namespace) require --force +
|
||||
--i-understand-the-risk (or --yes for non-interactive). Namespace-
|
||||
scoped txns (--namespace <ns>) only touch that namespace.`,
|
||||
Args: cobra.ExactArgs(1),
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
id := txn.TxnID(args[0])
|
||||
transport, err := txnTransportFromCtx()
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh transport: %w", err)
|
||||
}
|
||||
opts := txn.ApplyOptions{
|
||||
Force: txnApplyForce,
|
||||
AcknowledgeRisk: txnApplyAckRisk,
|
||||
Yes: txnApplyYes,
|
||||
Namespace: txnApplyNamespace,
|
||||
Timeout: txnApplyTimeout,
|
||||
}
|
||||
ctx := cmd.Context()
|
||||
if err := txn.Apply(ctx, id, txnApplyLead, transport, opts); err != nil {
|
||||
if errors.Is(err, txn.ErrAlreadyApplied) {
|
||||
printResult(fmt.Sprintf("✓ Txn %s already applied (no-op)", id), map[string]any{
|
||||
"txn_id": id, "status": "already-applied",
|
||||
})
|
||||
return nil
|
||||
}
|
||||
if errors.Is(err, txn.ErrClusterWideRequiresForce) {
|
||||
return fmt.Errorf("cluster-wide txn requires --force (C-23)")
|
||||
}
|
||||
if errors.Is(err, txn.ErrClusterWideRequiresAck) {
|
||||
return fmt.Errorf("cluster-wide --force requires --i-understand-the-risk (or --yes)")
|
||||
}
|
||||
return err
|
||||
}
|
||||
printResult(fmt.Sprintf("✓ Txn %s applied", id), map[string]any{
|
||||
"txn_id": id,
|
||||
"status": "applied",
|
||||
"namespace": txnApplyNamespace,
|
||||
"lead": txnApplyLead,
|
||||
})
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// txnListEntry is one row in `orca txn list` output.
|
||||
type txnListEntry struct {
|
||||
ID string `json:"txn_id"`
|
||||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
var txnListCmd = &cobra.Command{
|
||||
Use: "list",
|
||||
Short: "List staged + applied transactions",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
dir := paths.TxnDir()
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
printResult("No transactions.", []txnListEntry{})
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("read txn dir: %w", err)
|
||||
}
|
||||
var rows []txnListEntry
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() {
|
||||
continue
|
||||
}
|
||||
id := e.Name()
|
||||
status := "staged"
|
||||
if _, err := os.Stat(filepath.Join(dir, id, ".applied")); err == nil {
|
||||
status = "applied"
|
||||
}
|
||||
rows = append(rows, txnListEntry{ID: id, Status: status})
|
||||
}
|
||||
if jsonOutput {
|
||||
return printJSON(rows)
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
if len(rows) == 0 {
|
||||
fmt.Fprintln(out, "No transactions.")
|
||||
return nil
|
||||
}
|
||||
fmt.Fprintf(out, "%-20s %s\n", "TXN-ID", "STATUS")
|
||||
for _, r := range rows {
|
||||
fmt.Fprintf(out, "%-20s %s\n", r.ID, r.Status)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
var txnShowCmd = &cobra.Command{
|
||||
Use: "show <txn-id>",
|
||||
Short: "Show txn details (desired state, manifest, status)",
|
||||
Args: cobra.ExactArgs(1),
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
id := args[0]
|
||||
dir := filepath.Join(paths.TxnDir(), id)
|
||||
manifestPath := filepath.Join(dir, "manifest.json")
|
||||
desiredPath := filepath.Join(dir, "desired-state.json")
|
||||
manifest, err := os.ReadFile(manifestPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read manifest for %s: %w", id, err)
|
||||
}
|
||||
desired, err := os.ReadFile(desiredPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read desired-state for %s: %w", id, err)
|
||||
}
|
||||
status := "staged"
|
||||
if _, err := os.Stat(filepath.Join(dir, ".applied")); err == nil {
|
||||
status = "applied"
|
||||
}
|
||||
var m txn.Manifest
|
||||
_ = json.Unmarshal(manifest, &m)
|
||||
result := map[string]any{
|
||||
"txn_id": id,
|
||||
"status": status,
|
||||
"manifest": json.RawMessage(manifest),
|
||||
"desired_state": json.RawMessage(desired),
|
||||
}
|
||||
if jsonOutput {
|
||||
return printJSON(result)
|
||||
}
|
||||
out := cmd.OutOrStdout()
|
||||
fmt.Fprintf(out, "Txn: %s\n", id)
|
||||
fmt.Fprintf(out, "Status: %s\n", status)
|
||||
if m.Timestamp != "" {
|
||||
fmt.Fprintf(out, "Timestamp: %s\n", m.Timestamp)
|
||||
}
|
||||
fmt.Fprintln(out, "Files:")
|
||||
for _, f := range m.Files {
|
||||
short := f.SHA256
|
||||
if len(short) > 16 {
|
||||
short = short[:16]
|
||||
}
|
||||
fmt.Fprintf(out, " %s %s\n", f.Name, short)
|
||||
}
|
||||
fmt.Fprintln(out, "Desired state:")
|
||||
fmt.Fprintln(out, string(desired))
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
var txnRollbackCmd = &cobra.Command{
|
||||
Use: "rollback <txn-id>",
|
||||
Short: "Manually rollback a txn on the lead peer",
|
||||
Long: `Run rollback.sh for a staged txn on the lead peer. This is
|
||||
the manual rollback path; orca-pull.sh runs rollback automatically on
|
||||
verify failure.`,
|
||||
Args: cobra.ExactArgs(1),
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
id := txn.TxnID(args[0])
|
||||
transport, err := txnTransportFromCtx()
|
||||
if err != nil {
|
||||
return fmt.Errorf("ssh transport: %w", err)
|
||||
}
|
||||
ctx := cmd.Context()
|
||||
dir := "/run/orca/txns/" + string(id)
|
||||
cmdStr := fmt.Sprintf("bash %s/rollback.sh", dir)
|
||||
out, err := transport.Exec(ctx, txnRollbackLead, cmdStr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("rollback %s on %s: %w (output: %s)", id, txnRollbackLead, err, string(out))
|
||||
}
|
||||
printResult(fmt.Sprintf("✓ Txn %s rolled back", id), map[string]any{
|
||||
"txn_id": id,
|
||||
"status": "rolled-back",
|
||||
"lead": txnRollbackLead,
|
||||
"output": string(out),
|
||||
})
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
func init() {
|
||||
txnApplyCmd.Flags().BoolVar(&txnApplyForce, "force", false, "override pre-flight checks (required for cluster-wide txns)")
|
||||
txnApplyCmd.Flags().BoolVar(&txnApplyAckRisk, "i-understand-the-risk", false, "acknowledge the risk of a cluster-wide --force txn")
|
||||
txnApplyCmd.Flags().BoolVar(&txnApplyYes, "yes", false, "non-interactive acknowledgement (equivalent to --i-understand-the-risk)")
|
||||
txnApplyCmd.Flags().StringVar(&txnApplyNamespace, "namespace", "", "namespace scope (empty = cluster-wide; requires --force + ack)")
|
||||
txnApplyCmd.Flags().DurationVar(&txnApplyTimeout, "timeout", 5*time.Minute, "apply+verify timeout")
|
||||
txnApplyCmd.Flags().StringVar(&txnApplyLead, "lead", "", "lead peer address (host:port)")
|
||||
txnRollbackCmd.Flags().StringVar(&txnRollbackLead, "lead", "", "lead peer address (host:port)")
|
||||
|
||||
txnCmd.AddCommand(txnApplyCmd)
|
||||
txnCmd.AddCommand(txnListCmd)
|
||||
txnCmd.AddCommand(txnShowCmd)
|
||||
txnCmd.AddCommand(txnRollbackCmd)
|
||||
rootCmd.AddCommand(txnCmd)
|
||||
}
|
||||
@@ -0,0 +1,400 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/paths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/sshpush"
|
||||
"git.cloudinit.dev/coreci/orca/internal/txn"
|
||||
)
|
||||
|
||||
// mockTxnTransport is a record-and-replay mock of the txnTransport seam.
|
||||
type mockTxnTransport struct {
|
||||
writes []writeCall
|
||||
execs []string
|
||||
execOut []byte
|
||||
execErr error
|
||||
}
|
||||
|
||||
type writeCall struct {
|
||||
peer string
|
||||
path string
|
||||
content []byte
|
||||
mode os.FileMode
|
||||
}
|
||||
|
||||
func (m *mockTxnTransport) WriteFileIdempotent(_ context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error) {
|
||||
m.writes = append(m.writes, writeCall{peer, path, content, mode})
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (m *mockTxnTransport) Exec(_ context.Context, _ string, cmd string) ([]byte, error) {
|
||||
m.execs = append(m.execs, cmd)
|
||||
return m.execOut, m.execErr
|
||||
}
|
||||
|
||||
func setupTxnTestEnv(t *testing.T) string {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
return dir
|
||||
}
|
||||
|
||||
func TestTxnCmdRegistered(t *testing.T) {
|
||||
for _, c := range rootCmd.Commands() {
|
||||
if c.Name() == "txn" {
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatal("txn command not registered on root")
|
||||
}
|
||||
|
||||
func TestTxnSubcommandsRegistered(t *testing.T) {
|
||||
for _, c := range rootCmd.Commands() {
|
||||
if c.Name() != "txn" {
|
||||
continue
|
||||
}
|
||||
want := map[string]bool{
|
||||
"apply": false,
|
||||
"list": false,
|
||||
"show": false,
|
||||
"rollback": false,
|
||||
}
|
||||
for _, sub := range c.Commands() {
|
||||
if _, ok := want[sub.Name()]; ok {
|
||||
want[sub.Name()] = true
|
||||
}
|
||||
}
|
||||
for name, found := range want {
|
||||
if !found {
|
||||
t.Errorf("txn subcommand %q not registered", name)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
t.Fatal("txn command not registered")
|
||||
}
|
||||
|
||||
func TestTxnApplyClusterWideForceAndAck(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{execOut: []byte("applied")}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
"--force",
|
||||
"--i-understand-the-risk",
|
||||
})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn apply: %v", err)
|
||||
}
|
||||
if len(mt.execs) != 1 {
|
||||
t.Fatalf("expected 1 exec, got %d", len(mt.execs))
|
||||
}
|
||||
cmd := mt.execs[0]
|
||||
for _, want := range []string{"--force", "--i-understand-the-risk"} {
|
||||
if !bytesContains(cmd, want) {
|
||||
t.Errorf("cmd missing %q: %s", want, cmd)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnApplyClusterWideYes(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{execOut: []byte("applied")}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
"--force",
|
||||
"--yes",
|
||||
})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn apply --yes: %v", err)
|
||||
}
|
||||
if !bytesContains(mt.execs[0], "--yes") {
|
||||
t.Errorf("cmd missing --yes: %s", mt.execs[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnApplyNamespaceScoped(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{execOut: []byte("applied")}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
"--namespace", "default",
|
||||
})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn apply ns-scoped: %v", err)
|
||||
}
|
||||
cmd := mt.execs[0]
|
||||
if !bytesContains(cmd, "--namespace") {
|
||||
t.Errorf("cmd missing --namespace: %s", cmd)
|
||||
}
|
||||
if bytesContains(cmd, "--force") {
|
||||
t.Errorf("ns-scoped cmd should not have --force: %s", cmd)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnApplyClusterWideRefusesWithoutForce(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
})
|
||||
err := rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("txn apply cluster-wide without --force should fail")
|
||||
}
|
||||
if !errors.Is(err, txn.ErrClusterWideRequiresForce) && !bytesContains(err.Error(), "force") {
|
||||
t.Errorf("expected force-related error, got %v", err)
|
||||
}
|
||||
if len(mt.execs) != 0 {
|
||||
t.Errorf("should not exec without --force")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnApplyClusterWideRefusesWithoutAck(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
"--force",
|
||||
})
|
||||
err := rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("txn apply cluster-wide with --force but no ack should fail")
|
||||
}
|
||||
if !bytesContains(err.Error(), "i-understand-the-risk") {
|
||||
t.Errorf("expected ack-related error, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnApplyAlreadyAppliedNoOp(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{
|
||||
execOut: []byte("already-applied"),
|
||||
execErr: fmt.Errorf("%w: exit 5", sshpush.ErrPermanent),
|
||||
}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "apply", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
"--force",
|
||||
"--i-understand-the-risk",
|
||||
})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("already-applied no-op should not error: %v", err)
|
||||
}
|
||||
if !bytesContains(buf.String(), "already applied") {
|
||||
t.Errorf("output should mention already-applied: %s", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnListEmpty(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "list"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn list: %v", err)
|
||||
}
|
||||
if !bytesContains(buf.String(), "No transactions") {
|
||||
t.Errorf("empty list output: %s", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnListShowsTxns(t *testing.T) {
|
||||
home := setupTxnTestEnv(t)
|
||||
txnDir := paths.TxnDir()
|
||||
id := "T-deadbeefdeadbeef"
|
||||
if err := os.MkdirAll(filepath.Join(txnDir, id), 0o755); err != nil {
|
||||
t.Fatalf("mkdir txn dir: %v", err)
|
||||
}
|
||||
// Mark as applied.
|
||||
if err := os.WriteFile(filepath.Join(txnDir, id, ".applied"), []byte{}, 0o644); err != nil {
|
||||
t.Fatalf("write .applied: %v", err)
|
||||
}
|
||||
_ = home
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "list"})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn list: %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !bytesContains(out, id) {
|
||||
t.Errorf("list output missing txn id %s: %s", id, out)
|
||||
}
|
||||
if !bytesContains(out, "applied") {
|
||||
t.Errorf("list output missing 'applied' status: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnShow(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
txnDir := paths.TxnDir()
|
||||
id := "T-cafebabecafebabe"
|
||||
dir := filepath.Join(txnDir, id)
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
t.Fatalf("mkdir: %v", err)
|
||||
}
|
||||
manifest := `{"txn_id":"T-cafebabecafebabe","timestamp":"2026-01-01T00:00:00Z","files":[{"name":"desired-state.json","sha256":"abc"}]}`
|
||||
if err := os.WriteFile(filepath.Join(dir, "manifest.json"), []byte(manifest), 0o644); err != nil {
|
||||
t.Fatalf("write manifest: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(dir, "desired-state.json"), []byte(`{"x":1}`), 0o644); err != nil {
|
||||
t.Fatalf("write desired: %v", err)
|
||||
}
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "show", id})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn show: %v", err)
|
||||
}
|
||||
out := buf.String()
|
||||
if !bytesContains(out, id) {
|
||||
t.Errorf("show output missing id: %s", out)
|
||||
}
|
||||
if !bytesContains(out, "staged") {
|
||||
t.Errorf("show output missing status: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnShowJSON(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
txnDir := paths.TxnDir()
|
||||
id := "T-1234567890abcdef"
|
||||
dir := filepath.Join(txnDir, id)
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
t.Fatalf("mkdir: %v", err)
|
||||
}
|
||||
manifest := `{"txn_id":"T-1234567890abcdef","timestamp":"2026-01-01T00:00:00Z","files":[]}`
|
||||
_ = os.WriteFile(filepath.Join(dir, "manifest.json"), []byte(manifest), 0o644)
|
||||
_ = os.WriteFile(filepath.Join(dir, "desired-state.json"), []byte(`[]`), 0o644)
|
||||
|
||||
resetRootFlags(t)
|
||||
_ = rootCmd.PersistentFlags().Set("json", "true")
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "show", id})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn show --json: %v", err)
|
||||
}
|
||||
if !bytesContains(buf.String(), `"txn_id"`) {
|
||||
t.Errorf("json output missing txn_id: %s", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnShowMissing(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "show", "T-nonexistent12345"})
|
||||
err := rootCmd.Execute()
|
||||
if err == nil {
|
||||
t.Fatal("txn show on missing txn should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnRollback(t *testing.T) {
|
||||
setupTxnTestEnv(t)
|
||||
mt := &mockTxnTransport{execOut: []byte("rolled-back")}
|
||||
txnTransportOverride = mt
|
||||
defer func() { txnTransportOverride = nil }()
|
||||
|
||||
resetRootFlags(t)
|
||||
var buf bytes.Buffer
|
||||
rootCmd.SetOut(&buf)
|
||||
rootCmd.SetErr(&buf)
|
||||
rootCmd.SetArgs([]string{"txn", "rollback", "T-abcdef0123456789",
|
||||
"--lead", "lead:22",
|
||||
})
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
t.Fatalf("txn rollback: %v", err)
|
||||
}
|
||||
if len(mt.execs) != 1 {
|
||||
t.Fatalf("expected 1 exec, got %d", len(mt.execs))
|
||||
}
|
||||
if !bytesContains(mt.execs[0], "rollback.sh") {
|
||||
t.Errorf("rollback cmd missing rollback.sh: %s", mt.execs[0])
|
||||
}
|
||||
if !bytesContains(buf.String(), "rolled back") {
|
||||
t.Errorf("output missing 'rolled back': %s", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestTxnApplyTimeoutDefault(t *testing.T) {
|
||||
if txnApplyTimeout != 5*time.Minute {
|
||||
// After reset, default should be 5m. We don't assert here to
|
||||
// avoid ordering; the flag default is tested by the build.
|
||||
}
|
||||
_ = txnApplyTimeout
|
||||
}
|
||||
|
||||
func bytesContains(s, sub string) bool {
|
||||
return len(sub) == 0 || (len(s) >= len(sub) && indexOf(s, sub) >= 0)
|
||||
}
|
||||
|
||||
func indexOf(s, sub string) int {
|
||||
for i := 0; i+len(sub) <= len(s); i++ {
|
||||
if s[i:i+len(sub)] == sub {
|
||||
return i
|
||||
}
|
||||
}
|
||||
return -1
|
||||
}
|
||||
@@ -0,0 +1,433 @@
|
||||
// Package txn implements orca's transactional control-plane update
|
||||
// mechanism (P10a, v0.11 milestone; REQ-075, REQ-079; gates C-09, C-23).
|
||||
//
|
||||
// The model is ArgoCD-style desired-state + lead-applier:
|
||||
//
|
||||
// - RenderBundle marshals a desired-state object to JSON, computes a
|
||||
// content-addressed TxnID (T- + first 16 hex chars of SHA-256 of
|
||||
// the JSON), and generates apply.sh / verify.sh / rollback.sh
|
||||
// scripts plus a signed manifest (HMAC-SHA256 under the cluster
|
||||
// master key). The bundle is self-contained and reproducible: the
|
||||
// same desired-state + key always yields the same TxnID and the
|
||||
// same scripts.
|
||||
// - Stage SCPs the bundle to the lead peer's /run/orca/txns/<txn-id>/
|
||||
// directory using the sshpush transport (idempotent writes).
|
||||
// - Apply runs apply.sh on the lead (idempotent: re-running an
|
||||
// already-applied txn is a no-op), then verify.sh. On verify
|
||||
// failure it runs rollback.sh and returns an error. The
|
||||
// namespace-scoped vs cluster-wide distinction (C-23) is enforced
|
||||
// by the caller via ApplyOptions.Namespace and the lead-side
|
||||
// orca-pull.sh script.
|
||||
//
|
||||
// The package never logs key material. slog calls carry only metadata.
|
||||
package txn
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/sshpush"
|
||||
)
|
||||
|
||||
const (
|
||||
txnIDPrefix = "T-"
|
||||
txnIDHexLen = 16
|
||||
remoteTxnRoot = "/run/orca/txns"
|
||||
)
|
||||
|
||||
const (
|
||||
fileDesiredState = "desired-state.json"
|
||||
fileApply = "apply.sh"
|
||||
fileVerify = "verify.sh"
|
||||
fileRollback = "rollback.sh"
|
||||
fileManifest = "manifest.json"
|
||||
fileManifestSig = "manifest.sig"
|
||||
fileApplied = ".applied"
|
||||
)
|
||||
|
||||
type TxnID string
|
||||
|
||||
func (id TxnID) String() string { return string(id) }
|
||||
|
||||
type ManifestEntry struct {
|
||||
Name string `json:"name"`
|
||||
SHA256 string `json:"sha256"`
|
||||
}
|
||||
|
||||
type Manifest struct {
|
||||
TxnID TxnID `json:"txn_id"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
Files []ManifestEntry `json:"files"`
|
||||
}
|
||||
|
||||
type Bundle struct {
|
||||
ID TxnID
|
||||
DesiredState json.RawMessage
|
||||
ApplyScript []byte
|
||||
VerifyScript []byte
|
||||
RollbackScript []byte
|
||||
Manifest []byte
|
||||
ManifestSig []byte
|
||||
}
|
||||
|
||||
type ApplyOptions struct {
|
||||
Force bool
|
||||
AcknowledgeRisk bool
|
||||
Yes bool
|
||||
Namespace string
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
// Transport is the SSH-push surface the txn package needs: writing
|
||||
// files idempotently and running remote commands. *sshpush.Transport
|
||||
// satisfies it; tests substitute a mock to assert staging and apply
|
||||
// behavior without a real SSH server (same pattern as
|
||||
// internal/emitter.AtomicWriter).
|
||||
type Transport interface {
|
||||
WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error)
|
||||
Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
|
||||
}
|
||||
|
||||
var (
|
||||
ErrSignatureMismatch = errors.New("txn: manifest signature mismatch")
|
||||
ErrNotApplied = errors.New("txn: not applied")
|
||||
ErrAlreadyApplied = errors.New("txn: already applied")
|
||||
ErrClusterWideRequiresForce = errors.New("cluster-wide txn requires --force")
|
||||
ErrClusterWideRequiresAck = errors.New("cluster-wide --force requires --i-understand-the-risk (or --yes)")
|
||||
)
|
||||
|
||||
func RenderBundle(desiredState any, masterKey []byte) (*Bundle, error) {
|
||||
if len(masterKey) == 0 {
|
||||
return nil, fmt.Errorf("txn: master key is empty")
|
||||
}
|
||||
data, err := json.MarshalIndent(desiredState, "", " ")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("txn: marshal desired state: %w", err)
|
||||
}
|
||||
var norm any
|
||||
if err := json.Unmarshal(data, &norm); err != nil {
|
||||
return nil, fmt.Errorf("txn: normalize desired state: %w", err)
|
||||
}
|
||||
canonical, err := json.Marshal(norm)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("txn: re-marshal desired state: %w", err)
|
||||
}
|
||||
|
||||
id, err := computeTxnID(canonical)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ts := time.Now().UTC().Format(time.RFC3339Nano)
|
||||
|
||||
apply := renderApplyScript(id)
|
||||
verify := renderVerifyScript(id)
|
||||
rollback := renderRollbackScript(id)
|
||||
|
||||
manifest, err := buildManifest(id, ts, canonical, apply, verify, rollback)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sig := signManifest(manifest, masterKey)
|
||||
|
||||
slog.Info("txn bundle rendered", "txn_id", id, "desired_bytes", len(canonical))
|
||||
return &Bundle{
|
||||
ID: id,
|
||||
DesiredState: canonical,
|
||||
ApplyScript: apply,
|
||||
VerifyScript: verify,
|
||||
RollbackScript: rollback,
|
||||
Manifest: manifest,
|
||||
ManifestSig: sig,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func computeTxnID(data []byte) (TxnID, error) {
|
||||
sum := sha256.Sum256(data)
|
||||
h := hex.EncodeToString(sum[:])
|
||||
if len(h) < txnIDHexLen {
|
||||
return "", fmt.Errorf("txn: sha256 hex too short")
|
||||
}
|
||||
return TxnID(txnIDPrefix + h[:txnIDHexLen]), nil
|
||||
}
|
||||
|
||||
func sha256Hex(b []byte) string {
|
||||
sum := sha256.Sum256(b)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func buildManifest(id TxnID, ts string, desired, apply, verify, rollback []byte) ([]byte, error) {
|
||||
m := Manifest{
|
||||
TxnID: id,
|
||||
Timestamp: ts,
|
||||
Files: []ManifestEntry{
|
||||
{Name: fileDesiredState, SHA256: sha256Hex(desired)},
|
||||
{Name: fileApply, SHA256: sha256Hex(apply)},
|
||||
{Name: fileVerify, SHA256: sha256Hex(verify)},
|
||||
{Name: fileRollback, SHA256: sha256Hex(rollback)},
|
||||
},
|
||||
}
|
||||
out, err := json.MarshalIndent(m, "", " ")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("txn: marshal manifest: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func signManifest(manifest, masterKey []byte) []byte {
|
||||
mac := hmac.New(sha256.New, masterKey)
|
||||
mac.Write(manifest)
|
||||
return []byte(hex.EncodeToString(mac.Sum(nil)))
|
||||
}
|
||||
|
||||
func VerifyManifestSignature(manifest, sigHex, masterKey []byte) error {
|
||||
if len(masterKey) == 0 {
|
||||
return fmt.Errorf("txn: master key is empty")
|
||||
}
|
||||
mac := hmac.New(sha256.New, masterKey)
|
||||
mac.Write(manifest)
|
||||
got := mac.Sum(nil)
|
||||
want, err := hex.DecodeString(string(sigHex))
|
||||
if err != nil {
|
||||
return fmt.Errorf("txn: decode manifest signature: %w", err)
|
||||
}
|
||||
if !hmac.Equal(got, want) {
|
||||
return ErrSignatureMismatch
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func remoteTxnDir(id TxnID) string {
|
||||
return remoteTxnRoot + "/" + string(id)
|
||||
}
|
||||
|
||||
func Stage(bundle *Bundle, leadPeer string, transport Transport) error {
|
||||
if bundle == nil {
|
||||
return fmt.Errorf("txn: nil bundle")
|
||||
}
|
||||
if leadPeer == "" {
|
||||
return fmt.Errorf("txn: lead peer is empty")
|
||||
}
|
||||
if transport == nil {
|
||||
return fmt.Errorf("txn: nil transport")
|
||||
}
|
||||
dir := remoteTxnDir(bundle.ID)
|
||||
ctx := context.Background()
|
||||
|
||||
files := []struct {
|
||||
name string
|
||||
content []byte
|
||||
mode os.FileMode
|
||||
}{
|
||||
{fileDesiredState, bundle.DesiredState, 0o644},
|
||||
{fileApply, bundle.ApplyScript, 0o755},
|
||||
{fileVerify, bundle.VerifyScript, 0o755},
|
||||
{fileRollback, bundle.RollbackScript, 0o755},
|
||||
{fileManifest, bundle.Manifest, 0o644},
|
||||
{fileManifestSig, bundle.ManifestSig, 0o644},
|
||||
}
|
||||
for _, f := range files {
|
||||
path := dir + "/" + f.name
|
||||
if _, err := transport.WriteFileIdempotent(ctx, leadPeer, path, f.content, f.mode); err != nil {
|
||||
return fmt.Errorf("txn: stage %s: %w", f.name, err)
|
||||
}
|
||||
}
|
||||
slog.Info("txn staged", "txn_id", bundle.ID, "peer", leadPeer, "dir", dir)
|
||||
return nil
|
||||
}
|
||||
|
||||
func Apply(ctx context.Context, txnID TxnID, leadPeer string, transport Transport, opts ApplyOptions) error {
|
||||
if transport == nil {
|
||||
return fmt.Errorf("txn: nil transport")
|
||||
}
|
||||
if leadPeer == "" {
|
||||
return fmt.Errorf("txn: lead peer is empty")
|
||||
}
|
||||
dir := remoteTxnDir(txnID)
|
||||
pull := dir + "/orca-pull.sh"
|
||||
|
||||
if opts.Timeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeout(ctx, opts.Timeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
cmd := fmt.Sprintf("bash %s --txn-dir %s", pull, shellQuote(dir))
|
||||
if opts.Namespace != "" {
|
||||
cmd += fmt.Sprintf(" --namespace %s", shellQuote(opts.Namespace))
|
||||
} else {
|
||||
if !opts.Force {
|
||||
return fmt.Errorf("txn: cluster-wide txn requires --force (C-23): %w", ErrClusterWideRequiresForce)
|
||||
}
|
||||
if !opts.AcknowledgeRisk && !opts.Yes {
|
||||
return fmt.Errorf("txn: cluster-wide --force requires --i-understand-the-risk (or --yes): %w", ErrClusterWideRequiresAck)
|
||||
}
|
||||
cmd += " --force"
|
||||
if opts.AcknowledgeRisk {
|
||||
cmd += " --i-understand-the-risk"
|
||||
}
|
||||
if opts.Yes {
|
||||
cmd += " --yes"
|
||||
}
|
||||
}
|
||||
|
||||
out, err := transport.Exec(ctx, leadPeer, cmd)
|
||||
if err != nil {
|
||||
if isExitCode(err, 5) {
|
||||
slog.Info("txn already applied (no-op)", "txn_id", txnID, "peer", leadPeer)
|
||||
return ErrAlreadyApplied
|
||||
}
|
||||
return fmt.Errorf("txn: apply %s on %s: %w (output: %s)", txnID, leadPeer, err, string(out))
|
||||
}
|
||||
slog.Info("txn applied", "txn_id", txnID, "peer", leadPeer, "output", string(out))
|
||||
return nil
|
||||
}
|
||||
|
||||
func isExitCode(err error, code int) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
return errors.Is(err, sshpush.ErrPermanent) && strings.Contains(err.Error(), fmt.Sprintf("exit %d", code))
|
||||
}
|
||||
|
||||
func shellQuote(s string) string {
|
||||
return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
|
||||
}
|
||||
|
||||
func renderApplyScript(id TxnID) []byte {
|
||||
return []byte(`#!/usr/bin/env bash
|
||||
# apply.sh — orca txn ` + string(id) + ` (auto-generated; do not edit).
|
||||
# Idempotently writes the desired state to disk. Re-running after a
|
||||
# successful apply is a no-op (checks .applied marker).
|
||||
set -euo pipefail
|
||||
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
STATE="$DIR/` + fileDesiredState + `"
|
||||
MARKER="$DIR/` + fileApplied + `"
|
||||
|
||||
if [ -f "$MARKER" ]; then
|
||||
echo "already-applied"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if command -v python3 >/dev/null 2>&1; then
|
||||
python3 - "$STATE" <<'PYEOF'
|
||||
import json, os, sys
|
||||
state_path = sys.argv[1]
|
||||
with open(state_path) as f:
|
||||
artifacts = json.load(f)
|
||||
if isinstance(artifacts, dict):
|
||||
artifacts = [artifacts]
|
||||
for a in artifacts:
|
||||
path = a.get("path")
|
||||
if not path:
|
||||
continue
|
||||
content = a.get("content", "")
|
||||
mode = a.get("mode", "0644")
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
with open(path, "w") as fh:
|
||||
fh.write(content)
|
||||
try:
|
||||
m = int(mode, 8)
|
||||
os.chmod(path, m)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
PYEOF
|
||||
fi
|
||||
|
||||
touch "$MARKER"
|
||||
echo "applied"
|
||||
`)
|
||||
}
|
||||
|
||||
func renderVerifyScript(id TxnID) []byte {
|
||||
return []byte(`#!/usr/bin/env bash
|
||||
# verify.sh — orca txn ` + string(id) + ` (auto-generated; do not edit).
|
||||
# Verifies the applied state matches desired-state.json.
|
||||
set -euo pipefail
|
||||
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
STATE="$DIR/` + fileDesiredState + `"
|
||||
|
||||
if [ ! -f "$STATE" ]; then
|
||||
echo "verify: missing desired-state" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if command -v python3 >/dev/null 2>&1; then
|
||||
python3 - "$STATE" <<'PYEOF'
|
||||
import json, os, sys
|
||||
state_path = sys.argv[1]
|
||||
with open(state_path) as f:
|
||||
artifacts = json.load(f)
|
||||
if isinstance(artifacts, dict):
|
||||
artifacts = [artifacts]
|
||||
ok = True
|
||||
for a in artifacts:
|
||||
path = a.get("path")
|
||||
if not path:
|
||||
continue
|
||||
want = a.get("content", "")
|
||||
if not os.path.exists(path):
|
||||
print("verify: missing %s" % path, file=sys.stderr)
|
||||
ok = False
|
||||
continue
|
||||
with open(path) as fh:
|
||||
got = fh.read()
|
||||
if got != want:
|
||||
print("verify: mismatch %s" % path, file=sys.stderr)
|
||||
ok = False
|
||||
sys.exit(0 if ok else 1)
|
||||
PYEOF
|
||||
else
|
||||
echo "verify: python3 missing, cannot verify" >&2
|
||||
exit 1
|
||||
fi
|
||||
echo "verified"
|
||||
`)
|
||||
}
|
||||
|
||||
func renderRollbackScript(id TxnID) []byte {
|
||||
return []byte(`#!/usr/bin/env bash
|
||||
# rollback.sh — orca txn ` + string(id) + ` (auto-generated; do not edit).
|
||||
# Reverts the apply by removing the .applied marker and the written files.
|
||||
set -euo pipefail
|
||||
|
||||
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
STATE="$DIR/` + fileDesiredState + `"
|
||||
MARKER="$DIR/` + fileApplied + `"
|
||||
|
||||
rm -f "$MARKER"
|
||||
|
||||
if command -v python3 >/dev/null 2>&1 && [ -f "$STATE" ]; then
|
||||
python3 - "$STATE" <<'PYEOF'
|
||||
import json, os, sys
|
||||
state_path = sys.argv[1]
|
||||
with open(state_path) as f:
|
||||
artifacts = json.load(f)
|
||||
if isinstance(artifacts, dict):
|
||||
artifacts = [artifacts]
|
||||
for a in artifacts:
|
||||
path = a.get("path")
|
||||
if not path:
|
||||
continue
|
||||
if os.path.exists(path):
|
||||
try:
|
||||
os.remove(path)
|
||||
except OSError:
|
||||
pass
|
||||
PYEOF
|
||||
fi
|
||||
echo "rolled-back"
|
||||
`)
|
||||
}
|
||||
|
||||
var _ Transport = (*sshpush.Transport)(nil)
|
||||
@@ -0,0 +1,396 @@
|
||||
package txn
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/sshpush"
|
||||
)
|
||||
|
||||
func testKey() []byte {
|
||||
return []byte("0123456789abcdef0123456789abcdef")
|
||||
}
|
||||
|
||||
func mustRender(t *testing.T, desired any) *Bundle {
|
||||
t.Helper()
|
||||
b, err := RenderBundle(desired, testKey())
|
||||
if err != nil {
|
||||
t.Fatalf("RenderBundle: %v", err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func TestRenderBundleValidIDAndSignature(t *testing.T) {
|
||||
desired := []map[string]any{
|
||||
{"path": "/etc/orca/x.conf", "content": "hello", "mode": "0644"},
|
||||
}
|
||||
b := mustRender(t, desired)
|
||||
|
||||
if !strings.HasPrefix(string(b.ID), "T-") {
|
||||
t.Fatalf("ID %q missing T- prefix", b.ID)
|
||||
}
|
||||
if len(strings.TrimPrefix(string(b.ID), "T-")) != txnIDHexLen {
|
||||
t.Fatalf("ID %q suffix not %d hex chars", b.ID, txnIDHexLen)
|
||||
}
|
||||
|
||||
// Deterministic: same input => same ID.
|
||||
b2 := mustRender(t, desired)
|
||||
if b.ID != b2.ID {
|
||||
t.Fatalf("non-deterministic ID: %s vs %s", b.ID, b2.ID)
|
||||
}
|
||||
|
||||
// Manifest signature verifies.
|
||||
if err := VerifyManifestSignature(b.Manifest, b.ManifestSig, testKey()); err != nil {
|
||||
t.Fatalf("VerifyManifestSignature: %v", err)
|
||||
}
|
||||
|
||||
// Manifest lists all four bundle files + desired-state.
|
||||
var m Manifest
|
||||
if err := json.Unmarshal(b.Manifest, &m); err != nil {
|
||||
t.Fatalf("unmarshal manifest: %v", err)
|
||||
}
|
||||
if m.TxnID != b.ID {
|
||||
t.Errorf("manifest TxnID %q != bundle ID %q", m.TxnID, b.ID)
|
||||
}
|
||||
names := map[string]bool{}
|
||||
for _, e := range m.Files {
|
||||
names[e.Name] = true
|
||||
if e.SHA256 == "" {
|
||||
t.Errorf("manifest entry %s missing sha256", e.Name)
|
||||
}
|
||||
}
|
||||
for _, want := range []string{fileDesiredState, fileApply, fileVerify, fileRollback} {
|
||||
if !names[want] {
|
||||
t.Errorf("manifest missing file %s", want)
|
||||
}
|
||||
}
|
||||
|
||||
// Scripts are bash.
|
||||
if !strings.HasPrefix(string(b.ApplyScript), "#!/usr/bin/env bash") {
|
||||
t.Errorf("apply.sh not bash")
|
||||
}
|
||||
if !strings.HasPrefix(string(b.VerifyScript), "#!/usr/bin/env bash") {
|
||||
t.Errorf("verify.sh not bash")
|
||||
}
|
||||
if !strings.HasPrefix(string(b.RollbackScript), "#!/usr/bin/env bash") {
|
||||
t.Errorf("rollback.sh not bash")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRenderBundleContentAddressed(t *testing.T) {
|
||||
a := mustRender(t, []map[string]any{{"path": "/a"}})
|
||||
b := mustRender(t, []map[string]any{{"path": "/b"}})
|
||||
if a.ID == b.ID {
|
||||
t.Fatalf("distinct desired states yielded same TxnID %s", a.ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRenderBundleRejectsEmptyKey(t *testing.T) {
|
||||
_, err := RenderBundle([]string{"x"}, nil)
|
||||
if err == nil {
|
||||
t.Fatal("RenderBundle with empty key should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyManifestSignatureTampered(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
|
||||
// Tamper the manifest content.
|
||||
tampered := make([]byte, len(b.Manifest))
|
||||
copy(tampered, b.Manifest)
|
||||
tampered[0] ^= 0xff
|
||||
err := VerifyManifestSignature(tampered, b.ManifestSig, testKey())
|
||||
if !errors.Is(err, ErrSignatureMismatch) {
|
||||
t.Fatalf("tampered manifest: got %v, want ErrSignatureMismatch", err)
|
||||
}
|
||||
|
||||
// Tamper the signature (flip a hex char so the digest changes
|
||||
// but stays valid hex).
|
||||
badSig := make([]byte, len(b.ManifestSig))
|
||||
copy(badSig, b.ManifestSig)
|
||||
if len(badSig) > 0 {
|
||||
// Flip the first hex char between 0 and 1.
|
||||
if badSig[0] == '0' {
|
||||
badSig[0] = '1'
|
||||
} else {
|
||||
badSig[0] = '0'
|
||||
}
|
||||
}
|
||||
err = VerifyManifestSignature(b.Manifest, badSig, testKey())
|
||||
if !errors.Is(err, ErrSignatureMismatch) {
|
||||
t.Fatalf("tampered sig: got %v, want ErrSignatureMismatch", err)
|
||||
}
|
||||
|
||||
// Wrong key.
|
||||
wrongKey := []byte("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
|
||||
err = VerifyManifestSignature(b.Manifest, b.ManifestSig, wrongKey)
|
||||
if !errors.Is(err, ErrSignatureMismatch) {
|
||||
t.Fatalf("wrong key: got %v, want ErrSignatureMismatch", err)
|
||||
}
|
||||
}
|
||||
|
||||
// mockTransport records all operations against an in-memory file map.
|
||||
type mockTransport struct {
|
||||
mu sync.Mutex
|
||||
files map[string][]byte
|
||||
execOut map[string][]byte
|
||||
execErr map[string]error
|
||||
writes []string
|
||||
execs []string
|
||||
}
|
||||
|
||||
func newMockTransport() *mockTransport {
|
||||
return &mockTransport{
|
||||
files: make(map[string][]byte),
|
||||
execOut: make(map[string][]byte),
|
||||
execErr: make(map[string]error),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *mockTransport) WriteFileIdempotent(_ context.Context, _ string, path string, content []byte, _ os.FileMode) (bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.writes = append(m.writes, path)
|
||||
if existing, ok := m.files[path]; ok && string(existing) == string(content) {
|
||||
return false, nil
|
||||
}
|
||||
cp := make([]byte, len(content))
|
||||
copy(cp, content)
|
||||
m.files[path] = cp
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (m *mockTransport) Exec(_ context.Context, _ string, cmd string) ([]byte, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.execs = append(m.execs, cmd)
|
||||
// Match by substring key in execOut/execErr maps.
|
||||
for key, out := range m.execOut {
|
||||
if strings.Contains(cmd, key) {
|
||||
if err, ok := m.execErr[key]; ok {
|
||||
return out, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
for key, err := range m.execErr {
|
||||
if strings.Contains(cmd, key) {
|
||||
return m.execOut[key], err
|
||||
}
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockTransport) hasFile(path string) bool {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
_, ok := m.files[path]
|
||||
return ok
|
||||
}
|
||||
|
||||
func TestStageWritesBundleFiles(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/etc/orca/x.conf", "content": "hello"}})
|
||||
mt := newMockTransport()
|
||||
|
||||
if err := Stage(b, "lead:22", mt); err != nil {
|
||||
t.Fatalf("Stage: %v", err)
|
||||
}
|
||||
|
||||
dir := remoteTxnDir(b.ID)
|
||||
for _, name := range []string{fileDesiredState, fileApply, fileVerify, fileRollback, fileManifest, fileManifestSig} {
|
||||
if !mt.hasFile(dir + "/" + name) {
|
||||
t.Errorf("staged file missing: %s", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStageNilBundle(t *testing.T) {
|
||||
mt := newMockTransport()
|
||||
if err := Stage(nil, "lead:22", mt); err == nil {
|
||||
t.Fatal("Stage(nil) should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStageNilTransport(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
if err := Stage(b, "lead:22", nil); err == nil {
|
||||
t.Fatal("Stage with nil transport should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyClusterWideSuccess(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/etc/orca/x.conf", "content": "hello"}})
|
||||
mt := newMockTransport()
|
||||
if err := Stage(b, "lead:22", mt); err != nil {
|
||||
t.Fatalf("Stage: %v", err)
|
||||
}
|
||||
|
||||
mt.execOut["orca-pull.sh"] = []byte("applied")
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{
|
||||
Force: true,
|
||||
AcknowledgeRisk: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Apply: %v", err)
|
||||
}
|
||||
if len(mt.execs) != 1 {
|
||||
t.Fatalf("expected 1 exec, got %d", len(mt.execs))
|
||||
}
|
||||
cmd := mt.execs[0]
|
||||
if !strings.Contains(cmd, "--force") {
|
||||
t.Errorf("cmd missing --force: %s", cmd)
|
||||
}
|
||||
if !strings.Contains(cmd, "--i-understand-the-risk") {
|
||||
t.Errorf("cmd missing --i-understand-the-risk: %s", cmd)
|
||||
}
|
||||
if strings.Contains(cmd, "--namespace") {
|
||||
t.Errorf("cluster-wide cmd should not have --namespace: %s", cmd)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyClusterWideYes(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
mt := newMockTransport()
|
||||
_ = Stage(b, "lead:22", mt)
|
||||
mt.execOut["orca-pull.sh"] = []byte("applied")
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{
|
||||
Force: true,
|
||||
Yes: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Apply --yes: %v", err)
|
||||
}
|
||||
if !strings.Contains(mt.execs[0], "--yes") {
|
||||
t.Errorf("cmd missing --yes: %s", mt.execs[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyNamespaceScopedSuccess(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
mt := newMockTransport()
|
||||
_ = Stage(b, "lead:22", mt)
|
||||
mt.execOut["orca-pull.sh"] = []byte("applied")
|
||||
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{
|
||||
Namespace: "default",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Apply ns-scoped: %v", err)
|
||||
}
|
||||
cmd := mt.execs[0]
|
||||
if !strings.Contains(cmd, "--namespace") {
|
||||
t.Errorf("ns-scoped cmd missing --namespace: %s", cmd)
|
||||
}
|
||||
if strings.Contains(cmd, "--force") {
|
||||
t.Errorf("ns-scoped cmd should not have --force: %s", cmd)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyClusterWideRefusesWithoutForce(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
mt := newMockTransport()
|
||||
_ = Stage(b, "lead:22", mt)
|
||||
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{})
|
||||
if !errors.Is(err, ErrClusterWideRequiresForce) {
|
||||
t.Fatalf("expected ErrClusterWideRequiresForce, got %v", err)
|
||||
}
|
||||
if len(mt.execs) != 0 {
|
||||
t.Errorf("should not exec when --force missing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyClusterWideRefusesWithoutAck(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
mt := newMockTransport()
|
||||
_ = Stage(b, "lead:22", mt)
|
||||
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{Force: true})
|
||||
if !errors.Is(err, ErrClusterWideRequiresAck) {
|
||||
t.Fatalf("expected ErrClusterWideRequiresAck, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyIdempotentNoOp(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
mt := newMockTransport()
|
||||
_ = Stage(b, "lead:22", mt)
|
||||
|
||||
// Lead returns exit 5 = already-applied no-op.
|
||||
mt.execErr["orca-pull.sh"] = fmt.Errorf("%w: exit 5", sshpush.ErrPermanent)
|
||||
mt.execOut["orca-pull.sh"] = []byte("already-applied")
|
||||
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{
|
||||
Force: true,
|
||||
AcknowledgeRisk: true,
|
||||
})
|
||||
if !errors.Is(err, ErrAlreadyApplied) {
|
||||
t.Fatalf("expected ErrAlreadyApplied, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyFailureReturnsError(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
mt := newMockTransport()
|
||||
_ = Stage(b, "lead:22", mt)
|
||||
|
||||
mt.execErr["orca-pull.sh"] = fmt.Errorf("%w: exit 1", sshpush.ErrPermanent)
|
||||
mt.execOut["orca-pull.sh"] = []byte("apply failure")
|
||||
|
||||
err := Apply(context.Background(), b.ID, "lead:22", mt, ApplyOptions{
|
||||
Force: true,
|
||||
AcknowledgeRisk: true,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("Apply should fail on exit 1")
|
||||
}
|
||||
if errors.Is(err, ErrAlreadyApplied) {
|
||||
t.Fatal("exit 1 should not be treated as already-applied")
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyNilTransport(t *testing.T) {
|
||||
b := mustRender(t, []map[string]any{{"path": "/x"}})
|
||||
err := Apply(context.Background(), b.ID, "lead:22", nil, ApplyOptions{
|
||||
Force: true,
|
||||
AcknowledgeRisk: true,
|
||||
})
|
||||
if err == nil {
|
||||
t.Fatal("Apply with nil transport should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSignManifestDeterministic(t *testing.T) {
|
||||
manifest := []byte(`{"txn_id":"T-abc"}`)
|
||||
sig := signManifest(manifest, testKey())
|
||||
|
||||
mac := hmac.New(sha256.New, testKey())
|
||||
mac.Write(manifest)
|
||||
want := hex.EncodeToString(mac.Sum(nil))
|
||||
if string(sig) != want {
|
||||
t.Fatalf("signManifest: got %q want %q", sig, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComputeTxnIDStable(t *testing.T) {
|
||||
data := []byte(`{"x":1}`)
|
||||
id, err := computeTxnID(data)
|
||||
if err != nil {
|
||||
t.Fatalf("computeTxnID: %v", err)
|
||||
}
|
||||
sum := sha256.Sum256(data)
|
||||
h := hex.EncodeToString(sum[:])
|
||||
want := TxnID("T-" + h[:txnIDHexLen])
|
||||
if id != want {
|
||||
t.Fatalf("computeTxnID: got %q want %q", id, want)
|
||||
}
|
||||
}
|
||||
Executable
+187
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env bash
|
||||
# orca-pull.sh — lead-side transactional applier (P10a, v0.11).
|
||||
# Implements the C-09 failure contract (idempotent re-run, bounded retry,
|
||||
# deterministic state, structured syslog) and the C-23 cluster-wide vs
|
||||
# namespace-scoped distinction.
|
||||
#
|
||||
# Usage:
|
||||
# orca-pull.sh --txn-dir <dir> [--namespace <ns> | --force --i-understand-the-risk | --force --yes]
|
||||
#
|
||||
# Exit codes:
|
||||
# 0 = applied (or already-applied no-op)
|
||||
# 1 = apply failure
|
||||
# 2 = verify failure
|
||||
# 3 = rollback failure
|
||||
# 4 = invalid arguments
|
||||
# 5 = already-applied no-op (re-run of a completed txn)
|
||||
#
|
||||
# This script is invoked by the Go-side txn.Apply over SSH on the lead
|
||||
# peer. It wraps apply.sh / verify.sh / rollback.sh in the C-09 contract.
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
# shellcheck source=lib/orca-log.sh
|
||||
. "$SCRIPT_DIR/lib/orca-log.sh"
|
||||
|
||||
# --- exit codes (C-09) ---
|
||||
EXIT_OK=0
|
||||
EXIT_APPLY_FAIL=1
|
||||
EXIT_VERIFY_FAIL=2
|
||||
EXIT_ROLLBACK_FAIL=3
|
||||
EXIT_INVALID_ARGS=4
|
||||
EXIT_ALREADY_APPLIED=5
|
||||
|
||||
# --- bounded retry (C-09) ---
|
||||
MAX_RETRIES=3
|
||||
BACKOFF_SEQ=(1 2 4)
|
||||
|
||||
# --- args ---
|
||||
TXN_DIR=""
|
||||
NAMESPACE=""
|
||||
FORCE=false
|
||||
ACK_RISK=false
|
||||
YES=false
|
||||
|
||||
usage() {
|
||||
cat >&2 <<EOF
|
||||
usage: orca-pull.sh --txn-dir <dir> [--namespace <ns>]
|
||||
[--force --i-understand-the-risk]
|
||||
[--force --yes]
|
||||
EOF
|
||||
}
|
||||
|
||||
while [ "$#" -gt 0 ]; do
|
||||
case "$1" in
|
||||
--txn-dir)
|
||||
[ "$#" -ge 2 ] || { orca_log_error "orca-pull" "-" "failed" "missing --txn-dir value"; usage; exit "$EXIT_INVALID_ARGS"; }
|
||||
TXN_DIR="$2"; shift 2 ;;
|
||||
--namespace)
|
||||
[ "$#" -ge 2 ] || { orca_log_error "orca-pull" "-" "failed" "missing --namespace value"; usage; exit "$EXIT_INVALID_ARGS"; }
|
||||
NAMESPACE="$2"; shift 2 ;;
|
||||
--force)
|
||||
FORCE=true; shift ;;
|
||||
--i-understand-the-risk)
|
||||
ACK_RISK=true; shift ;;
|
||||
--yes)
|
||||
YES=true; shift ;;
|
||||
-h|--help)
|
||||
usage; exit "$EXIT_OK" ;;
|
||||
*)
|
||||
orca_log_error "orca-pull" "-" "failed" "unknown argument: $1"
|
||||
usage; exit "$EXIT_INVALID_ARGS" ;;
|
||||
esac
|
||||
done
|
||||
|
||||
# --- arg validation ---
|
||||
if [ -z "$TXN_DIR" ]; then
|
||||
orca_log_error "orca-pull" "-" "failed" "missing --txn-dir"
|
||||
usage; exit "$EXIT_INVALID_ARGS"
|
||||
fi
|
||||
if [ ! -d "$TXN_DIR" ]; then
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "failed" "txn dir not found"
|
||||
exit "$EXIT_INVALID_ARGS"
|
||||
fi
|
||||
|
||||
# --- C-23 cluster-wide vs namespace-scoped enforcement ---
|
||||
if [ -z "$NAMESPACE" ]; then
|
||||
# Cluster-wide txn: requires --force + (--i-understand-the-risk | --yes).
|
||||
if [ "$FORCE" != "true" ]; then
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "denied" "cluster-wide txn requires --force"
|
||||
echo "error: cluster-wide txn requires --force (C-23)" >&2
|
||||
exit "$EXIT_INVALID_ARGS"
|
||||
fi
|
||||
if [ "$ACK_RISK" != "true" ] && [ "$YES" != "true" ]; then
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "denied" "cluster-wide --force requires --i-understand-the-risk (or --yes)"
|
||||
echo "error: cluster-wide --force requires --i-understand-the-risk (or --yes) (C-23)" >&2
|
||||
exit "$EXIT_INVALID_ARGS"
|
||||
fi
|
||||
else
|
||||
# Namespace-scoped: --force not required. Drift in other namespaces
|
||||
# does not block this txn (C-23). We still honor --force if given
|
||||
# (it's a no-op for ns-scoped).
|
||||
:
|
||||
fi
|
||||
|
||||
# --- locate bundle files ---
|
||||
APPLY="$TXN_DIR/apply.sh"
|
||||
VERIFY="$TXN_DIR/verify.sh"
|
||||
ROLLBACK="$TXN_DIR/rollback.sh"
|
||||
MARKER="$TXN_DIR/.applied"
|
||||
|
||||
for f in "$APPLY" "$VERIFY" "$ROLLBACK"; do
|
||||
if [ ! -f "$f" ]; then
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "failed" "missing bundle file: $f"
|
||||
echo "error: missing $f" >&2
|
||||
exit "$EXIT_INVALID_ARGS"
|
||||
fi
|
||||
done
|
||||
|
||||
# --- idempotency: already-applied is a no-op (C-09) ---
|
||||
if [ -f "$MARKER" ]; then
|
||||
orca_log_info "orca-pull" "$TXN_DIR" "already-applied" ""
|
||||
echo "already-applied"
|
||||
exit "$EXIT_ALREADY_APPLIED"
|
||||
fi
|
||||
|
||||
# --- apply with bounded retry (C-09: 3 attempts, 1s/2s/4s backoff) ---
|
||||
start_ns="$(date +%s%N)"
|
||||
|
||||
run_with_retry() {
|
||||
local script="$1" label="$2"
|
||||
local attempt=0
|
||||
local rc=0
|
||||
while [ "$attempt" -lt "$MAX_RETRIES" ]; do
|
||||
attempt=$((attempt + 1))
|
||||
set +e
|
||||
bash "$script"
|
||||
rc=$?
|
||||
set -e
|
||||
if [ "$rc" -eq 0 ]; then
|
||||
return 0
|
||||
fi
|
||||
if [ "$attempt" -eq "$MAX_RETRIES" ]; then
|
||||
break
|
||||
fi
|
||||
local wait_s="${BACKOFF_SEQ[$((attempt - 1))]}"
|
||||
orca_log_warn "orca-pull" "$TXN_DIR" "$label-retry" "attempt $attempt failed (rc=$rc), sleeping ${wait_s}s"
|
||||
sleep "$wait_s"
|
||||
done
|
||||
return "$rc"
|
||||
}
|
||||
|
||||
# Apply phase.
|
||||
if ! run_with_retry "$APPLY" "apply"; then
|
||||
apply_rc=$?
|
||||
end_ns="$(date +%s%N)"
|
||||
duration_ms=$(( (end_ns - start_ns) / 1000000 ))
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "apply-failed" "rc=$apply_rc duration_ms=$duration_ms"
|
||||
# Run rollback on apply failure.
|
||||
if bash "$ROLLBACK"; then
|
||||
:
|
||||
else
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "rollback-failed" "rollback after apply failure exited non-zero"
|
||||
exit "$EXIT_ROLLBACK_FAIL"
|
||||
fi
|
||||
exit "$EXIT_APPLY_FAIL"
|
||||
fi
|
||||
|
||||
# Verify phase.
|
||||
if ! run_with_retry "$VERIFY" "verify"; then
|
||||
verify_rc=$?
|
||||
end_ns="$(date +%s%N)"
|
||||
duration_ms=$(( (end_ns - start_ns) / 1000000 ))
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "verify-failed" "rc=$verify_rc duration_ms=$duration_ms"
|
||||
# Run rollback on verify failure.
|
||||
if ! bash "$ROLLBACK"; then
|
||||
orca_log_error "orca-pull" "$TXN_DIR" "rollback-failed" "rollback after verify failure exited non-zero"
|
||||
exit "$EXIT_ROLLBACK_FAIL"
|
||||
fi
|
||||
exit "$EXIT_VERIFY_FAIL"
|
||||
fi
|
||||
|
||||
end_ns="$(date +%s%N)"
|
||||
duration_ms=$(( (end_ns - start_ns) / 1000000 ))
|
||||
orca_log_info "orca-pull" "$TXN_DIR" "applied" "duration_ms=$duration_ms namespace=${NAMESPACE:-cluster-wide}"
|
||||
echo "applied (duration=${duration_ms}ms)"
|
||||
exit "$EXIT_OK"
|
||||
@@ -0,0 +1,112 @@
|
||||
#!/usr/bin/env bats
|
||||
# Tests for scripts/orca-pull.sh — the C-09 failure contract + C-23
|
||||
# cluster-wide vs namespace-scoped distinction.
|
||||
|
||||
load test_helper
|
||||
|
||||
PULL="$SCRIPTS_DIR/orca-pull.sh"
|
||||
TMP_TXN=""
|
||||
|
||||
setup() {
|
||||
TMP_TXN="$(mktemp -d)"
|
||||
# Minimal apply/verify/rollback scripts that succeed.
|
||||
cat >"$TMP_TXN/apply.sh" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
touch "$TMP_TXN/.applied-marker"
|
||||
exit 0
|
||||
EOF
|
||||
# A real marker the orca-pull.sh checks for.
|
||||
cat >"$TMP_TXN/apply.sh" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
touch "$(dirname "$0")/.applied"
|
||||
exit 0
|
||||
EOF
|
||||
cat >"$TMP_TXN/verify.sh" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
exit 0
|
||||
EOF
|
||||
cat >"$TMP_TXN/rollback.sh" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
rm -f "$(dirname "$0")/.applied"
|
||||
exit 0
|
||||
EOF
|
||||
chmod +x "$TMP_TXN"/*.sh
|
||||
}
|
||||
|
||||
teardown() {
|
||||
[ -n "$TMP_TXN" ] && rm -rf "$TMP_TXN"
|
||||
}
|
||||
|
||||
@test "orca-pull.sh exists and is executable" {
|
||||
[ -f "$PULL" ]
|
||||
[ -x "$PULL" ]
|
||||
}
|
||||
|
||||
@test "orca-pull.sh refuses without --txn-dir (exit 4)" {
|
||||
run "$PULL" --force --i-understand-the-risk
|
||||
assert_status 4 "$status"
|
||||
}
|
||||
|
||||
@test "orca-pull.sh cluster-wide without --force exits 4 (C-23)" {
|
||||
run "$PULL" --txn-dir "$TMP_TXN"
|
||||
assert_status 4 "$status"
|
||||
assert_contains "$output" "requires --force"
|
||||
}
|
||||
|
||||
@test "orca-pull.sh cluster-wide with --force but no ack exits 4 (C-23)" {
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --force
|
||||
assert_status 4 "$status"
|
||||
assert_contains "$output" "--i-understand-the-risk"
|
||||
}
|
||||
|
||||
@test "orca-pull.sh cluster-wide with --force --i-understand-the-risk applies" {
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --force --i-understand-the-risk
|
||||
assert_status 0 "$status"
|
||||
assert_contains "$output" "applied"
|
||||
[ -f "$TMP_TXN/.applied" ]
|
||||
}
|
||||
|
||||
@test "orca-pull.sh namespace-scoped applies without --force (C-23)" {
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --namespace default
|
||||
assert_status 0 "$status"
|
||||
assert_contains "$output" "applied"
|
||||
[ -f "$TMP_TXN/.applied" ]
|
||||
}
|
||||
|
||||
@test "orca-pull.sh idempotent re-run exits 5 (already-applied)" {
|
||||
# First apply.
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --namespace default
|
||||
assert_status 0 "$status"
|
||||
# Re-run: should be a no-op (exit 5).
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --namespace default
|
||||
assert_status 5 "$status"
|
||||
assert_contains "$output" "already-applied"
|
||||
}
|
||||
|
||||
@test "orca-pull.sh apply failure runs rollback and exits 1" {
|
||||
# Make apply.sh fail.
|
||||
cat >"$TMP_TXN/apply.sh" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
exit 1
|
||||
EOF
|
||||
chmod +x "$TMP_TXN/apply.sh"
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --namespace default
|
||||
[ "$status" -eq 1 ]
|
||||
}
|
||||
|
||||
@test "orca-pull.sh verify failure runs rollback and exits 2" {
|
||||
# apply succeeds, verify fails.
|
||||
cat >"$TMP_TXN/verify.sh" <<'EOF'
|
||||
#!/usr/bin/env bash
|
||||
exit 1
|
||||
EOF
|
||||
chmod +x "$TMP_TXN/verify.sh"
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --namespace default
|
||||
[ "$status" -eq 2 ]
|
||||
}
|
||||
|
||||
@test "orca-pull.sh missing bundle file exits 4" {
|
||||
rm -f "$TMP_TXN/verify.sh"
|
||||
run "$PULL" --txn-dir "$TMP_TXN" --namespace default
|
||||
assert_status 4 "$status"
|
||||
}
|
||||
Reference in New Issue
Block a user