{{ explanation }}
{% if not privileged %}Somente leitura para o seu perfil: alterar para aprovação/automático, aprovar, aplicar e reverter exige administrador do GTI Gerenciador.
{% endif %} {% set rtc = ops.rtc %} {% if rtc and rtc.products %} @@ -18,7 +18,7 @@Decidido centralmente para esta instalação. Padrão: Desativada. Nenhuma configuração local no servidor do cliente.
Página {{ page }} de {{ pages }}{% if page > 1 %} · anterior{% endif %}{% if page < pages %} · próxima{% endif %}
-+
{% if mode == "APPROVAL_REQUIRED" %}{% endif %}
diff --git a/central/tests/test_heavy_job_backpressure.py b/central/tests/test_heavy_job_backpressure.py new file mode 100644 index 0000000..6383f14 --- /dev/null +++ b/central/tests/test_heavy_job_backpressure.py @@ -0,0 +1,67 @@ +"""Backpressure: at most HEAVY_JOB_CONCURRENCY_LIMIT whole-database jobs run at once across all Agents; the rest queue as PENDING.""" +from types import SimpleNamespace + +from sqlalchemy import select + +from central.app import jobs +from central.app.models import Agent, Job, Tenant +from central.tests.test_control_plane import setup_db + + +def _agents(session, n): + tenant = Tenant(name="Backpressure") + session.add(tenant) + session.flush() + out = [] + for i in range(n): + agent = Agent(tenant_id=tenant.id, name=f"a{i}", token_hash=f"t{i}".ljust(64, "x"), agent_version="1.2.1") + session.add(agent) + out.append(agent) + session.flush() + return out + + +def _limit(monkeypatch, value): + monkeypatch.setattr(jobs, "settings", SimpleNamespace(heavy_job_concurrency_limit=value)) + + +def test_only_the_configured_number_of_full_jobs_run_at_once_and_the_rest_wait(monkeypatch): + _limit(monkeypatch, 2) + with setup_db()() as session: + agents = _agents(session, 4) + queued = [jobs.create_job(session, a, "INITIAL_SYNC") for a in agents] + claimed = [jobs.claim_next_job(session, a) for a in agents] + assert [c is not None for c in claimed] == [True, True, False, False] # 2 running, 2 queued (backpressure) + assert [session.get(Job, q.id).status for q in queued] == ["RUNNING", "RUNNING", "PENDING", "PENDING"] + jobs.complete_job(session, session.get(Job, queued[0].id), {}) # a slot frees up + assert jobs.claim_next_job(session, agents[2]) is not None and jobs.claim_next_job(session, agents[3]) is None + + +def test_a_light_incremental_job_is_never_held_behind_a_queued_full_job(monkeypatch): + _limit(monkeypatch, 1) + with setup_db()() as session: + busy, waiting = _agents(session, 2) + jobs.create_job(session, busy, "REPROCESS_ALL") + assert jobs.claim_next_job(session, busy) is not None + full = jobs.create_job(session, waiting, "REPROCESS_ALL") + light = jobs.create_job(session, waiting, "REPROCESS_PRODUCTS", {"codigos": ["1"]}) + claimed = jobs.claim_next_job(session, waiting) + assert claimed is not None and claimed.id == light.id and session.get(Job, full.id).status == "PENDING" + + +def test_limit_zero_disables_backpressure_and_a_stale_running_job_does_not_hold_a_slot(monkeypatch): + from datetime import datetime, timedelta, timezone + _limit(monkeypatch, 0) + with setup_db()() as session: + agents = _agents(session, 3) + for a in agents: + jobs.create_job(session, a, "INITIAL_SYNC") + assert all(jobs.claim_next_job(session, a) is not None for a in agents) + _limit(monkeypatch, 1) + with setup_db()() as session: + dead, fresh = _agents(session, 2) + stale = jobs.create_job(session, dead, "INITIAL_SYNC") + session.execute(Job.__table__.update().where(Job.id == stale.id).values(status="RUNNING", started_at=datetime.now(timezone.utc) - timedelta(hours=3))) + session.commit() + jobs.create_job(session, fresh, "INITIAL_SYNC") + assert jobs.claim_next_job(session, fresh) is not None # the dead Agent's 3 h old RUNNING job does not block the queue diff --git a/central/tests/test_rtc_security.py b/central/tests/test_rtc_security.py index c231815..ea57919 100644 --- a/central/tests/test_rtc_security.py +++ b/central/tests/test_rtc_security.py @@ -97,3 +97,54 @@ def test_agent_can_reread_only_its_own_current_authorization_and_it_follows_the_ s.commit() body = client.get("/v1/agents/me/authorizations", headers=H).json() assert body["rtc_write_mode"] == "ENABLED" and set(body) == {"rtc_write_mode", "rtc_approved_codes", "rtc_rollbacks"} + + +def _become(client, role): + """Turn the logged-in test operator into a Gerenciador SSO collaborator with the given perfil (identity source of truth is the SSO).""" + from central.app.models import AdminUser + with client._session_factory() as s: + admin = s.scalars(select(AdminUser)).first() + admin.auth_source, admin.password_hash, admin.external_subject, admin.external_role = "GTI_GERENCIADOR", None, "sso-subject-1", role + s.commit() + + +def _post(client, path, **data): + return client.post(path, data={"csrf_token": client.cookies.get("gti_csrf"), **data}, follow_redirects=False) + + +def test_write_modes_are_restricted_to_gerenciador_admins_and_every_denial_is_audited(): + client, agent_id, _ = _panel_client() + _become(client, "PADRAO") + for mode, extra in (("ENABLED", {"confirm": "PREENCHER"}), ("APPROVAL_REQUIRED", {})): + assert _post(client, f"/agents/{agent_id}/rtc-mode", mode=mode, **extra).status_code == 403 + assert _post(client, f"/agents/{agent_id}/rtc-candidates", action="approve", all="1").status_code == 403 + assert _post(client, f"/agents/{agent_id}/rtc-candidates", action="apply_approved").status_code == 403 + assert _post(client, f"/agents/{agent_id}/rtc-rollback", audit_id="1", confirm="REVERTER").status_code == 403 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().rtc_write_mode == "DISABLED" + assert len([e for e in s.scalars(select(AuditEvent)) if e.action == "RTC_DENIED"]) == 5 + # read-only modes stay available to a normal collaborator, and the page tells them why the rest is disabled + assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="DRY_RUN").status_code == 303 + page = client.get(f"/agents/{agent_id}/rtc").text + assert "Somente leitura para o seu perfil" in page and "disabled" in page + + +def test_a_normal_collaborator_cannot_switch_off_a_privileged_mode_either(): + client, agent_id, _ = _panel_client() + with client._session_factory() as s: + s.scalars(select(Installation)).one().rtc_write_mode = "ENABLED" + s.commit() + _become(client, "PADRAO") + assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="DISABLED").status_code == 403 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().rtc_write_mode == "ENABLED" + + +def test_gerenciador_admin_can_change_modes_and_the_operator_is_recorded(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="ENABLED", confirm="PREENCHER").status_code == 303 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().rtc_write_mode == "ENABLED" + event = [e for e in s.scalars(select(AuditEvent)) if e.action == "RTC_WRITE_MODE"][-1] + assert event.context["to"] == "ENABLED" and event.context["role"] == "ADMIN" and event.context["operator"] diff --git a/deploy/hub/.env.example b/deploy/hub/.env.example index 7b45d0e..d9c2a71 100644 --- a/deploy/hub/.env.example +++ b/deploy/hub/.env.example @@ -43,3 +43,5 @@ FISCAL_INTERNAL_API_TOKEN= # Directory (host) where pg_dump backups are written by backup.sh BACKUP_DIR=./backups +# Max whole-database jobs (INITIAL_SYNC / REPROCESS_ALL) running at once across all Agents; others queue as PENDING. <=0 disables. +HEAVY_JOB_CONCURRENCY_LIMIT=3 diff --git a/docs/rtc_auto_apply.md b/docs/rtc_auto_apply.md index 14947df..2aa0f13 100644 --- a/docs/rtc_auto_apply.md +++ b/docs/rtc_auto_apply.md @@ -21,6 +21,12 @@ Sequência por produto (uma transação curta): reler tudo → recalcular a deci ## Reautorização imediata antes da escrita (1.2.1) O modo entregue no job pode envelhecer (operador desativa durante um lote). Antes de **cada** UPDATE o Agent relê a autorização atual no Hub (`GET /v1/agents/me/authorizations`, autenticado pelo token do próprio Agent; devolve só modo, aprovados e reversões da sua instalação; cache de 3 s). Se o modo mudou, se o produto perdeu a aprovação ou se o Hub não puder ser consultado, o produto é interrompido **sem escrever** (`AUTHORIZATION_CHANGED` / `AUTHORIZATION_UNVERIFIABLE`). Autorização não confirmada não é autorização. +## Quem pode mudar o que (RBAC, 1.2.1) +A identidade vem do SSO do GTI Gerenciador (`perfil`). **ADMIN** (e contas LOCAL de emergência) pode: `APPROVAL_REQUIRED`/`ENABLED`, aprovar/rejeitar, aplicar aprovados e reverter. Um colaborador **PADRAO** só vê a página e alterna `DISABLED`/`DRY_RUN` (somente leitura); tentativas restritas devolvem 403 e ficam auditadas (`RTC_DENIED`). Sair de um modo de escrita também é restrito a ADMIN. Toda mudança de modo, aprovação e reversão grava o operador e o perfil no evento de auditoria. + +## Backpressure de jobs pesados (1.2.1) +`HEAVY_JOB_CONCURRENCY_LIMIT` (padrão 3; `<=0` desliga): no máximo N jobs `INITIAL_SYNC`/`REPROCESS_ALL` rodam ao mesmo tempo em **todos** os Agents; os demais ficam `PENDING` (fila) e o Agent segue com heartbeat e polling. Um job leve nunca fica preso atrás de um pesado. Job RUNNING há mais de 1 h não ocupa vaga. Medição (8 Agents × 1.042 produtos, Hub de desenvolvimento): sem limite 51,9 produtos/s, CPU pico 149%, cada Agent até 160 s; com limite 3: 66,7 produtos/s, CPU pico 116%, cada Agent até 45 s (espera máx. 88 s), 0 falhas. + ## Preview e aprovação O Agent envia os candidatos (até 5.000 por job; `preview_truncated` sinaliza excesso) → tabela `rtc_candidates` (PENDING/APPROVED/REJECTED/APPLIED/STALE). O painel pagina no servidor (50/página), filtra por status, aprova/rejeita selecionados ou todos os pendentes e dispara "Aplicar aprovados agora". Uma proposta que muda invalida a aprovação anterior; produto que deixa de ser elegível numa avaliação completa vira `STALE`. diff --git a/docs/runs/final_corpus/concurrency_sim.json b/docs/runs/final_corpus/concurrency_sim.json new file mode 100644 index 0000000..5b4b759 --- /dev/null +++ b/docs/runs/final_corpus/concurrency_sim.json @@ -0,0 +1,28 @@ +{ + "agents": 8, + "products_per_agent": 1042, + "runs": [ + { + "limit": "none", + "wall_s": 160.6, + "products_per_s": 51.9, + "failures": 0, + "max_run_s": 160.6, + "max_queue_wait_s": 0.0, + "hub_cpu_peak_pct": 149.18, + "hub_cpu_avg_pct": 130.0, + "hub_mem_peak_mib": 89.19 + }, + { + "limit": 3, + "wall_s": 124.9, + "products_per_s": 66.7, + "failures": 0, + "max_run_s": 45.5, + "max_queue_wait_s": 88.4, + "hub_cpu_peak_pct": 115.81, + "hub_cpu_avg_pct": 71.3, + "hub_mem_peak_mib": 79.25 + } + ] +} \ No newline at end of file diff --git a/scripts/concurrency_sim.py b/scripts/concurrency_sim.py new file mode 100644 index 0000000..58fb0e5 --- /dev/null +++ b/scripts/concurrency_sim.py @@ -0,0 +1,73 @@ +"""Controlled concurrency simulation of whole-database classification against the LOCAL dev Hub. +N simulated Agents each classify M products (read-only Firebird sample + Hub /classify, exactly the Agent's HubClient path). +UNLIMITED = all N at once; LIMITED = only K at a time (what HEAVY_JOB_CONCURRENCY_LIMIT enforces at job claim). Samples the Hub container CPU/RAM (docker stats). + python scripts/concurrency_sim.py [--agents 8] [--products 2084] [--limit 3]""" +import argparse +import json +import os +import subprocess +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path[:0] = [str(ROOT), str(ROOT / "scripts"), str(ROOT / "agent" / "src")] +from dotenv import load_dotenv # noqa: E402 + +load_dotenv(ROOT / ".env") +import local_completion as lc # noqa: E402 +from gti_fiscal_agent import firebird_reader # noqa: E402 +from gti_fiscal_agent.client import HubClient # noqa: E402 +from gti_fiscal_agent.runner import to_hub_product # noqa: E402 + + +def sampler(stop, samples): + while not stop.is_set(): + out = subprocess.run(["docker", "stats", "--no-stream", "--format", "{{.CPUPerc}}|{{.MemUsage}}", "projeto-fiscal-hub-1"], capture_output=True, text=True).stdout.strip() + try: + cpu, mem = out.split("|"); samples.append((float(cpu.strip("%")), float(mem.split("/")[0].strip().replace("MiB", "").replace("GiB", "")) )) + except Exception: # noqa: BLE001 + pass + time.sleep(1) + + +def run(agents, products, limit): + gate = threading.Semaphore(limit if limit else agents) + results, samples, stop = [], [], threading.Event() + threading.Thread(target=sampler, args=(stop, samples), daemon=True).start() + + def one(i): + queued = time.time() + with gate: + started = time.time() + try: + client = HubClient(os.environ.get("HUB_URL", lc.HUB), os.environ["AGENT_DEV_TOKEN"]) + client.classify_batch(products) + return {"agent": i, "queue_wait_s": round(started - queued, 1), "run_s": round(time.time() - started, 1), "ok": True} + except Exception as error: # noqa: BLE001 + return {"agent": i, "queue_wait_s": round(started - queued, 1), "run_s": round(time.time() - started, 1), "ok": False, "error": type(error).__name__} + t0 = time.time() + with ThreadPoolExecutor(agents) as pool: + results = list(pool.map(one, range(agents))) + wall = time.time() - t0 + stop.set() + cpus = [c for c, _ in samples] or [0]; mems = [m for _, m in samples] or [0] + return {"limit": limit or "none", "wall_s": round(wall, 1), "products_per_s": round(agents * len(products) / wall, 1), "failures": sum(1 for r in results if not r["ok"]), + "max_run_s": max(r["run_s"] for r in results), "max_queue_wait_s": max(r["queue_wait_s"] for r in results), "hub_cpu_peak_pct": max(cpus), "hub_cpu_avg_pct": round(sum(cpus) / len(cpus), 1), "hub_mem_peak_mib": max(mems)} + + +def main(): + ap = argparse.ArgumentParser(); ap.add_argument("--agents", type=int, default=8); ap.add_argument("--products", type=int, default=2084); ap.add_argument("--limit", type=int, default=3) + a = ap.parse_args() + env = lc.env_for(lc.allowlist()["TESTESOFICIAIS"]) + base = firebird_reader.read_products(env) + sample = [to_hub_product(p) for p in (base * (a.products // len(base) + 1))[:a.products]] + out = {"agents": a.agents, "products_per_agent": a.products, "runs": [run(a.agents, sample, 0), run(a.agents, sample, a.limit)]} + print(json.dumps(out, indent=1)) + (ROOT / "docs" / "runs" / "final_corpus" / "concurrency_sim.json").write_text(json.dumps(out, indent=1), encoding="utf-8") + + +if __name__ == "__main__": + main()