Merge pull request #29 from solinode/fix-ci-races

Fix CI: data races and flaky logs tail test

Author: drk1rd

cli/internal/awsapi/awstest/awstest.go

@@ -22,6 +22,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -44,8 +45,22 @@ type Harness struct {
 	// Mux serves the native HomeCloud API (register routes with Router).
 	Mux    *http.ServeMux
 	Router *httpx.Router
-	// Audit records every audited call (action, resource).
-	Audit []string
+	// Audit records every audited call (action, resource); read it with AuditLog.
+	Audit   []string
+	auditMu sync.Mutex
+	// barrier orders a test's setup writes (fields set on services after New)
+	// before the requests that read them; the race detector can't see that
+	// ordering through the AWS CLI subprocess.
+	barrier sync.Mutex
+}
+
+func (h *Harness) sync() { h.barrier.Lock(); h.barrier.Unlock() }
+
+// AuditLog returns a copy of the audited calls so far.
+func (h *Harness) AuditLog() []string {
+	h.auditMu.Lock()
+	defer h.auditMu.Unlock()
+	return append([]string(nil), h.Audit...)
 }
 
 // New starts an endpoint with IAM (root user) and STS.
@@ -65,12 +80,15 @@ func New(t *testing.T) *Harness {
 	im.RegisterAWS()
 	h := &Harness{Env: env, IAM: im, Secrets: sec, AccessKeyID: boot.AccessKeyID, SecretKey: boot.SecretKey, Mux: http.NewServeMux()}
 	audit := func(p *httpx.Principal, action, resource string, r *http.Request, status int, took time.Duration) {
+		h.auditMu.Lock()
 		h.Audit = append(h.Audit, action+" "+resource)
+		h.auditMu.Unlock()
 	}
 	h.Router = &httpx.Router{Mux: h.Mux, Auth: im, Account: env.AccountID, Audit: audit}
 	im.Routes(h.Router)
 	aws := &awsapi.Handler{Creds: im, Account: env.AccountID, Audit: audit}
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		h.sync()
 		if awsapi.Match(r) && !strings.HasPrefix(r.Header.Get("Authorization"), "Bearer ") {
 			aws.ServeHTTP(w, r)
 			return
@@ -130,6 +148,7 @@ func (h *Harness) AWSErr(t *testing.T, args ...string) (string, error) {
 // AWSAs runs the AWS CLI with the given credentials.
 func (h *Harness) AWSAs(t *testing.T, akid, secret, token string, args ...string) (string, error) {
 	t.Helper()
+	h.sync()
 	cmd := exec.Command(awsBin(t), append([]string{"--output", "json"}, args...)...)
 	cmd.Env = h.Environ(akid, secret, token)
 	var out bytes.Buffer
@@ -173,6 +192,7 @@ func (h *Harness) Python(t *testing.T, script string) string {
 	if err := os.WriteFile(f, []byte(src), 0o600); err != nil {
 		t.Fatal(err)
 	}
+	h.sync()
 	cmd := exec.Command(python(t), f)
 	cmd.Env = h.Environ(h.AccessKeyID, h.SecretKey, "")
 	var out, errb bytes.Buffer
@@ -215,6 +235,7 @@ func (h *Harness) nativeAsRoot(t *testing.T, method, path string, body any) []by
 	} else {
 		rd = bytes.NewReader(nil)
 	}
+	h.sync()
 	req, _ := http.NewRequest(method, h.URL+path, rd)
 	req.Header.Set("Authorization", "Bearer "+h.AccessKeyID+":"+h.SecretKey)
 	req.Header.Set("Content-Type", "application/json")

cli/internal/svc/cloudwatch/aws_test.go

@@ -3,6 +3,7 @@ package cloudwatch
 import (
 	"bytes"
 	"encoding/json"
+	"os"
 	"os/exec"
 	"strconv"
 	"strings"
@@ -171,12 +172,21 @@ func TestLogsTailFollow(t *testing.T) {
 	defer cmd.Process.Kill()
 	time.Sleep(2 * time.Second)
 	h.AWS(t, "logs", "put-log-events", "--log-group-name", "follow", "--log-stream-name", "s", "--log-events", "timestamp="+msNow(0)+",message=second line")
-	deadline := time.Now().Add(15 * time.Second)
+	deadline := time.Now().Add(25 * time.Second)
 	for time.Now().Before(deadline) && !strings.Contains(out.String(), "second line") {
 		time.Sleep(250 * time.Millisecond)
 	}
-	_ = cmd.Process.Kill()
-	_ = cmd.Wait()
+	// Interrupt rather than kill: the CLI may block-buffer stdout when it isn't a
+	// terminal (the bundled AWS CLI ignores PYTHONUNBUFFERED) and flushes on exit.
+	_ = cmd.Process.Signal(os.Interrupt)
+	done := make(chan struct{})
+	go func() { _ = cmd.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		_ = cmd.Process.Kill()
+		<-done
+	}
 	if s := out.String(); !strings.Contains(s, "first line") || !strings.Contains(s, "second line") || strings.Count(s, "first line") != 1 {
 		t.Fatalf("tail --follow output:\n%s", s)
 	}

cli/internal/svc/dynamodb/aws_boto3_test.go

@@ -316,12 +316,12 @@ func TestIAMReadOnlyDeniedWrites(t *testing.T) {
 		t.Fatalf("reader transact: %v %s", err, out)
 	}
 	found := false
-	for _, a := range h.Audit {
+	for _, a := range h.AuditLog() {
 		if strings.HasPrefix(a, "dynamodb:PutItem arn:aws:dynamodb:") && strings.HasSuffix(a, ":table/acl") {
 			found = true
 		}
 	}
 	if !found {
-		t.Fatalf("audit: %v", h.Audit)
+		t.Fatalf("audit: %v", h.AuditLog())
 	}
 }

cli/internal/svc/secrets/aws_test.go

@@ -50,6 +50,12 @@ func expectErr(t *testing.T, h *awstest.Harness, code string, args ...string) {
 
 func TestSecretsManagerCLI(t *testing.T) {
 	h, sec, _, clk := setup(t)
+	// Rotations run in the background; let them finish before the test ends.
+	t.Cleanup(func() {
+		for deadline := time.Now().Add(60 * time.Second); sec.Rotating() && time.Now().Before(deadline); {
+			time.Sleep(50 * time.Millisecond)
+		}
+	})
 	arnRe := regexp.MustCompile(`^arn:aws:secretsmanager:us-east-1:` + h.Env.AccountID + `:secret:app/db-[A-Za-z0-9]{6}$`)
 
 	c := h.AWSJSON(t, "secretsmanager", "create-secret", "--name", "app/db", "--secret-string", `{"user":"a","pass":"1"}`,
@@ -372,7 +378,7 @@ func TestSecretRotation(t *testing.T) {
 	var mu sync.Mutex
 	var steps []string
 	failTest := false
-	sec.Lambda = secrets.InvokerFunc(func(ctx context.Context, name string, payload []byte) ([]byte, string, error) {
+	sec.SetLambda(secrets.InvokerFunc(func(ctx context.Context, name string, payload []byte) ([]byte, string, error) {
 		var ev struct{ Step string }
 		_ = json.Unmarshal(payload, &ev)
 		mu.Lock()
@@ -390,7 +396,7 @@ func TestSecretRotation(t *testing.T) {
 			return []byte(`{"errorMessage":` + strings.TrimSpace(strconvQuote(out.String())) + `}`), "Unhandled", nil
 		}
 		return []byte("null"), "", nil
-	})
+	}))
 
 	h.AWS(t, "secretsmanager", "create-secret", "--name", "rot", "--secret-string", "initial")
 	fn := "arn:aws:lambda:us-east-1:" + h.Env.AccountID + ":function:rotator"

cli/internal/svc/secrets/native_test.go

@@ -11,7 +11,7 @@ import (
 func TestSecretsNativeRotationPolicyStages(t *testing.T) {
 	h, sec, _, _ := setup(t)
 	sec.Routes(h.Router)
-	sec.Lambda = secrets.InvokerFunc(func(context.Context, string, []byte) ([]byte, string, error) { return []byte("null"), "", nil })
+	sec.SetLambda(secrets.InvokerFunc(func(context.Context, string, []byte) ([]byte, string, error) { return []byte("null"), "", nil }))
 	get := func(method, path string, body any) map[string]any {
 		var m map[string]any
 		if err := json.Unmarshal(h.Native(t, method, path, body), &m); err != nil {

cli/internal/svc/secrets/rotation.go

@@ -103,7 +103,7 @@ func (s *Service) rotate(az Authz, ref string, in RotateInput) (Secret, string,
 	if fn == "" {
 		return sec, "", core.Errf(http.StatusBadRequest, "InvalidRequestException", "No Lambda rotation function ARN is associated with this secret.")
 	}
-	if s.Lambda == nil {
+	if s.lambda() == nil {
 		return sec, "", core.Errf(http.StatusBadRequest, "InvalidRequestException", "Lambda rotation functions are not available on this server.")
 	}
 	if !strings.HasPrefix(fn, "arn:") {
@@ -167,6 +167,13 @@ func (s *Service) pendingRotation(sec Secret) bool {
 
 // startRotation adds the AWSPENDING placeholder version and runs the rotation
 // function's four steps in the background.
+// Rotating reports whether any rotation is still in progress.
+func (s *Service) Rotating() bool {
+	s.rotMu.Lock()
+	defer s.rotMu.Unlock()
+	return len(s.rotating) > 0
+}
+
 func (s *Service) startRotation(name, token string) (string, error) {
 	if token == "" {
 		token = uuid()
@@ -222,7 +229,12 @@ func (s *Service) runRotation(sec Secret, token string) {
 	}
 	for _, step := range []string{"createSecret", "setSecret", "testSecret", "finishSecret"} {
 		payload, _ := json.Marshal(map[string]string{"Step": step, "SecretId": sec.ARN, "ClientRequestToken": token, "RotationToken": uuid()})
-		out, fnErr, err := s.Lambda.Invoke(ctx, functionName(sec.RotationLambdaARN), payload)
+		inv := s.lambda()
+		if inv == nil {
+			fail(step + ": Lambda rotation functions are not available on this server")
+			return
+		}
+		out, fnErr, err := inv.Invoke(ctx, functionName(sec.RotationLambdaARN), payload)
 		if err != nil {
 			fail(fmt.Sprintf("%s: %s", step, errMessage(err)))
 			return

cli/internal/svc/secrets/secrets.go

@@ -118,6 +118,7 @@ type Service struct {
 	KMS KeyService
 	// Lambda runs rotation functions; nil disables rotation.
 	Lambda Invoker
+	cfgMu  sync.RWMutex // guards Lambda once the service is serving
 
 	rotMu    sync.Mutex
 	rotating map[string]bool
@@ -380,7 +381,7 @@ func (s *Service) PurgeExpired() {
 	s.purgeDeleted()
 	now := s.now()
 	for _, sec := range store.List[Secret](s.env.Store, cSecrets) {
-		if s.Lambda != nil && sec.DeletionDate == nil && sec.RotationEnabled && sec.NextRotation != nil && !now.Before(*sec.NextRotation) && !s.pendingRotation(sec) {
+		if s.lambda() != nil && sec.DeletionDate == nil && sec.RotationEnabled && sec.NextRotation != nil && !now.Before(*sec.NextRotation) && !s.pendingRotation(sec) {
 			_, _ = s.startRotation(sec.Name, "")
 		}
 	}
@@ -403,3 +404,16 @@ func uuid() string {
 	h := fmt.Sprintf("%x", b)
 	return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
 }
+
+// SetLambda replaces the rotation invoker while the service is running.
+func (s *Service) SetLambda(inv Invoker) {
+	s.cfgMu.Lock()
+	s.Lambda = inv
+	s.cfgMu.Unlock()
+}
+
+func (s *Service) lambda() Invoker {
+	s.cfgMu.RLock()
+	defer s.cfgMu.RUnlock()
+	return s.Lambda
+}

cli/internal/svc/sns/aws_test.go

@@ -60,7 +60,7 @@ func harness(t *testing.T) (*awstest.Harness, *Service, *fakeLambda) {
 	s.Routes(h.Router)
 	s.RegisterAWS()
 	// Tests deliver to endpoints on 127.0.0.1.
-	s.HTTP, s.CheckURL = &http.Client{Timeout: 5 * time.Second}, func(string) error { return nil }
+	s.SetHTTP(&http.Client{Timeout: 5 * time.Second}, func(string) error { return nil })
 	return h, s, fl
 }
 

cli/internal/svc/sns/sns.go

@@ -118,6 +118,7 @@ type Service struct {
 	// Both refuse loopback and link-local addresses by default.
 	HTTP     *http.Client
 	CheckURL func(string) error
+	cfgMu    sync.RWMutex // guards HTTP and CheckURL once the service is serving
 
 	mu    sync.Mutex
 	dedup map[string]map[string]dedupEntry // FIFO topic -> dedup ID
@@ -463,7 +464,7 @@ func (s *Service) subscribe(t Topic, in subscribeInput, authorize func(action, r
 		if err != nil || u.Scheme != in.Protocol || u.Host == "" {
 			return Subscription{}, errInvalid("Endpoint must match the specified protocol")
 		}
-		if err := s.CheckURL(in.Endpoint); err != nil {
+		if err := s.checkURL(in.Endpoint); err != nil {
 			return Subscription{}, errInvalid("Endpoint: %v", err)
 		}
 	case "email", "email-json":
@@ -1049,7 +1050,7 @@ func (s *Service) post(sub Subscription, body []byte, typ, id string, raw bool)
 	if raw {
 		req.Header.Set("x-amz-sns-rawdelivery", "true")
 	}
-	resp, err := s.HTTP.Do(req)
+	resp, err := s.client().Do(req)
 	if err != nil {
 		return err
 	}
@@ -1063,12 +1064,12 @@ func (s *Service) post(sub Subscription, body []byte, typ, id string, raw bool)
 // Notify delivers an alarm-style notification to a topic ARN or an http(s) URL.
 func (s *Service) Notify(target, subject, message string) {
 	if strings.HasPrefix(target, "http://") || strings.HasPrefix(target, "https://") {
-		if err := s.CheckURL(target); err != nil {
+		if err := s.checkURL(target); err != nil {
 			log.Printf("notify %s: %v", target, err)
 			return
 		}
 		body, _ := json.Marshal(map[string]string{"subject": subject, "message": message})
-		resp, err := s.HTTP.Post(target, "application/json", bytes.NewReader(body))
+		resp, err := s.client().Post(target, "application/json", bytes.NewReader(body))
 		if err != nil {
 			log.Printf("notify %s: %v", target, err)
 			return
@@ -1415,3 +1416,23 @@ func (s *Service) cert(c *httpx.Ctx) (any, error) {
 	c.MarkWritten()
 	return nil, nil
 }
+
+// SetHTTP replaces the delivery client and URL check while the service is running.
+func (s *Service) SetHTTP(c *http.Client, check func(string) error) {
+	s.cfgMu.Lock()
+	s.HTTP, s.CheckURL = c, check
+	s.cfgMu.Unlock()
+}
+
+func (s *Service) client() *http.Client {
+	s.cfgMu.RLock()
+	defer s.cfgMu.RUnlock()
+	return s.HTTP
+}
+
+func (s *Service) checkURL(u string) error {
+	s.cfgMu.RLock()
+	f := s.CheckURL
+	s.cfgMu.RUnlock()
+	return f(u)
+}

cli/internal/svc/sqs/aws_test.go

@@ -442,11 +442,11 @@ func TestIAMDenied(t *testing.T) {
 		t.Fatalf("send without permission: %v %s", err, out)
 	}
 	found := false
-	for _, a := range h.Audit {
+	for _, a := range h.AuditLog() {
 		found = found || a == "sqs:SendMessage arn:aws:sqs:us-east-1:"+h.Env.AccountID+":locked"
 	}
 	if !found {
-		t.Fatalf("audit %v", h.Audit)
+		t.Fatalf("audit %v", h.AuditLog())
 	}
 }