Announce an acknowledged node failure on the bus
Docs / docs (push) Successful in 21s
Playwright Tests / test-playwright (1, 2) (push) Failing after 2m38s
Playwright Tests / test-playwright (2, 2) (push) Failing after 1m38s
pre-commit / pre-commit (push) Failing after 2m46s
Test Backend / test-backend (push) Successful in 2m17s
Compose Smoke Test / test-compose (push) Successful in 31s
Playwright Tests / merge-reports (push) Failing after 1m2s
Docs / docs (push) Successful in 21s
Playwright Tests / test-playwright (1, 2) (push) Failing after 2m38s
Playwright Tests / test-playwright (2, 2) (push) Failing after 1m38s
pre-commit / pre-commit (push) Failing after 2m46s
Test Backend / test-backend (push) Successful in 2m17s
Compose Smoke Test / test-compose (push) Successful in 31s
Playwright Tests / merge-reports (push) Failing after 1m2s
Clearing a node's last error on the engine published nothing, so a second browser kept the marker until its next snapshot. One event carries the qualified node; the receiving client drops the marker without refetching. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013Gf7WaExcJ9bs3kfJXB3nK
This commit is contained in:
@@ -779,6 +779,15 @@ def acknowledge_node_error(
|
|||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
|
||||||
) from None
|
) from None
|
||||||
|
# Another browser holds the same marker and has no reason to refetch, so it
|
||||||
|
# would keep showing a failure that is gone until its next snapshot.
|
||||||
|
event_bus.publish(
|
||||||
|
{
|
||||||
|
"type": "node_error_acknowledged",
|
||||||
|
"node": f"{name}.{node_id}",
|
||||||
|
"ts": time.time(),
|
||||||
|
}
|
||||||
|
)
|
||||||
return Message(message=f"Cleared the failure on '{node_id}'")
|
return Message(message=f"Cleared the failure on '{node_id}'")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,10 +1,12 @@
|
|||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
|
|
||||||
|
import pytest
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
from sqlalchemy import func
|
from sqlalchemy import func
|
||||||
from sqlmodel import Session, select
|
from sqlmodel import Session, select
|
||||||
|
|
||||||
from fluksio.core.config import settings
|
from fluksio.core.config import settings
|
||||||
|
from fluksio.flow.events import event_bus
|
||||||
from fluksio.models import Run, RunArtifact, RunMetric, RunNode
|
from fluksio.models import Run, RunArtifact, RunMetric, RunNode
|
||||||
|
|
||||||
PREFIX = f"{settings.API_V1_STR}/flows"
|
PREFIX = f"{settings.API_V1_STR}/flows"
|
||||||
@@ -422,6 +424,34 @@ def test_a_node_that_raises_answers_with_its_error(
|
|||||||
client.delete(f"{PREFIX}/failing", headers=superuser_token_headers)
|
client.delete(f"{PREFIX}/failing", headers=superuser_token_headers)
|
||||||
|
|
||||||
|
|
||||||
|
def test_acknowledging_a_failure_is_announced_on_the_bus(
|
||||||
|
client: TestClient,
|
||||||
|
superuser_token_headers: dict[str, str],
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Another browser holds the same marker; the event is what clears it."""
|
||||||
|
saved = client.put(
|
||||||
|
f"{PREFIX}/acked", headers=superuser_token_headers, json=a_flow("acked")
|
||||||
|
).json()
|
||||||
|
client.post(
|
||||||
|
f"{PREFIX}/acked/publish",
|
||||||
|
headers=superuser_token_headers,
|
||||||
|
json={"version": saved["definition"]["version"]},
|
||||||
|
)
|
||||||
|
|
||||||
|
published: list[dict] = []
|
||||||
|
monkeypatch.setattr(event_bus, "publish", published.append)
|
||||||
|
response = client.post(
|
||||||
|
f"{PREFIX}/acked/nodes/sensor/acknowledge", headers=superuser_token_headers
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
event = next(e for e in published if e["type"] == "node_error_acknowledged")
|
||||||
|
assert event["node"] == "acked.sensor"
|
||||||
|
|
||||||
|
client.delete(f"{PREFIX}/acked", headers=superuser_token_headers)
|
||||||
|
|
||||||
|
|
||||||
def test_renaming_a_flow_leaves_nothing_under_the_old_name(
|
def test_renaming_a_flow_leaves_nothing_under_the_old_name(
|
||||||
client: TestClient, superuser_token_headers: dict[str, str]
|
client: TestClient, superuser_token_headers: dict[str, str]
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|||||||
@@ -167,6 +167,10 @@ export const liveStore = {
|
|||||||
getFailure(nodeId: string) {
|
getFailure(nodeId: string) {
|
||||||
return failures.get(nodeId)
|
return failures.get(nodeId)
|
||||||
},
|
},
|
||||||
|
/** Drop the marker for a failure the engine has already been told about. */
|
||||||
|
clearFailure(nodeId: string) {
|
||||||
|
if (failures.delete(nodeId)) notify(`failure:${nodeId}`)
|
||||||
|
},
|
||||||
/**
|
/**
|
||||||
* Dismiss a node's failure, here and on the engine.
|
* Dismiss a node's failure, here and on the engine.
|
||||||
*
|
*
|
||||||
|
|||||||
@@ -57,6 +57,7 @@ type FlowEvent =
|
|||||||
ts?: number
|
ts?: number
|
||||||
}
|
}
|
||||||
| { type: "node_status"; node: string; status: string; error?: string | null }
|
| { type: "node_status"; node: string; status: string; error?: string | null }
|
||||||
|
| { type: "node_error_acknowledged"; node: string; ts?: number }
|
||||||
| ({ type: "node_log" } & LogLine)
|
| ({ type: "node_log" } & LogLine)
|
||||||
| { type: "flow_paused"; flow: string; paused: boolean }
|
| { type: "flow_paused"; flow: string; paused: boolean }
|
||||||
| {
|
| {
|
||||||
@@ -235,6 +236,11 @@ function connect() {
|
|||||||
error: message.error,
|
error: message.error,
|
||||||
})
|
})
|
||||||
break
|
break
|
||||||
|
case "node_error_acknowledged":
|
||||||
|
// Someone dismissed it, here or in another browser. The engine has
|
||||||
|
// already forgotten it, so drop the marker rather than posting back.
|
||||||
|
liveStore.clearFailure(message.node)
|
||||||
|
break
|
||||||
case "node_log":
|
case "node_log":
|
||||||
liveStore.appendLog(message)
|
liveStore.appendLog(message)
|
||||||
break
|
break
|
||||||
|
|||||||
Reference in New Issue
Block a user