test: improve e2e invoice diagnostics (#4598)

Author: turip

.agents/skills/e2e/SKILL.md

@@ -241,6 +241,8 @@ Kafka is in the path for ingestion. Don't assert the meter value immediately aft
 
 Reference: `e2e/e2e_test.go` `TestIngest`.
 
+For async billing flows, make timeout failures self-diagnosing. Log the created resource IDs that connect the test to service logs (customer, subscription, invoice, pending line, charge when available), and during the polling loop log the externally visible state as JSON, such as invoice list entries and customer charge statuses. The E2E CI job streams docker compose logs during each test phase and archives per-service docker logs before each compose teardown/restart; keep those artifacts broad enough that a failed async hop can be correlated from the test output to the relevant service log.
+
 ## Testing conventions
 
 - **`require` vs `assert`**: `require` for fatal preconditions (no point continuing), `assert` for soft per-field checks. In table rows, use `assert.Equal(t, tc.expectedStatus, status, "%+v", problem)` for the status check so the subsequent body-shape assertion still fires and surfaces in the same failure. Reserve `require` for lifecycle tests where later steps depend on the earlier status being correct.

.github/workflows/ci.yaml

@@ -758,9 +758,18 @@ jobs:
           cat e2e/docker-compose.override.yaml
 
       - name: Launch Docker Compose infra
+        id: launch_e2e
         run: docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml up -d
         working-directory: e2e
 
+      - name: Start Docker Compose log stream for base tests
+        if: always() && steps.launch_e2e.outcome != 'skipped'
+        working-directory: e2e
+        run: |
+          mkdir -p artifacts/logs/docker-compose/base
+          docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml logs --no-color --timestamps --follow > artifacts/logs/docker-compose/base/compose-follow.log 2>&1 &
+          echo "$!" > artifacts/logs/docker-compose/base/compose-follow.pid
+
       - name: Set up Nix
         uses: nixbuild/nix-quick-install-action@9f63be77f412a248c9d9a65a4c82cf066cdf8f0c # v35
         with:
@@ -803,31 +812,70 @@ jobs:
           docker ps
 
       - name: Run base tests
+        id: run_base_tests
         env:
           OPENMETER_ADDRESS: http://localhost:38888
           TZ: UTC
         run: |
           nix develop --impure .#ci -c make -C e2e test-base
 
+      - name: Capture Docker Compose logs after base tests
+        if: always() && steps.launch_e2e.outcome != 'skipped'
+        working-directory: e2e
+        run: |
+          mkdir -p artifacts/logs/docker-compose/base
+          if [ -f artifacts/logs/docker-compose/base/compose-follow.pid ]; then
+            kill "$(cat artifacts/logs/docker-compose/base/compose-follow.pid)" 2>/dev/null || true
+            sleep 1
+          fi
+          docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml ps --all > artifacts/logs/docker-compose/base/compose-ps.txt 2>&1 || true
+          for service in $(docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml config --services); do
+            docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml logs --no-color --timestamps "$service" > "artifacts/logs/docker-compose/base/${service}.log" 2>&1 || true
+          done
+
       - name: Restart Docker Compose infra for credits-disabled tests
+        id: restart_credits_disabled_e2e
         run: |
           docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml down -v
           docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml -f docker-compose.credits-disabled.yaml up -d
         working-directory: e2e
 
+      - name: Start Docker Compose log stream for credits-disabled tests
+        if: always() && steps.restart_credits_disabled_e2e.outcome != 'skipped'
+        working-directory: e2e
+        run: |
+          mkdir -p artifacts/logs/docker-compose/credits-disabled
+          docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml -f docker-compose.credits-disabled.yaml logs --no-color --timestamps --follow > artifacts/logs/docker-compose/credits-disabled/compose-follow.log 2>&1 &
+          echo "$!" > artifacts/logs/docker-compose/credits-disabled/compose-follow.pid
+
       - name: Wait for worker to become ready with credits disabled
         run: |
           curl --fail --retry 10 --retry-max-time 120 --retry-all-errors http://localhost:30000/healthz
           curl --fail --retry 10 --retry-max-time 120 --retry-all-errors http://localhost:30001/healthz
           docker ps
 
       - name: Run credits-disabled tests
+        id: run_credits_disabled_tests
         env:
           OPENMETER_ADDRESS: http://localhost:38888
           TZ: UTC
         run: |
           nix develop --impure .#ci -c make -C e2e test-credits-disabled
 
+      - name: Capture Docker Compose logs after credits-disabled tests
+        if: always() && steps.restart_credits_disabled_e2e.outcome != 'skipped'
+        working-directory: e2e
+        run: |
+          mkdir -p artifacts/logs/docker-compose/credits-disabled
+          if [ -f artifacts/logs/docker-compose/credits-disabled/compose-follow.pid ]; then
+            kill "$(cat artifacts/logs/docker-compose/credits-disabled/compose-follow.pid)" 2>/dev/null || true
+            sleep 1
+          fi
+          docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml -f docker-compose.credits-disabled.yaml ps --all > artifacts/logs/docker-compose/credits-disabled/compose-ps.txt 2>&1 || true
+          for service in $(docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml -f docker-compose.credits-disabled.yaml config --services); do
+            docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml -f docker-compose.credits-disabled.yaml logs --no-color --timestamps "$service" > "artifacts/logs/docker-compose/credits-disabled/${service}.log" 2>&1 || true
+          done
+
       - name: Cleanup Docker Compose
         run: docker compose -f docker-compose.infra.yaml -f docker-compose.openmeter.yaml -f docker-compose.override.yaml down -v
         working-directory: e2e
@@ -839,6 +887,6 @@ jobs:
         with:
           name: "[${{ github.job }}] Openmeter logs"
           path: |
-            e2e/logs/openmeter/openmeter.log
-            e2e/logs/sink-worker/openmeter.log
+            e2e/logs/**
+            e2e/artifacts/logs/**
           retention-days: 14

e2e/.gitignore

@@ -1,2 +1,3 @@
 logs
+artifacts
 docker-compose.override.yaml

e2e/billinginvoices_v3_test.go

@@ -1,8 +1,10 @@
 package e2e
 
 import (
+	"encoding/json"
 	"net/http"
 	"slices"
+	"strconv"
 	"testing"
 	"time"
 
@@ -188,21 +190,39 @@ func TestV3GetBillingInvoice(t *testing.T) {
 		require.Equal(t, http.StatusCreated, lineResp.StatusCode(), "line: %s", string(lineResp.Body))
 		require.NotNil(t, lineResp.JSON201)
 
+		t.Logf("created standard pending invoice line: customer_id=%s invoice_id=%s lines=%s",
+			customerID,
+			lineResp.JSON201.Invoice.Id,
+			formatInvoiceLinesForLog(lineResp.JSON201.Lines),
+		)
+
 		ctx := t.Context()
 		customers := api.InvoiceListParamsCustomers{customerID}
-		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		expand := api.InvoiceListParamsExpand{api.InvoiceExpandLines}
+		pollAttempt := 0
+		assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+			pollAttempt++
+
 			listResp, err := v1.ListInvoicesWithResponse(ctx, &api.ListInvoicesParams{
 				Customers: &customers,
+				Expand:    &expand,
 				PageSize:  lo.ToPtr(api.PaginationPageSize(100)),
 			})
-			require.NoError(c, err)
-			require.Equal(c, http.StatusOK, listResp.StatusCode(), "list invoices: %s", string(listResp.Body))
-			require.NotNil(c, listResp.JSON200)
+			require.NoError(collect, err)
+			require.Equal(collect, http.StatusOK, listResp.StatusCode(), "list invoices: %s", string(listResp.Body))
+			require.NotNil(collect, listResp.JSON200)
+
+			chargeStatus, charges, problem, chargeErr := c.ListCustomerChargesForDiagnostics(ctx, customerID, withPageSize(100))
+			t.Logf("standard invoice poll %02d: invoices=%s charges=%s",
+				pollAttempt,
+				formatInvoicesForLog(listResp.JSON200.Items),
+				formatChargesForLog(chargeStatus, charges, problem, chargeErr),
+			)
 
 			standardInvoiceIdx := slices.IndexFunc(listResp.JSON200.Items, func(inv api.Invoice) bool {
 				return inv.Status != api.InvoiceStatusGathering
 			})
-			require.NotEqual(c, -1, standardInvoiceIdx, "expected charges to advance a pending line into a standard invoice")
+			require.NotEqual(collect, -1, standardInvoiceIdx, "expected charges to advance a pending line into a standard invoice")
 
 			invoiceID = listResp.JSON200.Items[standardInvoiceIdx].Id
 		}, time.Minute, time.Second)
@@ -263,3 +283,44 @@ func TestV3GetBillingInvoice(t *testing.T) {
 		assert.NotNil(t, problem)
 	})
 }
+
+func formatInvoicesForLog(invoices []api.Invoice) string {
+	return formatLogJSON(invoices)
+}
+
+func formatInvoiceLinesForLog(lines []api.InvoiceLine) string {
+	return formatLogJSON(lines)
+}
+
+func formatChargesForLog(status int, charges *apiv3.ChargePagePaginatedResponse, problem *v3Problem, err error) string {
+	if err != nil {
+		return formatLogJSON(struct {
+			Status int    `json:"status"`
+			Error  string `json:"error"`
+		}{
+			Status: status,
+			Error:  err.Error(),
+		})
+	}
+
+	if status != http.StatusOK || charges == nil {
+		return formatLogJSON(struct {
+			Status  int        `json:"status"`
+			Problem *v3Problem `json:"problem,omitempty"`
+		}{
+			Status:  status,
+			Problem: problem,
+		})
+	}
+
+	return formatLogJSON(charges)
+}
+
+func formatLogJSON(v any) string {
+	out, err := json.Marshal(v)
+	if err != nil {
+		return `{"marshal_error":` + strconv.Quote(err.Error()) + `}`
+	}
+
+	return string(out)
+}

e2e/v3helpers_test.go

@@ -115,28 +115,45 @@ func (p *v3Problem) ValidationErrors() []v3ValidationError {
 func (c *v3Client) do(method, path string, body any) (int, []byte, *v3Problem) {
 	c.t.Helper()
 
+	ctx, cancel := context.WithTimeout(c.t.Context(), v3RequestTimeout)
+	defer cancel()
+
+	status, raw, problem, err := c.doRequest(ctx, method, path, body)
+	require.NoError(c.t, err, "%s %s", method, path)
+
+	return status, raw, problem
+}
+
+func (c *v3Client) doRequest(ctx context.Context, method, path string, body any) (int, []byte, *v3Problem, error) {
 	var bodyReader io.Reader
 	if body != nil {
 		b, err := json.Marshal(body)
-		require.NoError(c.t, err)
+		if err != nil {
+			return 0, nil, nil, fmt.Errorf("marshal request body: %w", err)
+		}
+
 		bodyReader = bytes.NewReader(b)
 	}
 
-	ctx, cancel := context.WithTimeout(c.t.Context(), v3RequestTimeout)
-	defer cancel()
-
 	req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, bodyReader)
-	require.NoError(c.t, err)
+	if err != nil {
+		return 0, nil, nil, fmt.Errorf("create request: %w", err)
+	}
+
 	if body != nil {
 		req.Header.Set("Content-Type", "application/json")
 	}
 
 	resp, err := http.DefaultClient.Do(req)
-	require.NoError(c.t, err, "%s %s", method, path)
+	if err != nil {
+		return 0, nil, nil, fmt.Errorf("do request: %w", err)
+	}
 	defer resp.Body.Close()
 
 	raw, err := io.ReadAll(resp.Body)
-	require.NoError(c.t, err)
+	if err != nil {
+		return resp.StatusCode, nil, nil, fmt.Errorf("read response body: %w", err)
+	}
 
 	var problem *v3Problem
 	if resp.StatusCode >= 400 && len(raw) > 0 {
@@ -146,7 +163,7 @@ func (c *v3Client) do(method, path string, body any) (int, []byte, *v3Problem) {
 		}
 	}
 
-	return resp.StatusCode, raw, problem
+	return resp.StatusCode, raw, problem, nil
 }
 
 // decodeTyped is a shared helper used by the typed wrappers below. It returns
@@ -303,6 +320,27 @@ func (c *v3Client) ListCustomerCharges(customerID string, opts ...listOption) (i
 	return decodeTyped[apiv3.ChargePagePaginatedResponse](c, status, raw, problem, http.StatusOK)
 }
 
+func (c *v3Client) ListCustomerChargesForDiagnostics(ctx context.Context, customerID string, opts ...listOption) (int, *apiv3.ChargePagePaginatedResponse, *v3Problem, error) {
+	ctx, cancel := context.WithTimeout(ctx, v3RequestTimeout)
+	defer cancel()
+
+	status, raw, problem, err := c.doRequest(ctx, http.MethodGet, "/customers/"+customerID+"/charges"+buildPageQuery(opts), nil)
+	if err != nil {
+		return status, nil, nil, err
+	}
+
+	if status != http.StatusOK {
+		return status, nil, problem, nil
+	}
+
+	var charges apiv3.ChargePagePaginatedResponse
+	if err := json.Unmarshal(raw, &charges); err != nil {
+		return status, nil, nil, fmt.Errorf("decode response: %w", err)
+	}
+
+	return status, &charges, nil, nil
+}
+
 func (c *v3Client) ListCustomerChargesByStatus(customerID string, statuses []apiv3.BillingChargeStatus, opts ...listOption) (int, *apiv3.ChargePagePaginatedResponse, *v3Problem) {
 	vals := buildPageValues(opts)
 	if len(statuses) > 0 {