From 1219589f92508bcc7c7a1a6f7035a10cf4b47326 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Thu, 8 Oct 2026 14:47:54 -0400 Subject: [PATCH] fix(OMN-20682): grant the PR watcher its CI-bus topics and consumer group Since 18:28Z the operator Mac PR watcher (launchd ai.omninode.pr-watcher, `handler_pr_watcher_cli serve --bus-lane ci-bus`) crash-loops with InfraConnectionError ... Failed to start consumer for topic onex.cmd.omnibase_internal.pr-watcher-tick-requested.v1: [Error 29] TopicAuthorizationFailedError. It logs in to the ci-bus as the scoped lab-work principal, whose ACLs come from the ci-bus-labwork-user one-shot in docker/docker-compose.ci-bus.yml. Source: omnibase_internal node_pr_watcher_effect contract.yaml (omnibase_internal main b893625b). Subscribes onex.cmd.omnibase_internal.pr-watcher-tick-requested.v1; publishes onex.evt.omnibase_internal.pr-watcher-tick-completed.v1, onex.evt.omnibase_internal.pr-watcher-tick-failed.v1, onex.evt.omnimarket.pr-state-observed.v1 and onex.evt.omnimarket.repo-merged.v1 (already granted). runtime_dispatch.group_id: onex.omnibase_internal.node_pr_watcher_effect. The serve path passes that literal group_id to bus.subscribe; it does not call derive_service_group_id, so no env token prefix applies. EventBusKafka._resolve_effective_group_id appends .__i. when set and always .__t.. Change 1: ci-bus-topics one-shot LAB_TOPICS gains the four topics (it created none of them before). Readback echo changes from "sixteen" to "twenty-three" topics (5 CI + 18 lab; the old "sixteen" was already wrong at 19). Change 2: ci-bus-labwork-user LAB_TOPICS gains the same four topics (read, write, describe, literal). GROUP_PREFIXES gains the prefixed group grant onex.omnibase_internal.node_pr_watcher_effect. (trailing dot). ACL readback echo changes from "fourteen" to "eighteen" lab topics. Change 3: tests/unit/docker/test_ci_bus_broker_isolation_omn18691.py pins the four topics, len(granted) == 18, widens the topic regex to admit underscores (the old [a-z0-9.-] class did not match omnibase_internal names), and adds test_ci_bus_labwork_grants_the_pr_watcher_group_and_counts_match (group prefix present; both readback echoes state the real counts). Tests: 13 passed in that file; 86 passed in four neighbouring compose tests; pre-commit clean on both files. Applying the grants on the .201 ci-bus is a separate step, not done by this commit. --- docker/docker-compose.ci-bus.yml | 16 ++++-- .../test_ci_bus_broker_isolation_omn18691.py | 50 ++++++++++++++++--- 2 files changed, 54 insertions(+), 12 deletions(-) diff --git a/docker/docker-compose.ci-bus.yml b/docker/docker-compose.ci-bus.yml index 9b838ad018..8af30278fd 100644 --- a/docker/docker-compose.ci-bus.yml +++ b/docker/docker-compose.ci-bus.yml @@ -228,6 +228,9 @@ services: # serve` per host in a group .), on the same login. # OMN-20604 adds the lab job supervisor's submitted command topic # (omnimarket node_lab_job_submit_effect), on the same login, for fourteen total. + # OMN-20682 adds the operator-host PR watcher's four topics (omnibase_internal + # node_pr_watcher_effect: its tick command, tick completed and failed, and + # pr-state-observed), on the same login, for eighteen lab topics in total. ci-bus-topics: image: redpandadata/redpanda:v24.2.7 container_name: omninode-ci-bus-topics @@ -251,7 +254,7 @@ services: # addition does not hit the Redpanda default of 1000 (OMN-7302). /usr/bin/rpk cluster config set topic_partitions_per_shard 7000 $$ADMIN CI_TOPICS="onex.cmd.omnimarket.occ-autobind.v1 onex.cmd.omnimarket.occ-companion-effect-requested.v1 onex.evt.github.pr-merged.v1 onex.cmd.omnimarket.redeploy-start.v1 onex.evt.deploy.rebuild-completed.v1" - LAB_TOPICS="onex.cmd.omnimarket.lab-work-unit-requested.v1 onex.evt.omnimarket.lab-work-unit-completed.v1 onex.evt.omnimarket.lab-work-unit-failed.v1 onex.evt.omnimarket.lab-host-capacity-advertised.v1 onex.cmd.omnimarket.focused-test-run-requested.v1 onex.evt.omnimarket.focused-test-run-completed.v1 onex.evt.omnimarket.focused-test-run-failed.v1 onex.cmd.omnimarket.push-validation-requested.v1 onex.evt.omnimarket.push-validation-completed.v1 onex.evt.omnimarket.push-validation-failed.v1 onex.dlq.omnimarket.push-validation.v1 onex.evt.omnimarket.repo-merged.v1 onex.evt.omnimarket.canonical-clone-refreshed.v1 onex.cmd.omnimarket.lab-job-submitted.v1" + LAB_TOPICS="onex.cmd.omnimarket.lab-work-unit-requested.v1 onex.evt.omnimarket.lab-work-unit-completed.v1 onex.evt.omnimarket.lab-work-unit-failed.v1 onex.evt.omnimarket.lab-host-capacity-advertised.v1 onex.cmd.omnimarket.focused-test-run-requested.v1 onex.evt.omnimarket.focused-test-run-completed.v1 onex.evt.omnimarket.focused-test-run-failed.v1 onex.cmd.omnimarket.push-validation-requested.v1 onex.evt.omnimarket.push-validation-completed.v1 onex.evt.omnimarket.push-validation-failed.v1 onex.dlq.omnimarket.push-validation.v1 onex.evt.omnimarket.repo-merged.v1 onex.evt.omnimarket.canonical-clone-refreshed.v1 onex.cmd.omnimarket.lab-job-submitted.v1 onex.cmd.omnibase_internal.pr-watcher-tick-requested.v1 onex.evt.omnibase_internal.pr-watcher-tick-completed.v1 onex.evt.omnibase_internal.pr-watcher-tick-failed.v1 onex.evt.omnimarket.pr-state-observed.v1" TOPICS="$$CI_TOPICS $$LAB_TOPICS" for t in $$TOPICS; do if /usr/bin/rpk topic list $$BROKER | awk '{print $$1}' | grep -qx "$$t"; then @@ -270,7 +273,7 @@ services: for t in $$TOPICS; do /usr/bin/rpk topic list $$BROKER | awk '{print $$1}' | grep -qx "$$t" done - echo "OMN-18691 + OMN-20213: readback confirms all sixteen topics exist" + echo "OMN-18691 + OMN-20213: readback confirms all twenty-three topics exist" depends_on: redpanda: condition: service_healthy @@ -469,7 +472,7 @@ services: fi echo "OMN-20213: lab-work principal is not a superuser" # Kafka ACL creation is idempotent: an identical binding is a no-op. - LAB_TOPICS="onex.cmd.omnimarket.lab-work-unit-requested.v1 onex.evt.omnimarket.lab-work-unit-completed.v1 onex.evt.omnimarket.lab-work-unit-failed.v1 onex.evt.omnimarket.lab-host-capacity-advertised.v1 onex.cmd.omnimarket.focused-test-run-requested.v1 onex.evt.omnimarket.focused-test-run-completed.v1 onex.evt.omnimarket.focused-test-run-failed.v1 onex.cmd.omnimarket.push-validation-requested.v1 onex.evt.omnimarket.push-validation-completed.v1 onex.evt.omnimarket.push-validation-failed.v1 onex.dlq.omnimarket.push-validation.v1 onex.evt.omnimarket.repo-merged.v1 onex.evt.omnimarket.canonical-clone-refreshed.v1 onex.cmd.omnimarket.lab-job-submitted.v1" + LAB_TOPICS="onex.cmd.omnimarket.lab-work-unit-requested.v1 onex.evt.omnimarket.lab-work-unit-completed.v1 onex.evt.omnimarket.lab-work-unit-failed.v1 onex.evt.omnimarket.lab-host-capacity-advertised.v1 onex.cmd.omnimarket.focused-test-run-requested.v1 onex.evt.omnimarket.focused-test-run-completed.v1 onex.evt.omnimarket.focused-test-run-failed.v1 onex.cmd.omnimarket.push-validation-requested.v1 onex.evt.omnimarket.push-validation-completed.v1 onex.evt.omnimarket.push-validation-failed.v1 onex.dlq.omnimarket.push-validation.v1 onex.evt.omnimarket.repo-merged.v1 onex.evt.omnimarket.canonical-clone-refreshed.v1 onex.cmd.omnimarket.lab-job-submitted.v1 onex.cmd.omnibase_internal.pr-watcher-tick-requested.v1 onex.evt.omnibase_internal.pr-watcher-tick-completed.v1 onex.evt.omnibase_internal.pr-watcher-tick-failed.v1 onex.evt.omnimarket.pr-state-observed.v1" for t in $$LAB_TOPICS; do su_rpk security acl create --allow-principal "User:$$CI_BUS_LABWORK_SASL_USERNAME" \ --resource-pattern-type literal --topic "$$t" \ @@ -480,7 +483,10 @@ services: # random suffix, and EventBusKafka appends .__t.. These must be # prefixed grants rather than literal group names. # The host drift sync hub reads repo-merged in its own group prefix (OMN-20689). - GROUP_PREFIXES="local.omnimarket.node_lab_work_unit_effect.consume.v1. local.omnimarket.lab_work_client.consume.v1. local.omnimarket.node_focused_test_run_effect.consume.v1 local.omnimarket.delegated_test_loop_client.consume.v1. local.omnimarket.node_canonical_clone_refresh_effect.consume.v1. local.omnibase_internal.host_drift_sync.hub." + # The PR watcher (omnibase_internal node_pr_watcher_effect) subscribes with its + # contract's literal group_id; EventBusKafka appends .__t. (and .__i. + # when KAFKA_INSTANCE_ID is set), so the prefix ends in a dot (OMN-20682). + GROUP_PREFIXES="local.omnimarket.node_lab_work_unit_effect.consume.v1. local.omnimarket.lab_work_client.consume.v1. local.omnimarket.node_focused_test_run_effect.consume.v1 local.omnimarket.delegated_test_loop_client.consume.v1. local.omnimarket.node_canonical_clone_refresh_effect.consume.v1. local.omnibase_internal.host_drift_sync.hub. onex.omnibase_internal.node_pr_watcher_effect." for g in $$GROUP_PREFIXES; do su_rpk security acl create --allow-principal "User:$$CI_BUS_LABWORK_SASL_USERNAME" \ --resource-pattern-type prefixed --group "$$g" \ @@ -495,7 +501,7 @@ services: for t in $$LAB_TOPICS; do printf '%s\n' "$$ACLS" | grep -q "$$t" done - echo "OMN-20213 + OMN-20496: ACL readback confirms all fourteen lab topics" + echo "OMN-20213 + OMN-20496 + OMN-20682: ACL readback confirms all eighteen lab topics" # A down broker also hides topics, so the negative alone proves nothing. # Capture once as the lab-work login: the positive proves authentication # and authorization, and the negative proves a CI topic is invisible. diff --git a/tests/unit/docker/test_ci_bus_broker_isolation_omn18691.py b/tests/unit/docker/test_ci_bus_broker_isolation_omn18691.py index 3b86a0bf9c..8ae70f0a87 100644 --- a/tests/unit/docker/test_ci_bus_broker_isolation_omn18691.py +++ b/tests/unit/docker/test_ci_bus_broker_isolation_omn18691.py @@ -395,7 +395,8 @@ def test_ci_bus_topics_are_exactly_the_ci_bus_set(ci_bus: dict[str, Any]) -> Non lab-work units, host capacity, focused test runs and push validation (OMN-20213), and two more carry merge events and each host's canonical-clone refresh receipt (OMN-20496). OMN-20604 adds the lab job supervisor's submitted command topic, - for fourteen lab topics in total. + and OMN-20682 adds the operator-host PR watcher's four topics, for eighteen + lab topics in total. """ expected = { "onex.cmd.omnimarket.occ-autobind.v1", # onex-topic-allow: OMN-18691 CI-bus provisioning set @@ -417,10 +418,14 @@ def test_ci_bus_topics_are_exactly_the_ci_bus_set(ci_bus: dict[str, Any]) -> Non "onex.evt.omnimarket.repo-merged.v1", # onex-topic-allow: OMN-20496 "onex.evt.omnimarket.canonical-clone-refreshed.v1", # onex-topic-allow: OMN-20496 "onex.cmd.omnimarket.lab-job-submitted.v1", # onex-topic-allow: OMN-20604 + "onex.cmd.omnibase_internal.pr-watcher-tick-requested.v1", # onex-topic-allow: OMN-20682 + "onex.evt.omnibase_internal.pr-watcher-tick-completed.v1", # onex-topic-allow: OMN-20682 + "onex.evt.omnibase_internal.pr-watcher-tick-failed.v1", # onex-topic-allow: OMN-20682 + "onex.evt.omnimarket.pr-state-observed.v1", # onex-topic-allow: OMN-20682 } raw = CI_BUS_COMPOSE.read_text(encoding="utf-8") - found = set(re.findall(r"onex\.[a-z]+\.[a-z0-9.-]+\.v\d+", raw)) + found = set(re.findall(r"onex\.[a-z]+\.[a-z0-9._-]+\.v\d+", raw)) assert found == expected, ( "the CI-bus bring-up one-shot provisions a different topic set than this " f"test pins: compose-only={sorted(found - expected)}, " @@ -431,8 +436,9 @@ def test_ci_bus_topics_are_exactly_the_ci_bus_set(ci_bus: dict[str, Any]) -> Non ) # Cross-check the one name this repo owns a canonical constant for. The other - # eighteen are declared in omnimarket's topic registry, which this repo does not - # import -- stated rather than left as an apparent omission. + # twenty-two are declared in omnimarket's and omnibase_internal's topic + # registries, which this repo does not import -- stated rather than left as + # an apparent omission. from omnibase_infra.topics.platform_topic_suffixes import SUFFIX_GITHUB_PR_MERGED assert SUFFIX_GITHUB_PR_MERGED in expected, ( @@ -472,16 +478,46 @@ def test_ci_bus_labwork_principal_is_scoped_not_superuser( lab_topics = re.search(r'LAB_TOPICS="([^"]+)"', command) assert lab_topics is not None, "the lab-work ACL grant set must be explicit" granted = lab_topics.group(1).split() - assert len(granted) == 14 + assert len(granted) == 18 assert '--resource-pattern-type literal --topic "$$t"' in command topics_command = "\n".join(services["ci-bus-topics"]["command"]) - declared = set(re.findall(r"onex\.[a-z]+\.[a-z0-9.-]+\.v\d+", topics_command)) + declared = set(re.findall(r"onex\.[a-z]+\.[a-z0-9._-]+\.v\d+", topics_command)) assert set(granted) <= declared, "every lab-work grant must name a declared topic" # Include any separately added literal grant, not just the LAB_TOPICS loop. - referenced = set(re.findall(r"onex\.[a-z]+\.[a-z0-9.-]+\.v\d+", command)) + referenced = set(re.findall(r"onex\.[a-z]+\.[a-z0-9._-]+\.v\d+", command)) assert referenced <= declared, "lab-work ACLs and controls must use declared topics" +def test_ci_bus_labwork_grants_the_pr_watcher_group_and_counts_match( + ci_bus: dict[str, Any], +) -> None: + """OMN-20682: the PR watcher serve process subscribes with the contract group_id + onex.omnibase_internal.node_pr_watcher_effect, which EventBusKafka suffixes + with .__t., so the lab-work login needs that prefix; the readback echoes + must state the real counts. + """ + services = ci_bus.get("services") or {} + command = "\n".join(services["ci-bus-labwork-user"]["command"]) + topics_command = "\n".join(services["ci-bus-topics"]["command"]) + prefixes = re.search(r'GROUP_PREFIXES="([^"]+)"', command) + assert prefixes is not None + assert "onex.omnibase_internal.node_pr_watcher_effect." in prefixes.group(1).split() + + words = {18: "eighteen", 23: "twenty-three"} + lab_topics = re.search(r'LAB_TOPICS="([^"]+)"', command) + assert lab_topics is not None + granted = lab_topics.group(1).split() + assert f"ACL readback confirms all {words[len(granted)]} lab topics" in command + + ci_topics = re.search(r'CI_TOPICS="([^"]+)"', topics_command) + lab = re.search(r'LAB_TOPICS="([^"]+)"', topics_command) + assert ci_topics is not None + assert lab is not None + total = len(ci_topics.group(1).split()) + len(lab.group(1).split()) + assert total == 23 + assert f"readback confirms all {words[total]} topics exist" in topics_command + + def test_ci_bus_topics_one_shot_authenticates_once_sasl_is_on( ci_bus: dict[str, Any], ) -> None: