orca/internal/cli/concurrency_test.go

package cli

// concurrency_test.go covers the REQ-156 / P07 concurrency-safety
// fixes:
//
//   - T11: concurrent `secrets set` on the same namespace preserves all
//     keys (the flock serializes the read-modify-write so no key is
//     lost to a clobbering second writer).
//   - T12: a second `orca upgrade` invoked while the first is running
//     is rejected with "upgrade already in progress".
//   - T13: cache invalidation read-after-write - `node join` followed
//     by an immediate `node list` (with a populated stale cache) shows
//     the new node, not the stale cached list.
//   - T14: (in internal/webauthn) concurrent BeginRegistration does
//     not panic / race on the session map.
//
// These tests complement the per-fix unit tests in the relevant
// _test.go files; they specifically exercise the cross-cutting
// concurrency invariants the milestone hardens.

import (
	"bytes"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"git.cloudinit.dev/coreci/orca/internal/cache"
	"git.cloudinit.dev/coreci/orca/internal/paths"
	"git.cloudinit.dev/coreci/orca/internal/secrets"
)

// runCLI is a helper that resets root flags, wires a fresh output
// buffer, sets the given args, and runs rootCmd. Returns the captured
// output. The buffer must be wired AFTER resetRootFlags (which sets
// its own buffer).
func runCLI(t *testing.T, args ...string) (string, error) {
	t.Helper()
	resetRootFlags(t)
	var buf bytes.Buffer
	rootCmd.SetOut(&buf)
	rootCmd.SetErr(&buf)
	rootCmd.SetArgs(args)
	err := rootCmd.Execute()
	return buf.String(), err
}

// ---------------------------------------------------------------------------
// T11: concurrent secrets set preserves all keys
// ---------------------------------------------------------------------------

// TestSecretsConcurrentSetPreservesAllKeys runs 5 concurrent
// `orca secrets set` invocations against the SAME namespace, each
// setting a distinct key. Without the flock (P07 T2) the second writer
// would load-then-save and clobber the first, losing a key. With the
// flock all 5 keys must be present afterward.
//
// The cobra rootCmd is a package global and is NOT goroutine-safe
// (shared flag state), so we drive the secrets-set RunE body directly
// under real concurrency. This exercises the lockNSSecrets flock +
// loadMasterAndNSSecrets + saveNSSecrets path that the RunE uses.
func TestSecretsConcurrentSetPreservesAllKeys(t *testing.T) {
	ns := "concsetns"
	setupSecretsTestEnv(t, ns)

	const n = 5
	keys := make([]string, n)
	for i := 0; i < n; i++ {
		keys[i] = fmt.Sprintf("KEY_%d", i)
	}

	var wg sync.WaitGroup
	errs := make([]error, n)
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			// Replicate the secretsSetCmd RunE body under real
			// concurrency: lock -> load -> mutate -> save. The lock
			// serializes the read-modify-write so concurrent sets do
			// not clobber each other.
			release, err := lockNSSecrets(ns)
			if err != nil {
				errs[idx] = fmt.Errorf("lock: %w", err)
				return
			}
			defer release()
			nsKey, lines, err := loadMasterAndNSSecrets(ns)
			if err != nil {
				errs[idx] = err
				return
			}
			defer secrets.ZeroKey(nsKey)
			key := keys[idx]
			value := fmt.Sprintf("value_%d", idx)
			newLine := key + "=" + value
			j := findKeyIndex(lines, key)
			if j >= 0 {
				lines[j] = newLine
			} else {
				lines = append(lines, newLine)
			}
			errs[idx] = saveNSSecrets(ns, nsKey, lines)
		}(i)
	}
	wg.Wait()

	for i, err := range errs {
		if err != nil {
			t.Fatalf("goroutine %d: %v", i, err)
		}
	}

	// All 5 keys must be present.
	out, err := runCLI(t, "secrets", "list", ns)
	if err != nil {
		t.Fatalf("secrets list: %v", err)
	}
	for _, k := range keys {
		if !strings.Contains(out, k) {
			t.Errorf("key %q missing after concurrent set (flock did not serialize): %s", k, out)
		}
	}
}

// TestSecretsConcurrentSetViaCLI is the cobra-driven variant. cobra's
// rootCmd is not goroutine-safe (shared flag globals), so we serialize
// the Execute() calls. This still exercises the flock because the
// load+save happens inside RunE. Confirms the CLI path itself (with
// flock) does not lose keys under repeated serial sets.
func TestSecretsConcurrentSetViaCLI(t *testing.T) {
	ns := "conccli"
	setupSecretsTestEnv(t, ns)

	const n = 5
	for i := 0; i < n; i++ {
		if _, err := runCLI(t, "secrets", "set", ns, fmt.Sprintf("K_%d=v_%d", i, i)); err != nil {
			t.Fatalf("secrets set %d: %v", i, err)
		}
	}
	out, err := runCLI(t, "secrets", "list", ns)
	if err != nil {
		t.Fatalf("secrets list: %v", err)
	}
	for i := 0; i < n; i++ {
		k := fmt.Sprintf("K_%d", i)
		if !strings.Contains(out, k) {
			t.Errorf("key %q missing after serial CLI sets: %s", k, out)
		}
	}
}

// ---------------------------------------------------------------------------
// T12: concurrent upgrade rejection
// ---------------------------------------------------------------------------

// TestUpgradeConcurrentLockRejected verifies that a second upgrade
// invocation while the first holds the upgrade.lock is rejected with
// "upgrade already in progress".
func TestUpgradeConcurrentLockRejected(t *testing.T) {
	setupUpgradeTest(t)
	resetUpgradeFlags()

	// Manually create the upgrade.lock as if a first upgrade is in
	// progress (the lock file content is just diagnostic; its
	// EXISTENCE is what blocks the second caller via O_CREATE|O_EXCL).
	lockPath := filepath.Join(paths.ClusterDir(), "upgrade.lock")
	if err := os.MkdirAll(filepath.Dir(lockPath), 0o755); err != nil {
		t.Fatalf("mkdir cluster: %v", err)
	}
	if err := os.WriteFile(lockPath, []byte("pid=999 started=2026-01-01T00:00:00Z\n"), 0o600); err != nil {
		t.Fatalf("write lock: %v", err)
	}
	defer os.Remove(lockPath)

	// A dry-run upgrade must now be rejected because the lock exists.
	_, err := runCLI(t, "upgrade", "--to", "v0.11.0", "--dry-run")
	if err == nil {
		t.Fatal("upgrade with stale lock should fail, got nil")
	}
	if !strings.Contains(err.Error(), "upgrade already in progress") {
		t.Errorf("unexpected error: %v", err)
	}
}

// TestUpgradeLockReleasedOnSuccess verifies the upgrade.lock is
// removed after a successful (dry-run) upgrade so a subsequent upgrade
// is not blocked by a stale lock.
func TestUpgradeLockReleasedOnSuccess(t *testing.T) {
	setupUpgradeTest(t)
	setupUpgradeTestWithMocks(t)
	rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--dry-run"})
	if err := rootCmd.Execute(); err != nil {
		t.Fatalf("upgrade dry-run: %v", err)
	}
	lockPath := filepath.Join(paths.ClusterDir(), "upgrade.lock")
	if _, err := os.Stat(lockPath); err == nil {
		t.Errorf("upgrade.lock still exists after successful dry-run (not released): %s", lockPath)
	}
}

// TestUpgradeLockReleasedOnError verifies the lock is released even
// when the upgrade fails mid-run (the defer in runUpgrade covers the
// error path).
func TestUpgradeLockReleasedOnError(t *testing.T) {
	setupUpgradeTest(t)
	setupUpgradeTestWithMocks(t)
	// Force a failure: --to with a version that triggers a cutover
	// whose verification fails. The runner reports :443 (cutover
	// needed) and the http check returns 502 (verification fail).
	runner := &mockUpgradeRunner{
		outputs: map[string][]byte{
			"ss -tlnp": []byte(":443"),
		},
	}
	upgradeRunnerOverride = runner
	httpClientOverride = func(url string) (int, error) { return 502, nil }
	rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0"})
	_ = rootCmd.Execute() // expected to fail
	lockPath := filepath.Join(paths.ClusterDir(), "upgrade.lock")
	if _, err := os.Stat(lockPath); err == nil {
		t.Errorf("upgrade.lock still exists after failed upgrade (not released on error): %s", lockPath)
	}
}

// ---------------------------------------------------------------------------
// T13: cache invalidation read-after-write
// ---------------------------------------------------------------------------

// TestCacheInvalidationNodeJoinReadAfterWrite verifies that after
// `node join` invalidates the `nodes` cache class, an immediate
// `node list` (which would otherwise serve a STALE cached list) shows
// the just-joined node.
//
// Setup: populate the cache with a stale nodes list (missing the new
// node). Without T5's invalidation, the second `node list` would serve
// the stale list and the new node would be invisible until the TTL
// expired. With T5, the join invalidates the class and the list
// re-reads from the DB.
func TestCacheInvalidationNodeJoinReadAfterWrite(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()

	// Seed the cache with a stale nodes list (a sentinel node that
	// does NOT exist in the DB). The TTL is long so it would be
	// served on a subsequent list without invalidation.
	c, err := cache.Open(paths.CacheDB())
	if err != nil {
		t.Fatalf("open cache: %v", err)
	}
	stale := `[{"id":"stale-id","name":"stale-node","address":"10.0.0.99:8443","state":"ready"}]`
	if err := c.Set(cacheNodeClass, cacheListKey, []byte(stale), 10*time.Minute); err != nil {
		t.Fatalf("set stale cache: %v", err)
	}
	c.Close()

	// Confirm the stale entry is served by a fresh list (proving the
	// cache is populated and would be hit).
	staleOut, err := runCLI(t, "node", "list")
	if err != nil {
		t.Fatalf("stale node list: %v", err)
	}
	if !strings.Contains(staleOut, "stale-node") {
		t.Fatalf("precondition: stale cache not served: %s", staleOut)
	}

	// Join a real node. T5 invalidates the `nodes` cache class.
	if _, err := runCLI(t, "node", "join", "--name", "freshnode", "--addr", "10.0.0.42:8443"); err != nil {
		t.Fatalf("node join: %v", err)
	}

	// Immediate list: the stale sentinel must be GONE (invalidated)
	// and the real fresh node must be present (read from the DB).
	out, err := runCLI(t, "node", "list")
	if err != nil {
		t.Fatalf("node list after join: %v", err)
	}
	if strings.Contains(out, "stale-node") {
		t.Errorf("stale cache still served after join (invalidation missing): %s", out)
	}
	if !strings.Contains(out, "freshnode") {
		t.Errorf("fresh node missing from list after join (cache not re-read): %s", out)
	}
}

// TestCacheInvalidationNSCreateReadAfterWrite is the ns variant: a
// stale `namespaces` cache is invalidated by `ns create` so the next
// `ns list` shows the new namespace.
func TestCacheInvalidationNSCreateReadAfterWrite(t *testing.T) {
	root := t.TempDir()
	t.Setenv("ORCA_HOME", root)
	writeDefaultsNS(t, root)

	// Seed a stale namespaces cache containing only _defaults.
	c, err := cache.Open(paths.CacheDB())
	if err != nil {
		t.Fatalf("open cache: %v", err)
	}
	stale := `[{"name":"_defaults","path":"` + filepath.Join(root, "_defaults") + `","default":true}]`
	if err := c.Set(cacheNamespaceClass, cacheListKey, []byte(stale), 10*time.Minute); err != nil {
		t.Fatalf("set stale: %v", err)
	}
	c.Close()

	// Confirm stale served.
	resetRootFlags(t)
	resetNSFlags()
	staleOut, err := runCLI(t, "ns", "list")
	if err != nil {
		t.Fatalf("stale ns list: %v", err)
	}
	if !strings.Contains(staleOut, "_defaults") {
		t.Fatalf("precondition: stale ns cache not served: %s", staleOut)
	}

	// Create a new namespace. T5 invalidates the `namespaces` cache.
	resetRootFlags(t)
	resetNSFlags()
	if _, err := runCLI(t, "ns", "create", "newns"); err != nil {
		t.Fatalf("ns create: %v", err)
	}

	// Immediate list: must show the new namespace (read from disk,
	// not the stale cache).
	resetRootFlags(t)
	resetNSFlags()
	out, err := runCLI(t, "ns", "list")
	if err != nil {
		t.Fatalf("ns list after create: %v", err)
	}
	if !strings.Contains(out, "newns") {
		t.Errorf("new namespace missing from list after create (cache not invalidated/re-read): %s", out)
	}
}

// TestCacheInvalidationJobRunReadAfterWrite verifies `job run`
// invalidates the `jobs` cache so a stale cached job list is not
// served after a new job runs.
func TestCacheInvalidationJobRunReadAfterWrite(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()

	// Seed a stale jobs cache (a sentinel job that does not exist).
	c, err := cache.Open(paths.CacheDB())
	if err != nil {
		t.Fatalf("open cache: %v", err)
	}
	stale := `[{"id":"stale-job","name":"stale","status":"complete","exit_code":0}]`
	if err := c.Set(cacheJobClass, cacheListKey, []byte(stale), 10*time.Minute); err != nil {
		t.Fatalf("set stale: %v", err)
	}
	c.Close()

	// Confirm stale served.
	staleOut, err := runCLI(t, "job", "list")
	if err != nil {
		t.Fatalf("stale job list: %v", err)
	}
	if !strings.Contains(staleOut, "stale") {
		t.Fatalf("precondition: stale job cache not served: %s", staleOut)
	}

	// Write a job spec and run it. T5 invalidates the `jobs` cache.
	specDir := t.TempDir()
	specPath := filepath.Join(specDir, "job.md")
	specBody := "---\n" +
		"kind: Job\n" +
		"name: cacheinv-job\n" +
		"runtime:\n" +
		"  one_of: process\n" +
		"  command: /bin/true\n" +
		"---\n# cacheinv\n\nRuns /bin/true.\n"
	if err := os.WriteFile(specPath, []byte(specBody), 0o644); err != nil {
		t.Fatalf("write spec: %v", err)
	}
	if _, err := runCLI(t, "job", "run", specPath); err != nil {
		t.Fatalf("job run: %v", err)
	}

	// Immediate list: the stale sentinel must be gone; the real job
	// must be present (read from the DB).
	out, err := runCLI(t, "job", "list")
	if err != nil {
		t.Fatalf("job list after run: %v", err)
	}
	if strings.Contains(out, "stale-job") {
		t.Errorf("stale job cache still served after run (invalidation missing): %s", out)
	}
	if !strings.Contains(out, "cacheinv-job") {
		t.Errorf("new job missing from list after run (cache not re-read): %s", out)
	}
}

// TestCacheInvalidateHelperDirectly is a small unit test for the
// cacheInvalidate helper itself: it confirms a populated class is
// empty after the helper runs.
func TestCacheInvalidateHelperDirectly(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	c, err := cache.Open(paths.CacheDB())
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	if err := c.Set(cacheNodeClass, cacheListKey, []byte("x"), 0); err != nil {
		t.Fatalf("set: %v", err)
	}
	c.Close()

	cacheInvalidate(cacheNodeClass)

	c2, err := cache.Open(paths.CacheDB())
	if err != nil {
		t.Fatalf("reopen: %v", err)
	}
	defer c2.Close()
	if _, _, err := c2.Get(cacheNodeClass, cacheListKey); err == nil {
		t.Errorf("nodes/list still present after cacheInvalidate")
	}
}