From 16ebd810a56af0fdf97df2ee73f1a37e8646eef1 Mon Sep 17 00:00:00 2001 From: SimpleTest Date: Mon, 20 Jul 2026 06:38:32 +0300 Subject: [PATCH] fix: harden runtime and reduce request overhead --- .dialyzer_ignore.exs | 1 - .env.e2e.example | 7 +- .env.example | 10 +- .env.load.example | 9 +- Dockerfile.postgis | 9 + Dockerfile.socket-proxy | 11 + compose.cpu-replay.yaml | 14 ++ compose.load.yaml | 2 +- compose.quality.yaml | 2 +- compose.yaml | 99 ++++++++- config/runtime.exs | 6 +- .../who-need-help/templates/deployments.yaml | 4 +- .../who-need-help/templates/migrate-job.yaml | 2 + .../who-need-help/templates/services.yaml | 17 ++ deploy/helm/who-need-help/values.yaml | 4 +- docs/performance.md | 46 ++++ lib/who_need_help/accounts.ex | 67 ++++-- lib/who_need_help/activities.ex | 72 +++++-- lib/who_need_help/activities/activity.ex | 1 + lib/who_need_help/application.ex | 16 +- lib/who_need_help/catalog.ex | 23 +- .../catalog/category_proposal.ex | 1 + lib/who_need_help/catalog_moderation.ex | 2 +- lib/who_need_help/help.ex | 103 +++++++-- lib/who_need_help/messaging.ex | 12 +- lib/who_need_help/tracking.ex | 189 ++++++++++------- .../tracking_presence_cleanup.ex | 32 +++ lib/who_need_help/trust.ex | 107 +++++++--- lib/who_need_help/workers/expire_requests.ex | 70 ++++--- .../controllers/metrics_controller.ex | 12 +- .../live/activity_live/index.ex | 2 +- .../live/category_proposal_live.ex | 2 +- lib/who_need_help_web/live/moderation_live.ex | 2 +- .../live/request_live/show.ex | 92 +++++--- lib/who_need_help_web/metrics_access.ex | 11 + lib/who_need_help_web/presence.ex | 53 +++++ lib/who_need_help_web/telemetry.ex | 81 +++++++- lib/who_need_help_web/worker_metrics_plug.ex | 76 +++++++ .../dashboards/who-need-help-overview.json | 196 ++++++++++++++++++ ops/observability/prometheus.template.yml | 10 + ops/observability/rules.yml | 7 + scripts/clean-deploy-verify.sh | 10 +- scripts/e2e-run.sh | 2 +- scripts/ensure-local-e2e-env.sh | 8 +- scripts/ensure-local-load-env.sh | 3 + scripts/load-run.sh | 24 ++- scripts/observability-run.sh | 112 ++++++++-- scripts/quality.sh | 121 ++++++++++- scripts/upgrade-rehearsal-compose.sh | 4 +- test/who_need_help/accounts_test.exs | 23 ++ test/who_need_help/activities_test.exs | 5 + test/who_need_help/mutual_aid_flow_test.exs | 192 +++++++++++++++++ .../workers/expire_requests_test.exs | 65 ++++++ .../worker_metrics_plug_test.exs | 37 ++++ 54 files changed, 1791 insertions(+), 297 deletions(-) create mode 100644 Dockerfile.postgis create mode 100644 Dockerfile.socket-proxy create mode 100644 compose.cpu-replay.yaml create mode 100644 lib/who_need_help/tracking_presence_cleanup.ex create mode 100644 lib/who_need_help_web/metrics_access.ex create mode 100644 lib/who_need_help_web/worker_metrics_plug.ex create mode 100644 test/who_need_help/workers/expire_requests_test.exs create mode 100644 test/who_need_help_web/worker_metrics_plug_test.exs diff --git a/.dialyzer_ignore.exs b/.dialyzer_ignore.exs index c01bce1..4661c98 100644 --- a/.dialyzer_ignore.exs +++ b/.dialyzer_ignore.exs @@ -11,6 +11,5 @@ # backend path and the same warning class. [ {"lib/who_need_help/help.ex", :call_without_opaque}, - {"lib/who_need_help/tracking.ex", :call_without_opaque}, {"lib/who_need_help_web/gettext.ex", :call_without_opaque} ] diff --git a/.env.e2e.example b/.env.e2e.example index b5de97b..e42a29e 100644 --- a/.env.e2e.example +++ b/.env.e2e.example @@ -3,6 +3,7 @@ # reuses the staging Compose project or database volume. HTTP_PORT=0 MAILPIT_PORT=0 +DOCKER_SOCKET_GID=REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID TRAEFIK_TRUSTED_IPS=127.0.0.1/32 TRAEFIK_RETRY_ATTEMPTS=3 TRAEFIK_PROJECT_CONSTRAINT=GENERATED_UNIQUE_E2E_PROJECT @@ -19,7 +20,11 @@ POSTGRES_DB=who_need_help_e2e POSTGRES_USER=postgres POSTGRES_PASSWORD=GENERATE_INDEPENDENT_E2E_DATABASE_PASSWORD DATABASE_URL=ecto://postgres:GENERATE_URL_SAFE_PASSWORD@db/who_need_help_e2e -POOL_SIZE=10 +WEB_POOL_SIZE=4 +WORKER_POOL_SIZE=2 +MIGRATE_POOL_SIZE=2 +WEB_REPLICAS=2 +WORKER_REPLICAS=2 ERLANG_PORT_LIMIT=65536 SECRET_KEY_BASE=GENERATE_INDEPENDENT_E2E_SECRET_KEY_BASE diff --git a/.env.example b/.env.example index d7f4c7f..ee2811b 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,8 @@ # required values. Replace every credential before any public deployment. HTTP_PORT=4010 MAILPIT_PORT=8027 +MAILPIT_BIND_ADDRESS=127.0.0.1 +DOCKER_SOCKET_GID=REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID # Comma-separated proxy IP/CIDR values whose X-Forwarded-* headers Traefik # accepts. Keep loopback locally; set the exact VPN proxy address for staging. TRAEFIK_TRUSTED_IPS=127.0.0.1/32 @@ -10,7 +12,7 @@ TRAEFIK_RETRY_ATTEMPTS=3 # own project constraint, router/service name, Docker network, and Host rule. TRAEFIK_PROJECT_CONSTRAINT=who_need_help TRAEFIK_APP_NAME=who-need-help -TRAEFIK_DOCKER_NETWORK=who_need_help_internal +TRAEFIK_DOCKER_NETWORK=who_need_help_ingress TRAEFIK_ROUTER_RULE='PathPrefix(`/`)' PHX_HOST=localhost PHX_SCHEME=http @@ -62,7 +64,11 @@ POSTGRES_USER=postgres POSTGRES_PASSWORD=replace-with-a-local-or-deployment-secret DATABASE_URL=ecto://postgres:replace-with-url-encoded-password@db/who_need_help -POOL_SIZE=10 +WEB_POOL_SIZE=4 +WORKER_POOL_SIZE=2 +MIGRATE_POOL_SIZE=2 +WEB_REPLICAS=2 +WORKER_REPLICAS=2 # Maximum simultaneously existing Erlang ports (files, sockets and drivers). # Keeping this explicit prevents a host's very large nofile ulimit from making # every BEAM instance preallocate a multi-gigabyte port table. diff --git a/.env.load.example b/.env.load.example index 1dd5f37..d402f29 100644 --- a/.env.load.example +++ b/.env.load.example @@ -6,6 +6,7 @@ LOAD_PROJECT=who_need_help_load LOAD_HOST=load.local HTTP_PORT=4110 MAILPIT_PORT=8127 +DOCKER_SOCKET_GID=REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID PHX_HOST=load.local PHX_SCHEME=https PHX_URL_PORT=443 @@ -13,14 +14,18 @@ TRAEFIK_TRUSTED_IPS=127.0.0.1/32 TRAEFIK_RETRY_ATTEMPTS=3 TRAEFIK_PROJECT_CONSTRAINT=who_need_help_load TRAEFIK_APP_NAME=who-need-help-load -TRAEFIK_DOCKER_NETWORK=who_need_help_load_internal +TRAEFIK_DOCKER_NETWORK=who_need_help_load_ingress TRAEFIK_ROUTER_RULE='Host(`load.local`)' POSTGRES_DB=who_need_help_load POSTGRES_USER=wnh_load POSTGRES_PASSWORD=GENERATE_POSTGRES_PASSWORD DATABASE_URL=GENERATE_DATABASE_URL -POOL_SIZE=10 +WEB_POOL_SIZE=4 +WORKER_POOL_SIZE=2 +MIGRATE_POOL_SIZE=2 +WEB_REPLICAS=2 +WORKER_REPLICAS=2 ERLANG_PORT_LIMIT=65536 SECRET_KEY_BASE=GENERATE_SECRET_KEY_BASE HANDOVER_SECRET=GENERATE_HANDOVER_SECRET diff --git a/Dockerfile.postgis b/Dockerfile.postgis new file mode 100644 index 0000000..0109a66 --- /dev/null +++ b/Dockerfile.postgis @@ -0,0 +1,9 @@ +# syntax=docker/dockerfile:1.20.0 + +FROM postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6 + +# PostgreSQL 18's image-owned volume path is already writable by this account. +# Starting non-root removes the entrypoint's need for the bundled gosu binary. +RUN rm /usr/local/bin/gosu + +USER postgres diff --git a/Dockerfile.socket-proxy b/Dockerfile.socket-proxy new file mode 100644 index 0000000..30088c9 --- /dev/null +++ b/Dockerfile.socket-proxy @@ -0,0 +1,11 @@ +# syntax=docker/dockerfile:1.20.0 +FROM tecnativa/docker-socket-proxy:v0.4.2@sha256:1f3a6f303320723d199d2316a3e82b2e2685d86c275d5e3deeaf182573b47476 + +USER root + +# The upstream release image is immutable but its Alpine packages predate +# currently available security fixes. Keep the reviewed proxy implementation +# and apply the repository's current fixes; scripts/quality.sh scans the result. +RUN apk upgrade --no-cache + +USER haproxy diff --git a/compose.cpu-replay.yaml b/compose.cpu-replay.yaml new file mode 100644 index 0000000..6c60c2c --- /dev/null +++ b/compose.cpu-replay.yaml @@ -0,0 +1,14 @@ +# Optional measurement-only override. It constrains CPU visibility for each +# long-running BEAM container so scheduler-count and memory behavior can be +# reproduced on a many-core development host. Values are experiment inputs, +# not production recommendations or minimum requirements. +services: + web: + cpus: ${CPU_REPLAY_WEB_CPUS:?Set CPU_REPLAY_WEB_CPUS for this experiment} + environment: + ERL_ZFLAGS: "+Q ${ERLANG_PORT_LIMIT:-65536} +S ${CPU_REPLAY_WEB_SCHEDULERS:?Set CPU_REPLAY_WEB_SCHEDULERS}:${CPU_REPLAY_WEB_SCHEDULERS:?Set CPU_REPLAY_WEB_SCHEDULERS}" + + worker: + cpus: ${CPU_REPLAY_WORKER_CPUS:?Set CPU_REPLAY_WORKER_CPUS for this experiment} + environment: + ERL_ZFLAGS: "+Q ${ERLANG_PORT_LIMIT:-65536} +S ${CPU_REPLAY_WORKER_SCHEDULERS:?Set CPU_REPLAY_WORKER_SCHEDULERS}:${CPU_REPLAY_WORKER_SCHEDULERS:?Set CPU_REPLAY_WORKER_SCHEDULERS}" diff --git a/compose.load.yaml b/compose.load.yaml index 62400e8..1381d82 100644 --- a/compose.load.yaml +++ b/compose.load.yaml @@ -15,7 +15,7 @@ services: edge: aliases: - ${LOAD_HOST} - internal: {} + ingress: {} migrate: image: who-need-help:load diff --git a/compose.quality.yaml b/compose.quality.yaml index 7ce7a73..0e8120c 100644 --- a/compose.quality.yaml +++ b/compose.quality.yaml @@ -1,6 +1,6 @@ services: db: - image: postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6 + image: ${QUALITY_POSTGIS_IMAGE:?Set QUALITY_POSTGIS_IMAGE} environment: POSTGRES_DB: postgres POSTGRES_USER: ${QUALITY_POSTGRES_USER:?Set QUALITY_POSTGRES_USER} diff --git a/compose.yaml b/compose.yaml index 1ab0648..f3c6627 100644 --- a/compose.yaml +++ b/compose.yaml @@ -16,7 +16,6 @@ x-app-environment: &app-environment PHX_SCHEME: ${PHX_SCHEME:?Set PHX_SCHEME in .env} PHX_URL_PORT: ${PHX_URL_PORT:?Set PHX_URL_PORT in .env} PORT: "4000" - POOL_SIZE: ${POOL_SIZE:?Set POOL_SIZE in .env after measuring the target profile} SMTP_RELAY: ${SMTP_RELAY:?Set SMTP_RELAY in .env} SMTP_PORT: ${SMTP_PORT:?Set SMTP_PORT in .env} SMTP_USERNAME: ${SMTP_USERNAME:-} @@ -45,11 +44,40 @@ x-app-environment: &app-environment PUSH_HTTP_RETRY_DELAY_MS: ${PUSH_HTTP_RETRY_DELAY_MS:-} services: + docker-api-proxy: + image: who-need-help:socket-proxy-local + build: + context: . + dockerfile: Dockerfile.socket-proxy + environment: + CONTAINERS: "1" + EVENTS: "1" + INFO: "1" + NETWORKS: "1" + PING: "1" + POST: "0" + VERSION: "1" + LOG_LEVEL: warning + volumes: + - /var/run/docker.sock:/var/run/docker.sock:ro + group_add: + - ${DOCKER_SOCKET_GID:?Set DOCKER_SOCKET_GID to the numeric group of /var/run/docker.sock} + networks: [docker-api] + read_only: true + tmpfs: + - /run:uid=99,gid=99,mode=0755 + - /tmp + cap_drop: [ALL] + security_opt: + - no-new-privileges:true + restart: unless-stopped + proxy: image: traefik:v3.7.8@sha256:4299bbed850421258fc5448c2e0e6ad350981d4d335a68de11b92448aedbefe5 command: - --api.dashboard=false - --providers.docker=true + - --providers.docker.endpoint=tcp://docker-api-proxy:2375 - --providers.docker.exposedbydefault=false - --providers.docker.constraints=Label(`com.docker.compose.project`,`${TRAEFIK_PROJECT_CONSTRAINT:-who_need_help}`) - --entrypoints.web.address=:80 @@ -57,13 +85,22 @@ services: - --entrypoints.web.forwardedheaders.trustedips=${TRAEFIK_TRUSTED_IPS:-127.0.0.1/32} ports: - "${HTTP_PORT:-4010}:80" - volumes: - - /var/run/docker.sock:/var/run/docker.sock:ro - networks: [edge, internal] + depends_on: + - docker-api-proxy + networks: [docker-api, edge, ingress] + read_only: true + tmpfs: + - /tmp + cap_drop: [ALL] + security_opt: + - no-new-privileges:true restart: unless-stopped db: - image: postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6 + image: who-need-help:postgis-local + build: + context: . + dockerfile: Dockerfile.postgis environment: POSTGRES_DB: ${POSTGRES_DB:?Set POSTGRES_DB in .env} POSTGRES_USER: ${POSTGRES_USER:?Set POSTGRES_USER in .env} @@ -79,13 +116,22 @@ services: volumes: - postgres_data:/var/lib/postgresql networks: [internal] + security_opt: + - no-new-privileges:true restart: unless-stopped mailpit: image: axllent/mailpit:v1.30.4@sha256:5a49a77c5bdbe7c5474450b4f46348d09949df3695257729c93a30369382d4f6 + user: "65534:65534" ports: - - "${MAILPIT_PORT:-8027}:8025" + - "${MAILPIT_BIND_ADDRESS:-127.0.0.1}:${MAILPIT_PORT:-8027}:8025" networks: [edge, internal] + read_only: true + tmpfs: + - /tmp + cap_drop: [ALL] + security_opt: + - no-new-privileges:true restart: unless-stopped migrate: @@ -95,11 +141,18 @@ services: environment: <<: *app-environment APP_ROLE: migrate + POOL_SIZE: ${MIGRATE_POOL_SIZE:-${POOL_SIZE:?Set MIGRATE_POOL_SIZE in .env}} command: ["/app/bin/migrate"] depends_on: db: condition: service_healthy networks: [internal] + read_only: true + tmpfs: + - /tmp + cap_drop: [ALL] + security_opt: + - no-new-privileges:true restart: "no" web: @@ -108,12 +161,13 @@ services: <<: *app-environment APP_ROLE: web PHX_SERVER: "true" + POOL_SIZE: ${WEB_POOL_SIZE:-${POOL_SIZE:?Set WEB_POOL_SIZE in .env}} depends_on: migrate: condition: service_completed_successfully labels: - traefik.enable=true - - traefik.docker.network=${TRAEFIK_DOCKER_NETWORK:-who_need_help_internal} + - traefik.docker.network=${TRAEFIK_DOCKER_NETWORK:-who_need_help_ingress} - traefik.http.routers.${TRAEFIK_APP_NAME:-who-need-help}.rule=${TRAEFIK_ROUTER_RULE:-PathPrefix(`/`)} - traefik.http.routers.${TRAEFIK_APP_NAME:-who-need-help}.entrypoints=web - traefik.http.routers.${TRAEFIK_APP_NAME:-who-need-help}.service=${TRAEFIK_APP_NAME:-who-need-help} @@ -126,8 +180,14 @@ services: timeout: 3s retries: 10 deploy: - replicas: 2 - networks: [internal] + replicas: ${WEB_REPLICAS:-2} + networks: [ingress, internal, egress] + read_only: true + tmpfs: + - /tmp + cap_drop: [ALL] + security_opt: + - no-new-privileges:true restart: unless-stopped worker: @@ -135,19 +195,36 @@ services: environment: <<: *app-environment APP_ROLE: worker + POOL_SIZE: ${WORKER_POOL_SIZE:-${POOL_SIZE:?Set WORKER_POOL_SIZE in .env}} command: ["/app/bin/who_need_help", "start"] depends_on: migrate: condition: service_completed_successfully + healthcheck: + test: ["CMD", "curl", "--fail", "--silent", "http://localhost:4000/healthz/ready"] + interval: 10s + timeout: 3s + retries: 10 deploy: - replicas: 2 - networks: [internal] + replicas: ${WORKER_REPLICAS:-2} + networks: [internal, egress] + read_only: true + tmpfs: + - /tmp + cap_drop: [ALL] + security_opt: + - no-new-privileges:true restart: unless-stopped networks: + docker-api: + internal: true edge: + ingress: + internal: true internal: internal: true + egress: volumes: postgres_data: diff --git a/config/runtime.exs b/config/runtime.exs index de75ed7..b58f412 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -179,16 +179,16 @@ config :who_need_help, ), push_delivery_options: Keyword.delete(push_configuration, :adapter) -if config_env() == :prod and app_role == :web do +if config_env() == :prod and app_role in [:web, :worker] do metrics_token = System.get_env("METRICS_TOKEN") || raise """ - environment variable METRICS_TOKEN is missing for the web role. + environment variable METRICS_TOKEN is missing for the web or worker role. Generate an independent random value and store it in the deployment secret. """ if metrics_token == "" do - raise "METRICS_TOKEN must not be empty for the web role." + raise "METRICS_TOKEN must not be empty for the web or worker role." end config :who_need_help, :metrics_token, metrics_token diff --git a/deploy/helm/who-need-help/templates/deployments.yaml b/deploy/helm/who-need-help/templates/deployments.yaml index d147fca..33b4ba3 100644 --- a/deploy/helm/who-need-help/templates/deployments.yaml +++ b/deploy/helm/who-need-help/templates/deployments.yaml @@ -80,7 +80,7 @@ spec: - name: PORT value: {{ $root.Values.app.port | quote }} - name: POOL_SIZE - value: {{ $root.Values.app.poolSize | quote }} + value: {{ $settings.poolSize | quote }} - name: SMTP_RELAY value: {{ $root.Values.app.smtpRelay | quote }} - name: SMTP_PORT @@ -93,7 +93,6 @@ spec: value: {{ $root.Values.app.mapTileUrl | quote }} - name: DNS_CLUSTER_QUERY value: "{{ include "who-need-help.fullname" $root }}-headless.{{ $root.Release.Namespace }}.svc.cluster.local" - {{- if eq $component "web" }} ports: - name: http containerPort: 4000 @@ -107,7 +106,6 @@ spec: path: /healthz/live port: http periodSeconds: 10 - {{- end }} securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true diff --git a/deploy/helm/who-need-help/templates/migrate-job.yaml b/deploy/helm/who-need-help/templates/migrate-job.yaml index 9a5493c..80c4db3 100644 --- a/deploy/helm/who-need-help/templates/migrate-job.yaml +++ b/deploy/helm/who-need-help/templates/migrate-job.yaml @@ -35,6 +35,8 @@ spec: value: migrate - name: ERL_ZFLAGS value: {{ printf "+Q %d" (int .Values.app.erlangPortLimit) | quote }} + - name: POOL_SIZE + value: {{ .Values.app.migratePoolSize | quote }} securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true diff --git a/deploy/helm/who-need-help/templates/services.yaml b/deploy/helm/who-need-help/templates/services.yaml index 856c1d8..457e0b7 100644 --- a/deploy/helm/who-need-help/templates/services.yaml +++ b/deploy/helm/who-need-help/templates/services.yaml @@ -19,6 +19,23 @@ spec: --- apiVersion: v1 kind: Service +metadata: + name: {{ include "who-need-help.fullname" . }}-worker-metrics + labels: + {{- include "who-need-help.labels" . | nindent 4 }} + app.kubernetes.io/component: worker +spec: + type: ClusterIP + selector: + {{- include "who-need-help.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: worker + ports: + - name: http + port: 4000 + targetPort: http +--- +apiVersion: v1 +kind: Service metadata: name: {{ include "who-need-help.fullname" . }}-headless labels: diff --git a/deploy/helm/who-need-help/values.yaml b/deploy/helm/who-need-help/values.yaml index 03f6e98..4b91478 100644 --- a/deploy/helm/who-need-help/values.yaml +++ b/deploy/helm/who-need-help/values.yaml @@ -5,9 +5,11 @@ image: web: replicas: 2 + poolSize: "4" worker: replicas: 2 + poolSize: "2" service: type: ClusterIP @@ -19,7 +21,7 @@ app: scheme: https urlPort: "443" port: "4000" - poolSize: "10" + migratePoolSize: "2" # OTP otherwise derives this from the container nofile ulimit. Some # Kubernetes runtimes expose a value so large that each BEAM instance # preallocates a multi-gigabyte port table. diff --git a/docs/performance.md b/docs/performance.md index 2562606..2e9c2fa 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -21,6 +21,23 @@ before any rollout. Compose and Helm now set that value explicitly, and the startup and rolling-verification scripts check the effective value in every running web and worker VM. +On a host with more CPUs than the intended server, the optional +`compose.cpu-replay.yaml` override can constrain the CPUs visible to each +long-running BEAM container: + +```sh +CPU_REPLAY_WEB_CPUS=1 CPU_REPLAY_WORKER_CPUS=1 \ +CPU_REPLAY_WEB_SCHEDULERS=1 CPU_REPLAY_WORKER_SCHEDULERS=1 \ + docker compose -f compose.yaml -f compose.cpu-replay.yaml up -d --wait +``` + +All four values are explicit experiment inputs. The override does not set a memory +limit or represent a whole-server CPU budget. The scheduler inputs make total +and online BEAM scheduler counts match the intended replay instead of leaving +offline scheduler threads sized from the development host. None of these +inputs is a production sizing recommendation. Running ordinary +`docker compose up` without the override removes the replay constraints. + After the Helm rollout, the unchanged kind control-plane container measured 1.796 GiB, a reduction of 7.952 GiB (81.6%). Its four application pod cgroups measured 188.5–215.6 MiB in the first post-change sample. A subsequent @@ -295,6 +312,35 @@ Ignored before/after evidence: - `output/performance/pool-sql-final-20260720/` - `output/performance/pool-sql-chat-fixed-canonical-20260720/` +## Observed role-specific connection pools + +On 2026-07-20 the same 3-web/2-worker, 40 public HTTP VU, 40 Phoenix +heartbeat VU, and 8 authenticated mutual-aid VU profile was repeated with +four Repo connections per web replica and two per worker replica. The migrate +role also started and completed with a two-connection pool. These are measured +development defaults, not production minimums. + +The 16 configured long-running Repo slots produced 38,678 HTTP requests with +zero failures, 1,760 complete authenticated page/chat/tracking chains, and 240 +Phoenix heartbeat sessions. The application recorded 131,409 web-side queries +with 34.82 microseconds average pool queue time, compared with 31.15 +microseconds in the earlier 10-connections-per-replica run. PostgreSQL observed +21 peak client backends, 6 peak active backends, 76 minimum non-reserved +connection headroom, no idle transaction, no lock wait, no rollback, no +deadlock, no conflict, and no temporary file. + +The run-scoped application log had no match for DBConnection/Postgrex +connection errors, pool timeouts, exceptions, deadlocks, `FATAL`, or `PANIC`. +All 1,760 messages, tracking sessions, and tracking samples were recorded; +explicit stop left zero current positions, and fixture cleanup restored the +tracked table counts exactly. The result supports using web=4, worker=2, and +migrate=2 as the repository's current measured starting profile. It does not +establish saturation, an SLO, or a production capacity limit. + +Ignored evidence: + +- `output/performance/role-pools-4-2-20260720/` + ## Observed 10-minute authenticated soak Observed on 2026-07-19 with the same 3-web/2-worker topology and the same diff --git a/lib/who_need_help/accounts.ex b/lib/who_need_help/accounts.ex index 2f331a8..c5e32a8 100644 --- a/lib/who_need_help/accounts.ex +++ b/lib/who_need_help/accounts.ex @@ -11,6 +11,43 @@ defmodule WhoNeedHelp.Accounts do ## Database getters + @public_user_fields [:id, :display_name, :bio, :tip_url, :inserted_at] + @public_social_identity_fields [ + :id, + :provider, + :profile_url, + :handle, + :verified_at, + :user_id, + :inserted_at + ] + + @doc """ + Returns the deliberately small user projection used by public and + participant-facing views. + + Authentication, account settings, and moderation queries must continue to + load the full schema explicitly. In particular, this projection never loads + email, password hashes, roles, or moderation notes. + """ + def public_user_query(options \\ []) do + query = + from user in User, + select: struct(user, ^@public_user_fields) + + if Keyword.get(options, :social_identities, false) do + public_social_identity = public_social_identity_query() + preload(query, social_identities: ^public_social_identity) + else + query + end + end + + def public_social_identity_query do + from identity in SocialIdentity, + select: struct(identity, ^@public_social_identity_fields) + end + @doc """ Gets a user by email. @@ -72,33 +109,31 @@ defmodule WhoNeedHelp.Accounts do def eligible_for_trust_actions?(_user), do: false def eligible_user_id?(user_id) do - case Repo.get(User, user_id) do - %User{} = user -> eligible_for_trust_actions?(user) - nil -> false - end + Repo.exists?( + from user in User, + where: + user.id == ^user_id and user.moderation_status == :active and + not is_nil(user.confirmed_at) and not is_nil(user.accepted_terms_at) + ) end def moderator?(%User{role: role}), do: role in [:moderator, :admin] def moderator?(_user), do: false - def moderator_authorized?(%User{id: id}) do - case Repo.get(User, id) do - %User{} = user -> moderator?(user) - nil -> false - end - end + def moderator_authorized?(%User{id: id, role: role}) + when role in [:moderator, :admin], + do: + Repo.exists?( + from user in User, where: user.id == ^id and user.role in [:moderator, :admin] + ) def moderator_authorized?(_user), do: false def admin?(%User{role: :admin}), do: true def admin?(_user), do: false - def admin_authorized?(%User{id: id}) do - case Repo.get(User, id) do - %User{} = user -> admin?(user) - nil -> false - end - end + def admin_authorized?(%User{id: id, role: :admin}), + do: Repo.exists?(from user in User, where: user.id == ^id and user.role == :admin) def admin_authorized?(_user), do: false diff --git a/lib/who_need_help/activities.ex b/lib/who_need_help/activities.ex index 0206785..e3eb830 100644 --- a/lib/who_need_help/activities.ex +++ b/lib/who_need_help/activities.ex @@ -5,6 +5,7 @@ defmodule WhoNeedHelp.Activities do import Ecto.Query + alias WhoNeedHelp.Accounts alias WhoNeedHelp.Accounts.Scope alias WhoNeedHelp.Activities.{Activity, Message, Participant} alias WhoNeedHelp.Catalog @@ -36,6 +37,7 @@ defmodule WhoNeedHelp.Activities do now = DateTime.utc_now(:second) limit = Pagination.limit(options) cursor = Pagination.cursor(options) + public_creator = Accounts.public_user_query(social_identities: true) Activity |> where( @@ -59,9 +61,9 @@ defmodule WhoNeedHelp.Activities do |> after_open_activity(cursor) |> order_by([activity], asc: activity.starts_at, asc: activity.id) |> limit(^(limit + 1)) - |> preload(category: :parent, creator: :social_identities, participants: :user) + |> with_approved_participant_count() + |> preload([activity], category: :parent, creator: ^public_creator) |> Repo.all() - |> Enum.map(&public_activity/1) |> Pagination.page(limit, &{&1.starts_at, &1.id}) end @@ -82,7 +84,8 @@ defmodule WhoNeedHelp.Activities do |> before_my_activity(cursor) |> order_by([activity], desc: activity.starts_at, desc: activity.id) |> limit(^(limit + 1)) - |> preload(category: :parent, participants: :user) + |> with_approved_participant_count() + |> preload([activity], category: :parent) |> Repo.all() |> Pagination.page(limit, &{&1.starts_at, &1.id}) end @@ -121,13 +124,11 @@ defmodule WhoNeedHelp.Activities do end def get_activity(%Scope{user: user}, id) do - case Repo.get(Activity, id) do + case get_loaded_activity(id) do nil -> {:error, :not_found} activity -> - activity = load_activity(activity.id) - cond do activity.creator_id == user.id -> {:ok, activity} @@ -154,6 +155,7 @@ defmodule WhoNeedHelp.Activities do if activity.creator_id == user.id or approved_participant?(activity, user.id) do limit = Pagination.limit(options, 50) cursor = Pagination.cursor(options) + public_user = Accounts.public_user_query() page = Message @@ -161,7 +163,7 @@ defmodule WhoNeedHelp.Activities do |> before_message(cursor) |> order_by([message], desc: message.inserted_at, desc: message.id) |> limit(^(limit + 1)) - |> preload(:sender) + |> preload([message], sender: ^public_user) |> Repo.all() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) @@ -356,7 +358,7 @@ defmodule WhoNeedHelp.Activities do |> Repo.insert() |> case do {:ok, message} -> - message = Repo.preload(message, :sender) + message = Repo.preload(message, sender: Accounts.public_user_query()) Phoenix.PubSub.broadcast( WhoNeedHelp.PubSub, @@ -559,18 +561,34 @@ defmodule WhoNeedHelp.Activities do end defp load_activity(id) do - recent_messages = - from message in Message, - order_by: [desc: message.inserted_at, desc: message.id], - limit: 50, - preload: :sender - Activity |> Repo.get!(id) - |> Repo.preload( + |> preload_activity() + end + + defp get_loaded_activity(id) do + Activity + |> Repo.get(id) + |> preload_activity() + end + + defp preload_activity(nil), do: nil + + defp preload_activity(activity) do + public_user = Accounts.public_user_query(social_identities: true) + public_sender = Accounts.public_user_query() + + recent_messages = + Message + |> order_by([message], desc: message.inserted_at, desc: message.id) + |> limit(50) + |> preload([message], sender: ^public_sender) + + Repo.preload( + activity, category: :parent, - creator: :social_identities, - participants: [user: :social_identities], + creator: public_user, + participants: [user: public_user], messages: recent_messages ) end @@ -587,8 +605,24 @@ defmodule WhoNeedHelp.Activities do } end - defp public_activity(activity) do - %{activity | participants: Enum.filter(activity.participants, &(&1.status == :approved))} + defp with_approved_participant_count(query) do + counts = + Participant + |> where([participant], participant.status == :approved) + |> group_by([participant], participant.activity_id) + |> select([participant], %{ + activity_id: participant.activity_id, + approved_count: count(participant.id) + }) + + query + |> join(:left, [activity], count in subquery(counts), + on: count.activity_id == activity.id, + as: :approved_counts + ) + |> select_merge([approved_counts: count], %{ + approved_participant_count: fragment("coalesce(?, 0)", count.approved_count) + }) end defp locked_activity(id) do diff --git a/lib/who_need_help/activities/activity.ex b/lib/who_need_help/activities/activity.ex index c570862..a64e3db 100644 --- a/lib/who_need_help/activities/activity.ex +++ b/lib/who_need_help/activities/activity.ex @@ -24,6 +24,7 @@ defmodule WhoNeedHelp.Activities.Activity do field :completed_at, :utc_datetime field :hidden_at, :utc_datetime field :hidden_reason, :string + field :approved_participant_count, :integer, virtual: true, default: 0 belongs_to :creator, WhoNeedHelp.Accounts.User belongs_to :category, WhoNeedHelp.Catalog.Category diff --git a/lib/who_need_help/application.ex b/lib/who_need_help/application.ex index 4e78aab..8db5bf5 100644 --- a/lib/who_need_help/application.ex +++ b/lib/who_need_help/application.ex @@ -26,9 +26,19 @@ defmodule WhoNeedHelp.Application do role_children = case Application.fetch_env!(:who_need_help, :app_role) do - :web -> [{Oban, oban_client_config}, WhoNeedHelpWeb.Presence, WhoNeedHelpWeb.Endpoint] - :worker -> [{Oban, oban_config}] - :migrate -> [{Oban, oban_client_config}] + :web -> + [ + {Oban, oban_client_config}, + WhoNeedHelpWeb.Presence, + WhoNeedHelp.TrackingPresenceCleanup, + WhoNeedHelpWeb.Endpoint + ] + + :worker -> + [{Oban, oban_config}, WhoNeedHelpWeb.WorkerMetricsPlug] + + :migrate -> + [{Oban, oban_client_config}] end children = common_children ++ role_children diff --git a/lib/who_need_help/catalog.ex b/lib/who_need_help/catalog.ex index efd4626..4e0c84e 100644 --- a/lib/who_need_help/catalog.ex +++ b/lib/who_need_help/catalog.ex @@ -40,13 +40,15 @@ defmodule WhoNeedHelp.Catalog do def paginate_proposals(options \\ []) do limit = Pagination.limit(options) cursor = Pagination.cursor(options) + public_user = Accounts.public_user_query() CategoryProposal |> where([p], p.status == :open) - |> preload([:proposer, :parent, :votes]) |> before_proposal(cursor) |> order_by([proposal], desc: proposal.inserted_at, desc: proposal.id) |> limit(^(limit + 1)) + |> with_vote_count() + |> preload([proposal], proposer: ^public_user, parent: []) |> Repo.all() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end @@ -163,7 +165,8 @@ defmodule WhoNeedHelp.Catalog do |> before_proposal(cursor) |> order_by([proposal], desc: proposal.inserted_at, desc: proposal.id) |> limit(^(limit + 1)) - |> preload([:proposer, :parent, :merged_into, :reviewed_by, :votes]) + |> with_vote_count() + |> preload([proposal], [:parent, :merged_into]) |> Repo.all() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) else @@ -833,6 +836,22 @@ defmodule WhoNeedHelp.Catalog do ] end + defp with_vote_count(query) do + counts = + CategoryVote + |> group_by([vote], vote.proposal_id) + |> select([vote], %{proposal_id: vote.proposal_id, vote_count: count(vote.id)}) + + query + |> join(:left, [proposal], count in subquery(counts), + on: count.proposal_id == proposal.id, + as: :vote_counts + ) + |> select_merge([vote_counts: count], %{ + vote_count: fragment("coalesce(?, 0)", count.vote_count) + }) + end + defp before_proposal(query, nil), do: query defp before_proposal(query, {inserted_at, id}) do diff --git a/lib/who_need_help/catalog/category_proposal.ex b/lib/who_need_help/catalog/category_proposal.ex index 318d9c7..abeaac1 100644 --- a/lib/who_need_help/catalog/category_proposal.ex +++ b/lib/who_need_help/catalog/category_proposal.ex @@ -16,6 +16,7 @@ defmodule WhoNeedHelp.Catalog.CategoryProposal do belongs_to :reviewed_by, WhoNeedHelp.Accounts.User field :reviewed_at, :utc_datetime field :moderation_note, :string + field :vote_count, :integer, virtual: true, default: 0 has_many :votes, WhoNeedHelp.Catalog.CategoryVote, foreign_key: :proposal_id timestamps(type: :utc_datetime) diff --git a/lib/who_need_help/catalog_moderation.ex b/lib/who_need_help/catalog_moderation.ex index 8a15ac0..f9dfb14 100644 --- a/lib/who_need_help/catalog_moderation.ex +++ b/lib/who_need_help/catalog_moderation.ex @@ -19,7 +19,7 @@ defmodule WhoNeedHelp.CatalogModeration do reason: proposal.reason, mode: proposal.mode, parent_slug: parent_slug(proposal.parent), - community_votes: length(proposal.votes) + community_votes: proposal.vote_count } end) diff --git a/lib/who_need_help/help.ex b/lib/who_need_help/help.ex index ed56737..97da558 100644 --- a/lib/who_need_help/help.ex +++ b/lib/who_need_help/help.ex @@ -57,8 +57,8 @@ defmodule WhoNeedHelp.Help do |> after_open_request(cursor) |> order_by([request], asc: request.expires_at, asc: request.id) |> limit(^(limit + 1)) - |> preload([:category, :requester, assignment: :helper]) |> Repo.all() + |> preload_request_relations() |> Pagination.page(limit, &{&1.expires_at, &1.id}) end @@ -77,8 +77,8 @@ defmodule WhoNeedHelp.Help do |> before_my_request(cursor) |> order_by([request], desc: request.inserted_at, desc: request.id) |> limit(^(limit + 1)) - |> preload([:category, :requester, assignment: :helper]) |> Repo.all() + |> preload_request_relations() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end @@ -107,31 +107,39 @@ defmodule WhoNeedHelp.Help do def get_request!(id) do HelpRequest |> Repo.get!(id) - |> Repo.preload([ - :category, - requester: :social_identities, - assignment: [helper: :social_identities] - ]) + |> then(&preload_request_relations([&1], social_identities: true)) + |> hd() end def get_request(%Scope{user: user} = scope, id) do - case Repo.get(HelpRequest, id) do + request = + HelpRequest + |> Repo.get(id) + |> case do + nil -> + nil + + request -> + [request] + |> preload_request_relations(social_identities: true) + |> hd() + end + + case request do nil -> {:error, :not_found} request -> - request = get_request!(request.id) - cond do - Accounts.moderator_authorized?(user) -> - {:ok, request} - request.requester_id == user.id -> {:ok, request} request.assignment && participant?(scope, request.assignment) -> {:ok, request} + Accounts.moderator_authorized?(user) -> + {:ok, request} + not is_nil(request.hidden_at) -> {:error, :not_found} @@ -146,7 +154,10 @@ defmodule WhoNeedHelp.Help do def get_assignment_for_participant(%Scope{} = scope, id) do with {:ok, id} <- Ecto.UUID.cast(id), - %Assignment{} = assignment <- Repo.get(Assignment, id), + %Assignment{} = assignment <- + Assignment + |> Repo.get(id) + |> Repo.preload(:request), true <- participant?(scope, assignment) do {:ok, assignment} else @@ -248,7 +259,12 @@ defmodule WhoNeedHelp.Help do {:ok, assignment} -> request = get_request!(assignment.request_id) broadcast({:request_updated, request}) - {:ok, Repo.preload(assignment, [:helper, :request])} + + {:ok, + Repo.preload(assignment, + helper: Accounts.public_user_query(), + request: [] + )} other -> other @@ -389,6 +405,55 @@ defmodule WhoNeedHelp.Help do user.id in [assignment.helper_id, request.requester_id] end + defp preload_request_relations(requests, options \\ []) do + requests = Repo.preload(requests, [:category, :assignment]) + + user_ids = + requests + |> Enum.flat_map(fn request -> + [request.requester_id, request.assignment && request.assignment.helper_id] + end) + |> Enum.reject(&is_nil/1) + |> Enum.uniq() + + users = + Accounts.public_user_query() + |> where([user], user.id in ^user_ids) + |> Repo.all() + |> maybe_preload_social_identities(options) + |> Map.new(&{&1.id, &1}) + + Enum.map(requests, fn request -> + assignment = + case request.assignment do + %Assignment{} = assignment -> + %{assignment | helper: Map.fetch!(users, assignment.helper_id)} + + nil -> + nil + end + + request + |> Map.put(:requester, Map.fetch!(users, request.requester_id)) + |> Map.put(:assignment, assignment) + |> attach_assignment_request() + end) + end + + defp maybe_preload_social_identities(users, options) do + if Keyword.get(options, :social_identities, false) do + Repo.preload(users, social_identities: Accounts.public_social_identity_query()) + else + users + end + end + + defp attach_assignment_request(%HelpRequest{assignment: %Assignment{} = assignment} = request) do + %{request | assignment: %{assignment | request: request}} + end + + defp attach_assignment_request(%HelpRequest{} = request), do: request + def requester?(%Scope{user: user}, %HelpRequest{requester_id: id}), do: user.id == id def helper?(%Scope{user: user}, %Assignment{helper_id: id}), do: user.id == id @@ -506,7 +571,13 @@ defmodule WhoNeedHelp.Help do request = get_request!(assignment.request_id) broadcast({:request_updated, request}) if assignment.status == :completed, do: Trust.record_completion_signals(assignment) - {:ok, Repo.preload(assignment, [:helper, :request], force: true)} + + {:ok, + Repo.preload( + assignment, + [helper: Accounts.public_user_query(), request: []], + force: true + )} end defp after_transition(other), do: other diff --git a/lib/who_need_help/messaging.ex b/lib/who_need_help/messaging.ex index 2f25ef6..c878a7e 100644 --- a/lib/who_need_help/messaging.ex +++ b/lib/who_need_help/messaging.ex @@ -2,6 +2,7 @@ defmodule WhoNeedHelp.Messaging do @moduledoc "Durable match chat with PubSub fan-out after commit." import Ecto.Query + alias WhoNeedHelp.Accounts alias WhoNeedHelp.Accounts.Scope alias WhoNeedHelp.Help alias WhoNeedHelp.Help.Assignment @@ -20,10 +21,13 @@ defmodule WhoNeedHelp.Messaging do end def paginate_messages(%Scope{} = scope, %Assignment{} = assignment, options \\ []) do + assignment = Repo.preload(assignment, :request) + if Trust.eligible?(scope) and Help.participant?(scope, assignment) and not blocked_assignment?(scope, assignment) do limit = Pagination.limit(options, 50) cursor = Pagination.cursor(options) + public_user = Accounts.public_user_query() page = Message @@ -31,7 +35,7 @@ defmodule WhoNeedHelp.Messaging do |> before_message(cursor) |> order_by([message], desc: message.inserted_at, desc: message.id) |> limit(^(limit + 1)) - |> preload(:sender) + |> preload([message], sender: ^public_user) |> Repo.all() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) @@ -42,10 +46,12 @@ defmodule WhoNeedHelp.Messaging do end def send_message(%Scope{user: user} = scope, %Assignment{} = assignment, attrs) do + assignment = Repo.preload(assignment, :request) + with {:ok, _limit} <- Trust.authorize_action(scope, :send_message), true <- Help.participant?(scope, assignment), false <- blocked_assignment?(scope, assignment) do - request = assignment |> Repo.preload(:request) |> Map.fetch!(:request) + request = Map.fetch!(assignment, :request) recipient_id = counterpart_id(user.id, assignment, request) result = @@ -66,7 +72,7 @@ defmodule WhoNeedHelp.Messaging do end) with {:ok, message} <- result do - message = Repo.preload(message, :sender) + message = Repo.preload(message, sender: Accounts.public_user_query()) Phoenix.PubSub.broadcast( WhoNeedHelp.PubSub, diff --git a/lib/who_need_help/tracking.ex b/lib/who_need_help/tracking.ex index 70f5e0c..5bc5427 100644 --- a/lib/who_need_help/tracking.ex +++ b/lib/who_need_help/tracking.ex @@ -9,7 +9,7 @@ defmodule WhoNeedHelp.Tracking do """ import Ecto.Query - alias Ecto.Multi + alias WhoNeedHelp.Accounts alias WhoNeedHelp.Accounts.Scope alias WhoNeedHelp.Help alias WhoNeedHelp.Help.Assignment @@ -21,15 +21,11 @@ defmodule WhoNeedHelp.Tracking do Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, "tracking:#{assignment_id}") end - def active_session?(%Scope{user: user} = scope, %Assignment{} = assignment) do - Help.participant?(scope, assignment) && - Repo.exists?( - from session in TrackingSession, - where: - session.assignment_id == ^assignment.id and session.user_id == ^user.id and - session.active - ) - end + def active_session?(%Scope{user: user} = scope, %Assignment{} = assignment), + do: not is_nil(active_session(scope, assignment, user.id)) + + def active_session(%Scope{user: user} = scope, %Assignment{} = assignment), + do: active_session(scope, assignment, user.id) def start_session( %Scope{user: user} = scope, @@ -97,43 +93,31 @@ defmodule WhoNeedHelp.Tracking do def stop_session(%Scope{user: user} = scope, %Assignment{} = assignment) do if Help.participant?(scope, assignment) do - case Repo.get_by(TrackingSession, - assignment_id: assignment.id, - user_id: user.id, - active: true - ) do - nil -> - {:ok, :already_stopped} - - session -> - now = DateTime.utc_now(:second) - - Multi.new() - |> Multi.delete_all(:positions, where(Position, tracking_session_id: ^session.id)) - |> Multi.update( - :session, - TrackingSession.changeset(session, %{active: false, ended_at: now}) - ) - |> Repo.transaction() - |> case do - {:ok, _} -> - Phoenix.PubSub.broadcast( - WhoNeedHelp.PubSub, - "tracking:#{assignment.id}", - {:tracking_stopped, user.id} - ) - - {:ok, :stopped} - - {:error, _step, reason, _} -> - {:error, reason} - end - end + stop_locked_session(assignment.id, user.id) else {:error, :forbidden} end end + def stop_browser_session(assignment_id, user_id, tracking_session_id) do + with {:ok, assignment_id} <- Ecto.UUID.cast(assignment_id), + {:ok, user_id} <- Ecto.UUID.cast(user_id), + {:ok, tracking_session_id} <- Ecto.UUID.cast(tracking_session_id), + %Assignment{} = assignment <- + Assignment + |> Repo.get(assignment_id) + |> Repo.preload(:request), + %Accounts.User{} = user <- Repo.get(Accounts.User, user_id) do + if Help.participant?(Scope.for_user(user), assignment) do + stop_locked_session(assignment.id, user.id, tracking_session_id) + else + {:error, :forbidden} + end + else + _missing_or_invalid -> {:ok, :already_stopped} + end + end + def list_current_positions(%Scope{} = scope, %Assignment{} = assignment) do if Trust.eligible?(scope) and Help.participant?(scope, assignment) do TrackingSession @@ -176,6 +160,67 @@ defmodule WhoNeedHelp.Tracking do end) end + defp active_session(scope, assignment, user_id) do + if Help.participant?(scope, assignment) do + Repo.get_by(TrackingSession, + assignment_id: assignment.id, + user_id: user_id, + active: true + ) + end + end + + defp stop_locked_session(assignment_id, user_id, tracking_session_id \\ nil) do + result = + Repo.transact(fn -> + query = + from session in TrackingSession, + where: + session.assignment_id == ^assignment_id and session.user_id == ^user_id and + session.active, + lock: "FOR UPDATE" + + query = + if tracking_session_id, + do: where(query, [session], session.id == ^tracking_session_id), + else: query + + case Repo.one(query) do + nil -> + {:ok, :already_stopped} + + session -> + Position + |> where([position], position.tracking_session_id == ^session.id) + |> Repo.delete_all() + + now = DateTime.utc_now(:second) + + case Repo.update(TrackingSession.changeset(session, %{active: false, ended_at: now})) do + {:ok, _session} -> {:ok, :stopped} + {:error, changeset} -> {:error, changeset} + end + end + end) + + case result do + {:ok, :stopped} -> + Phoenix.PubSub.broadcast( + WhoNeedHelp.PubSub, + "tracking:#{assignment_id}", + {:tracking_stopped, user_id} + ) + + {:ok, :stopped} + + {:ok, :already_stopped} -> + {:ok, :already_stopped} + + {:error, reason} -> + {:error, reason} + end + end + defp public_position(%Position{ position: %Geo.Point{coordinates: {lng, lat}}, accuracy_meters: accuracy, @@ -188,7 +233,6 @@ defmodule WhoNeedHelp.Tracking do previous = Position |> where([position], position.tracking_session_id == ^session.id) - |> lock("FOR UPDATE") |> Repo.one() now = DateTime.utc_now(:second) @@ -214,21 +258,9 @@ defmodule WhoNeedHelp.Tracking do }) |> Repo.update() - assignment = - Assignment - |> where([candidate], candidate.id == ^assignment.id) - |> lock("FOR UPDATE") - |> Repo.one!() - - assignment = - maybe_mark_helper_movement(assignment, user.id, session.movement_observed_at, now) - - assignment = maybe_mark_proximity(assignment, session, position, now) - - evidence = %{ - helper_movement_observed_at: assignment.helper_movement_observed_at, - proximity_observed_at: assignment.proximity_observed_at - } + maybe_mark_helper_movement(assignment, user.id, session.movement_observed_at, now) + maybe_mark_proximity(assignment, session, position, now) + evidence = assignment_evidence(assignment.id) {:ok, {position, evidence}} end @@ -237,17 +269,21 @@ defmodule WhoNeedHelp.Tracking do defp maybe_mark_helper_movement(assignment, user_id, movement_observed_at, now) do if assignment.helper_id == user_id and not is_nil(movement_observed_at) and is_nil(assignment.helper_movement_observed_at) do - assignment - |> Assignment.changeset(%{helper_movement_observed_at: now}) - |> Repo.update!() - else - assignment + Assignment + |> where( + [candidate], + candidate.id == ^assignment.id and + is_nil(candidate.helper_movement_observed_at) + ) + |> Repo.update_all(set: [helper_movement_observed_at: now, updated_at: now]) end + + :ok end - defp maybe_mark_proximity(%Assignment{proximity_observed_at: observed} = assignment, _, _, _) + defp maybe_mark_proximity(%Assignment{proximity_observed_at: observed}, _, _, _) when not is_nil(observed), - do: assignment + do: :ok defp maybe_mark_proximity(assignment, session, position, now) do counterpart_positions = @@ -262,12 +298,25 @@ defmodule WhoNeedHelp.Tracking do |> Repo.all() if Enum.any?(counterpart_positions, &accuracy_envelopes_overlap?(&1, position)) do - assignment - |> Assignment.changeset(%{proximity_observed_at: now}) - |> Repo.update!() - else - assignment + Assignment + |> where( + [candidate], + candidate.id == ^assignment.id and is_nil(candidate.proximity_observed_at) + ) + |> Repo.update_all(set: [proximity_observed_at: now, updated_at: now]) end + + :ok + end + + defp assignment_evidence(assignment_id) do + Assignment + |> where([assignment], assignment.id == ^assignment_id) + |> select([assignment], %{ + helper_movement_observed_at: assignment.helper_movement_observed_at, + proximity_observed_at: assignment.proximity_observed_at + }) + |> Repo.one!() end defp movement_evidence(nil, _current), do: 0.0 diff --git a/lib/who_need_help/tracking_presence_cleanup.ex b/lib/who_need_help/tracking_presence_cleanup.ex new file mode 100644 index 0000000..22f066c --- /dev/null +++ b/lib/who_need_help/tracking_presence_cleanup.ex @@ -0,0 +1,32 @@ +defmodule WhoNeedHelp.TrackingPresenceCleanup do + @moduledoc false + + use GenServer + + alias WhoNeedHelp.Tracking + alias WhoNeedHelpWeb.Presence + + def start_link(options), do: GenServer.start_link(__MODULE__, options, name: __MODULE__) + + def maybe_stop(key, assignment_id, user_id, tracking_session_id) do + GenServer.cast( + __MODULE__, + {:maybe_stop, key, assignment_id, user_id, tracking_session_id} + ) + end + + @impl GenServer + def init(_options), do: {:ok, %{}} + + @impl GenServer + def handle_cast( + {:maybe_stop, key, assignment_id, user_id, tracking_session_id}, + state + ) do + if Presence.get_by_key(Presence.tracking_topic(), key) == [] do + Tracking.stop_browser_session(assignment_id, user_id, tracking_session_id) + end + + {:noreply, state} + end +end diff --git a/lib/who_need_help/trust.ex b/lib/who_need_help/trust.ex index 2bed1e7..44ab858 100644 --- a/lib/who_need_help/trust.ex +++ b/lib/who_need_help/trust.ex @@ -25,21 +25,17 @@ defmodule WhoNeedHelp.Trust do } def authorize_action(%Scope{user: %User{} = user}, action) do - user = Repo.get(User, user.id) + if Accounts.eligible_user_id?(user.id) do + case RateLimiter.check(action, user.id) do + {:error, :rate_limited} = error -> + create_velocity_signal_once(user.id, action) + error - cond do - not Accounts.eligible_for_trust_actions?(user) -> - {:error, :account_not_eligible} - - true -> - case RateLimiter.check(action, user.id) do - {:error, :rate_limited} = error -> - create_velocity_signal_once(user.id, action) - error - - result -> - result - end + result -> + result + end + else + {:error, :account_not_eligible} end end @@ -107,26 +103,45 @@ defmodule WhoNeedHelp.Trust do def paginate_visible_reviews(user_id, options \\ []) do limit = Pagination.limit(options) cursor = Pagination.cursor(options) + public_user = Accounts.public_user_query() Review |> where([review], review.reviewee_id == ^user_id and not is_nil(review.revealed_at)) |> before_review(cursor) |> order_by([review], desc: review.inserted_at, desc: review.id) |> limit(^(limit + 1)) - |> preload(:reviewer) + |> preload([review], reviewer: ^public_user) |> Repo.all() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end - def reputation(user_id) do + @empty_reputation %{ + completed: 0, + unique_people: 0, + verified_handovers: 0, + location_supported: 0, + rating: nil + } + + def reputation(user_id), do: Map.fetch!(reputations([user_id]), user_id) + + def reputations([]), do: %{} + + def reputations(user_ids) when is_list(user_ids) do + user_ids = Enum.uniq(user_ids) + helper_rows = Assignment |> join(:inner, [assignment], request in HelpRequest, on: request.id == assignment.request_id ) - |> where([assignment], assignment.status == :completed and assignment.helper_id == ^user_id) + |> where( + [assignment], + assignment.status == :completed and assignment.helper_id in ^user_ids + ) |> select([assignment, request], %{ id: assignment.id, + user_id: assignment.helper_id, counterpart_id: request.requester_id, verified: not is_nil(assignment.handover_verified_at), location_supported: @@ -141,29 +156,64 @@ defmodule WhoNeedHelp.Trust do ) |> where( [assignment, request], - assignment.status == :completed and request.requester_id == ^user_id and - assignment.helper_id != ^user_id + assignment.status == :completed and request.requester_id in ^user_ids and + assignment.helper_id != request.requester_id ) - |> select([assignment], %{ + |> select([assignment, request], %{ id: assignment.id, + user_id: request.requester_id, counterpart_id: assignment.helper_id, verified: not is_nil(assignment.handover_verified_at), location_supported: not is_nil(assignment.proximity_observed_at) }) - aggregate = + aggregates = helper_rows |> union_all(^requester_rows) |> subquery() + |> group_by([row], row.user_id) |> select([row], %{ + user_id: row.user_id, completed: count(row.id), unique_people: count(row.counterpart_id, :distinct), verified_handovers: filter(count(row.id), row.verified), location_supported: filter(count(row.id), row.location_supported) }) - |> Repo.one!() - Map.put(aggregate, :rating, average_rating(user_id)) + ratings = + Review + |> where( + [review], + review.reviewee_id in ^user_ids and not is_nil(review.revealed_at) + ) + |> group_by([review], review.reviewee_id) + |> select([review], %{user_id: review.reviewee_id, rating: avg(review.rating)}) + + rows = + aggregates + |> subquery() + |> join(:left, [aggregate], rating in subquery(ratings), + on: rating.user_id == aggregate.user_id + ) + |> select([aggregate, rating], %{ + user_id: aggregate.user_id, + completed: aggregate.completed, + unique_people: aggregate.unique_people, + verified_handovers: aggregate.verified_handovers, + location_supported: aggregate.location_supported, + rating: rating.rating + }) + |> Repo.all() + + Enum.reduce(rows, Map.new(user_ids, &{&1, @empty_reputation}), fn row, reputations -> + Map.put( + reputations, + row.user_id, + row + |> Map.delete(:user_id) + |> Map.update!(:rating, &decimal_average/1) + ) + end) end def leaderboard do @@ -389,13 +439,14 @@ defmodule WhoNeedHelp.Trust do def paginate_blocks(%Scope{user: user}, options \\ []) do limit = Pagination.limit(options) cursor = Pagination.cursor(options) + public_user = Accounts.public_user_query() Block |> where([block], block.blocker_id == ^user.id) |> before_block(cursor) |> order_by([block], desc: block.inserted_at, desc: block.id) |> limit(^(limit + 1)) - |> preload(:blocked) + |> preload([block], blocked: ^public_user) |> Repo.all() |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end @@ -685,14 +736,6 @@ defmodule WhoNeedHelp.Trust do |> Repo.insert() end - defp average_rating(user_id) do - Review - |> where([review], review.reviewee_id == ^user_id and not is_nil(review.revealed_at)) - |> select([review], avg(review.rating)) - |> Repo.one() - |> decimal_average() - end - defp users_by_id([]), do: %{} defp users_by_id(ids) do diff --git a/lib/who_need_help/workers/expire_requests.ex b/lib/who_need_help/workers/expire_requests.ex index bbb1b6c..e1c9e71 100644 --- a/lib/who_need_help/workers/expire_requests.ex +++ b/lib/who_need_help/workers/expire_requests.ex @@ -12,25 +12,39 @@ defmodule WhoNeedHelp.Workers.ExpireRequests do def perform(_job) do now = DateTime.utc_now(:second) - ids = - HelpRequest - |> where([r], r.status == :open and r.expires_at <= ^now) - |> select([request], request.id) - |> Repo.all() + with {:ok, expired} <- expire_available(now, 0), + {:ok, cleanup} <- Tracking.cleanup_finished_sessions() do + {pruned_buckets, _} = RateLimiter.prune_expired() - expired = - Enum.count(ids, fn id -> - Repo.transact(fn -> - request = - HelpRequest - |> where( - [request], - request.id == ^id and request.status == :open and request.expires_at <= ^now - ) - |> lock("FOR UPDATE") - |> Repo.one() + {:ok, + %{ + expired: expired, + tracking_positions_deleted: cleanup.positions_deleted, + tracking_sessions_ended: cleanup.sessions_ended, + rate_limit_buckets_pruned: pruned_buckets + }} + end + end - if request do + # Claim a single row per short transaction. This keeps memory bounded and + # allows another worker to make progress without waiting for a large backlog + # lock. The recursion is tail-recursive and stops on the first empty claim. + defp expire_available(now, expired) do + result = + Repo.transact(fn -> + request = + HelpRequest + |> where([request], request.status == :open and request.expires_at <= ^now) + |> order_by([request], asc: request.expires_at, asc: request.id) + |> limit(1) + |> lock("FOR UPDATE SKIP LOCKED") + |> Repo.one() + + case request do + nil -> + {:ok, :empty} + + request -> with {:ok, request} <- request |> Ecto.Changeset.change(status: :expired) @@ -39,25 +53,13 @@ defmodule WhoNeedHelp.Workers.ExpireRequests do Trust.audit(nil, "request.expired", "request", request.id) do {:ok, request} end - else - {:error, :already_transitioned} - end - end) - |> case do - {:ok, _request} -> true - _other -> false end end) - {:ok, cleanup} = Tracking.cleanup_finished_sessions() - {pruned_buckets, _} = RateLimiter.prune_expired() - - {:ok, - %{ - expired: expired, - tracking_positions_deleted: cleanup.positions_deleted, - tracking_sessions_ended: cleanup.sessions_ended, - rate_limit_buckets_pruned: pruned_buckets - }} + case result do + {:ok, :empty} -> {:ok, expired} + {:ok, %HelpRequest{}} -> expire_available(now, expired + 1) + {:error, reason} -> {:error, reason} + end end end diff --git a/lib/who_need_help_web/controllers/metrics_controller.ex b/lib/who_need_help_web/controllers/metrics_controller.ex index aea0839..6bf417b 100644 --- a/lib/who_need_help_web/controllers/metrics_controller.ex +++ b/lib/who_need_help_web/controllers/metrics_controller.ex @@ -1,6 +1,8 @@ defmodule WhoNeedHelpWeb.MetricsController do use WhoNeedHelpWeb, :controller + alias WhoNeedHelpWeb.MetricsAccess + @content_type "text/plain; version=0.0.4" # Sobelow's HTML-oriented SendResp check cannot infer this fixed Prometheus @@ -9,7 +11,7 @@ defmodule WhoNeedHelpWeb.MetricsController do def show(conn, _params) do token = Application.get_env(:who_need_help, :metrics_token) - if authorized?(get_req_header(conn, "authorization"), token) do + if MetricsAccess.authorized?(get_req_header(conn, "authorization"), token) do body = TelemetryMetricsPrometheus.Core.scrape(:prometheus_metrics) conn @@ -23,12 +25,4 @@ defmodule WhoNeedHelpWeb.MetricsController do |> send_resp(:unauthorized, "Unauthorized\n") end end - - defp authorized?(["Bearer " <> candidate], expected) - when is_binary(expected) and expected != "" and - byte_size(candidate) == byte_size(expected) do - Plug.Crypto.secure_compare(candidate, expected) - end - - defp authorized?(_authorization, _expected), do: false end diff --git a/lib/who_need_help_web/live/activity_live/index.ex b/lib/who_need_help_web/live/activity_live/index.ex index 3e53d9c..95db499 100644 --- a/lib/who_need_help_web/live/activity_live/index.ex +++ b/lib/who_need_help_web/live/activity_live/index.ex @@ -175,7 +175,7 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do )} - {Enum.count(activity.participants, &(&1.status == :approved))}/{activity.capacity} + {activity.approved_participant_count}/{activity.capacity}

{activity.title}

diff --git a/lib/who_need_help_web/live/category_proposal_live.ex b/lib/who_need_help_web/live/category_proposal_live.ex index b666f06..bfa66d4 100644 --- a/lib/who_need_help_web/live/category_proposal_live.ex +++ b/lib/who_need_help_web/live/category_proposal_live.ex @@ -148,7 +148,7 @@ defmodule WhoNeedHelpWeb.CategoryProposalLive do

diff --git a/lib/who_need_help_web/live/moderation_live.ex b/lib/who_need_help_web/live/moderation_live.ex index 8ed1dd9..10362cc 100644 --- a/lib/who_need_help_web/live/moderation_live.ex +++ b/lib/who_need_help_web/live/moderation_live.ex @@ -478,7 +478,7 @@ defmodule WhoNeedHelpWeb.ModerationLive do

{proposal.reason}

- {gettext("%{count} community votes", count: length(proposal.votes))} + {gettext("%{count} community votes", count: proposal.vote_count)}

<.form diff --git a/lib/who_need_help_web/live/request_live/show.ex b/lib/who_need_help_web/live/request_live/show.ex index 2abdd94..c01c740 100644 --- a/lib/who_need_help_web/live/request_live/show.ex +++ b/lib/who_need_help_web/live/request_live/show.ex @@ -2,6 +2,7 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do use WhoNeedHelpWeb, :live_view alias WhoNeedHelp.{Help, Messaging, Tracking, Trust} + alias WhoNeedHelpWeb.Presence @e2e_routes Application.compile_env(:who_need_help, :e2e_routes, false) @@ -16,6 +17,8 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do socket = socket |> assign(:subscribed_assignment_id, nil) + |> assign(:tracking_presence_key, nil) + |> assign(:tracking_session_id, nil) |> assign(:native_client, native_client) |> assign( :e2e_runtime_node, @@ -33,13 +36,18 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do socket = maybe_subscribe_assignment(socket, request) - tracking_active = + tracking_session = request.assignment && - Tracking.active_session?(socket.assigns.current_scope, request.assignment) + Tracking.active_session(socket.assigns.current_scope, request.assignment) socket = socket - |> load(request, tracking_active) + |> assign( + :tracking_session_id, + if(tracking_session, do: tracking_session.id, else: nil) + ) + |> load(request, not is_nil(tracking_session)) + |> maybe_track_browser_presence() |> maybe_start_native_tracking() {:ok, socket} @@ -88,7 +96,10 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do |> assign_positions(positions) |> then(fn socket -> if user_id == socket.assigns.current_scope.user.id do - assign(socket, :tracking_active, false) + socket + |> assign(:tracking_active, false) + |> assign(:tracking_session_id, nil) + |> maybe_untrack_browser_presence() else socket end @@ -171,10 +182,12 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do def handle_event("start-tracking", _, socket) do case Tracking.start_session(socket.assigns.current_scope, socket.assigns.assignment) do - {:ok, _} -> + {:ok, tracking_session} -> {:noreply, socket |> assign(:tracking_active, true) + |> assign(:tracking_session_id, tracking_session.id) + |> maybe_track_browser_presence() |> maybe_start_native_tracking()} {:error, reason} -> @@ -199,6 +212,8 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do {:noreply, socket |> assign(:tracking_active, false) + |> assign(:tracking_session_id, nil) + |> maybe_untrack_browser_presence() |> maybe_stop_native_tracking()} {:error, reason} -> @@ -268,13 +283,19 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do def handle_event("block-user", _, socket) do case Trust.block(socket.assigns.current_scope, socket.assigns.other_user_id) do {:ok, _block} -> + if socket.assigns.tracking_active and socket.assigns.assignment do + Tracking.stop_session(socket.assigns.current_scope, socket.assigns.assignment) + end + {:noreply, socket + |> maybe_untrack_browser_presence() |> assign(:blocked_by_current, true) |> assign(:messages, []) |> assign(:messages_cursor, nil) |> assign(:positions, %{}) |> assign(:tracking_active, false) + |> assign(:tracking_session_id, nil) |> assign( :markers, Jason.encode!(request_markers(socket.assigns.current_scope, socket.assigns.request)) @@ -302,25 +323,6 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do end end - @impl true - def terminate(_reason, socket) do - case socket.assigns do - %{ - tracking_active: true, - native_client: false, - current_scope: scope, - assignment: assignment - } - when not is_nil(assignment) -> - Tracking.stop_session(scope, assignment) - - _ -> - :ok - end - - :ok - end - defp maybe_start_native_tracking( %{assigns: %{native_client: true, tracking_active: true, assignment: assignment}} = socket @@ -337,6 +339,39 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do defp maybe_stop_native_tracking(socket), do: socket + defp maybe_track_browser_presence( + %{ + assigns: %{ + native_client: false, + tracking_active: true, + tracking_presence_key: nil, + tracking_session_id: tracking_session_id, + assignment: assignment, + current_scope: %{user: user} + } + } = socket + ) + when not is_nil(assignment) do + if connected?(socket) do + case Presence.track_browser(self(), assignment.id, user.id, tracking_session_id) do + {:ok, key} -> assign(socket, :tracking_presence_key, key) + _error -> socket + end + else + socket + end + end + + defp maybe_track_browser_presence(socket), do: socket + + defp maybe_untrack_browser_presence(%{assigns: %{tracking_presence_key: key}} = socket) + when is_binary(key) do + Presence.untrack_browser(self(), key) + assign(socket, :tracking_presence_key, nil) + end + + defp maybe_untrack_browser_presence(socket), do: socket + defp transition(socket, fun, success) do case fun.(socket.assigns.current_scope, socket.assigns.assignment.id) do {:ok, _} -> {:noreply, put_flash(socket, :info, success)} @@ -366,6 +401,11 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do do: Tracking.list_current_positions(socket.assigns.current_scope, assignment), else: %{} + reputations = + [request.requester_id, assignment && assignment.helper_id] + |> Enum.reject(&is_nil/1) + |> Trust.reputations() + socket |> assign(:page_title, request.title) |> assign(:request, request) @@ -373,10 +413,10 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do :structured_details, structured_details(request, Gettext.get_locale(WhoNeedHelpWeb.Gettext)) ) - |> assign(:requester_reputation, Trust.reputation(request.requester_id)) + |> assign(:requester_reputation, Map.fetch!(reputations, request.requester_id)) |> assign( :helper_reputation, - if(assignment, do: Trust.reputation(assignment.helper_id), else: nil) + if(assignment, do: Map.fetch!(reputations, assignment.helper_id), else: nil) ) |> assign(:assignment, assignment) |> assign(:participant, participant) diff --git a/lib/who_need_help_web/metrics_access.ex b/lib/who_need_help_web/metrics_access.ex new file mode 100644 index 0000000..c5216c7 --- /dev/null +++ b/lib/who_need_help_web/metrics_access.ex @@ -0,0 +1,11 @@ +defmodule WhoNeedHelpWeb.MetricsAccess do + @moduledoc false + + def authorized?(["Bearer " <> candidate], expected) + when is_binary(expected) and expected != "" and + byte_size(candidate) == byte_size(expected) do + Plug.Crypto.secure_compare(candidate, expected) + end + + def authorized?(_authorization, _expected), do: false +end diff --git a/lib/who_need_help_web/presence.ex b/lib/who_need_help_web/presence.ex index c16950f..e7670dc 100644 --- a/lib/who_need_help_web/presence.ex +++ b/lib/who_need_help_web/presence.ex @@ -2,4 +2,57 @@ defmodule WhoNeedHelpWeb.Presence do use Phoenix.Presence, otp_app: :who_need_help, pubsub_server: WhoNeedHelp.PubSub + + @tracking_topic "browser-tracking-presence" + + def tracking_topic, do: @tracking_topic + + def tracking_key(assignment_id, user_id), do: "#{assignment_id}:#{user_id}" + + def track_browser(pid, assignment_id, user_id, tracking_session_id) do + key = tracking_key(assignment_id, user_id) + + case track(pid, @tracking_topic, key, %{ + assignment_id: assignment_id, + user_id: user_id, + tracking_session_id: tracking_session_id + }) do + {:ok, _ref} -> {:ok, key} + {:error, {:already_tracked, _pid, @tracking_topic, ^key}} -> {:ok, key} + other -> other + end + end + + def untrack_browser(pid, key), do: untrack(pid, @tracking_topic, key) + + @impl Phoenix.Presence + def init(_options), do: {:ok, %{}} + + @impl Phoenix.Presence + def handle_metas(@tracking_topic, %{leaves: leaves}, presences, state) do + Enum.each(leaves, fn {key, %{metas: metas}} -> + if not Map.has_key?(presences, key) do + case List.first(metas) do + %{ + assignment_id: assignment_id, + user_id: user_id, + tracking_session_id: tracking_session_id + } -> + WhoNeedHelp.TrackingPresenceCleanup.maybe_stop( + key, + assignment_id, + user_id, + tracking_session_id + ) + + _other -> + :ok + end + end + end) + + {:ok, state} + end + + def handle_metas(_topic, _diff, _presences, state), do: {:ok, state} end diff --git a/lib/who_need_help_web/telemetry.ex b/lib/who_need_help_web/telemetry.ex index f8f8d30..eba7ced 100644 --- a/lib/who_need_help_web/telemetry.ex +++ b/lib/who_need_help_web/telemetry.ex @@ -9,7 +9,7 @@ defmodule WhoNeedHelpWeb.Telemetry do @impl true def init(_arg) do reporter_children = - if Application.fetch_env!(:who_need_help, :app_role) == :web do + if Application.fetch_env!(:who_need_help, :app_role) in [:web, :worker] do [ {TelemetryMetricsPrometheus.Core, name: :prometheus_metrics, metrics: prometheus_metrics(), start_async: false} @@ -43,6 +43,15 @@ defmodule WhoNeedHelpWeb.Telemetry do end, description: "Cumulative HTTP request duration" ), + distribution("who_need_help.http.request.duration.seconds", + event_name: [:phoenix, :endpoint, :stop], + measurement: :duration, + unit: {:native, :second}, + reporter_options: [ + buckets: [0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0] + ], + description: "HTTP request duration distribution" + ), counter("who_need_help.http.exceptions.total", event_name: [:phoenix, :router_dispatch, :exception], measurement: :duration, @@ -76,6 +85,48 @@ defmodule WhoNeedHelpWeb.Telemetry do end, description: "Cumulative time waiting for a database connection" ), + distribution("who_need_help.database.query.duration.seconds", + event_name: [:who_need_help, :repo, :query], + measurement: :total_time, + unit: {:native, :second}, + reporter_options: [ + buckets: [ + 0.000_05, + 0.000_1, + 0.000_25, + 0.000_5, + 0.001, + 0.0025, + 0.005, + 0.01, + 0.025, + 0.05, + 0.1 + ] + ], + description: "Database query duration distribution" + ), + distribution("who_need_help.database.query.queue.duration.seconds", + event_name: [:who_need_help, :repo, :query], + measurement: fn measurements -> Map.get(measurements, :queue_time, 0) end, + unit: {:native, :second}, + reporter_options: [ + buckets: [ + 0.000_01, + 0.000_025, + 0.000_05, + 0.000_1, + 0.000_25, + 0.000_5, + 0.001, + 0.0025, + 0.005, + 0.01, + 0.025 + ] + ], + description: "Time waiting for an Ecto connection" + ), sum("who_need_help.database.query.decode.duration.microseconds.total", event_name: [:who_need_help, :repo, :query], measurement: fn measurements -> @@ -90,6 +141,34 @@ defmodule WhoNeedHelpWeb.Telemetry do measurement: :duration, description: "Completed WebSocket connection attempts" ), + counter("who_need_help.oban.jobs.completed.total", + event_name: [:oban, :job, :stop], + measurement: :duration, + tags: [:queue], + description: "Completed Oban jobs" + ), + counter("who_need_help.oban.jobs.failed.total", + event_name: [:oban, :job, :exception], + measurement: :duration, + tags: [:queue], + description: "Failed Oban job attempts" + ), + sum("who_need_help.oban.job.duration.microseconds.total", + event_name: [:oban, :job, :stop], + measurement: fn measurements -> + System.convert_time_unit(measurements.duration, :native, :microsecond) + end, + tags: [:queue], + description: "Cumulative successful Oban execution duration" + ), + sum("who_need_help.oban.job.queue.duration.microseconds.total", + event_name: [:oban, :job, :stop], + measurement: fn measurements -> + System.convert_time_unit(measurements.queue_time, :native, :microsecond) + end, + tags: [:queue], + description: "Cumulative Oban queue wait duration" + ), last_value("who_need_help.vm.memory.total.bytes", event_name: [:vm, :memory], measurement: :total, diff --git a/lib/who_need_help_web/worker_metrics_plug.ex b/lib/who_need_help_web/worker_metrics_plug.ex new file mode 100644 index 0000000..6e8d39a --- /dev/null +++ b/lib/who_need_help_web/worker_metrics_plug.ex @@ -0,0 +1,76 @@ +defmodule WhoNeedHelpWeb.WorkerMetricsPlug do + @moduledoc """ + Internal-only health and Prometheus endpoint for worker-role releases. + + Compose and Kubernetes keep this listener off public ingress. Metrics still + require the same independent bearer credential as web replicas. + """ + + import Plug.Conn + + alias WhoNeedHelpWeb.MetricsAccess + + @content_type "text/plain; version=0.0.4" + + def init(options), do: options + + def call(%Plug.Conn{method: "GET", path_info: ["healthz", "live"]} = conn, _options) do + json(conn, :ok, %{status: "ok", node: to_string(Node.self()), role: "worker"}) + end + + def call(%Plug.Conn{method: "GET", path_info: ["healthz", "ready"]} = conn, _options) do + database_ready? = + match?({:ok, _}, Ecto.Adapters.SQL.query(WhoNeedHelp.Repo, "SELECT 1", [])) + + if database_ready? and is_pid(Oban.whereis(Oban)) do + json(conn, :ok, %{status: "ready", node: to_string(Node.self()), role: "worker"}) + else + json(conn, :service_unavailable, %{status: "not_ready", role: "worker"}) + end + end + + # Sobelow's HTML-oriented SendResp check cannot infer this fixed Prometheus + # text content type. The response is generated by the metrics registry and + # is not rendered in an HTML context. + # sobelow_skip ["XSS.SendResp"] + def call(%Plug.Conn{method: "GET", path_info: ["metrics"]} = conn, _options) do + token = Application.get_env(:who_need_help, :metrics_token) + + if MetricsAccess.authorized?(get_req_header(conn, "authorization"), token) do + body = TelemetryMetricsPrometheus.Core.scrape(:prometheus_metrics) + + conn + |> put_resp_header("cache-control", "no-store") + |> put_resp_content_type(@content_type) + |> send_resp(:ok, body) + else + conn + |> put_resp_header("cache-control", "no-store") + |> put_resp_header("www-authenticate", "Bearer") + |> send_resp(:unauthorized, "Unauthorized\n") + end + end + + def call(conn, _options), do: send_resp(conn, :not_found, "Not found\n") + + def child_spec(_options) do + endpoint_config = Application.fetch_env!(:who_need_help, WhoNeedHelpWeb.Endpoint) + port = endpoint_config |> Keyword.fetch!(:http) |> Keyword.fetch!(:port) + + Bandit.child_spec( + plug: __MODULE__, + scheme: :http, + ip: {0, 0, 0, 0}, + port: port, + startup_log: false + ) + |> Supervisor.child_spec(id: __MODULE__) + end + + defp json(conn, status, body) do + conn + |> put_resp_header("cache-control", "no-store") + |> put_resp_content_type("application/json") + |> send_resp(status, Jason.encode!(body)) + end +end diff --git a/ops/observability/grafana/dashboards/who-need-help-overview.json b/ops/observability/grafana/dashboards/who-need-help-overview.json index a7fd7f0..a9ac15a 100644 --- a/ops/observability/grafana/dashboards/who-need-help-overview.json +++ b/ops/observability/grafana/dashboards/who-need-help-overview.json @@ -173,6 +173,202 @@ ], "title": "Database query rate by replica", "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "wnh-prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 16 + }, + "id": 5, + "targets": [ + { + "editorMode": "code", + "expr": "sum(up{job=\"who-need-help-worker\"})", + "legendFormat": "available", + "range": true, + "refId": "A" + } + ], + "title": "Available worker replicas", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "wnh-prometheus" + }, + "fieldConfig": { + "defaults": { + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 16, + "x": 8, + "y": 16 + }, + "id": 6, + "targets": [ + { + "editorMode": "code", + "expr": "histogram_quantile(0.95, sum by (le, instance) (rate(who_need_help_http_request_duration_seconds_bucket{job=\"who-need-help-web\"}[1m])))", + "legendFormat": "p95 {{instance}}", + "range": true, + "refId": "A" + }, + { + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le, instance) (rate(who_need_help_http_request_duration_seconds_bucket{job=\"who-need-help-web\"}[1m])))", + "legendFormat": "p99 {{instance}}", + "range": true, + "refId": "B" + } + ], + "title": "HTTP request latency by web replica", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "wnh-prometheus" + }, + "fieldConfig": { + "defaults": { + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 24 + }, + "id": 7, + "targets": [ + { + "editorMode": "code", + "expr": "histogram_quantile(0.95, sum by (le, instance) (rate(who_need_help_database_query_duration_seconds_bucket[1m])))", + "legendFormat": "query p95 {{instance}}", + "range": true, + "refId": "A" + }, + { + "editorMode": "code", + "expr": "histogram_quantile(0.95, sum by (le, instance) (rate(who_need_help_database_query_queue_duration_seconds_bucket[1m])))", + "legendFormat": "pool wait p95 {{instance}}", + "range": true, + "refId": "B" + } + ], + "title": "Database query and pool-wait latency", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "wnh-prometheus" + }, + "fieldConfig": { + "defaults": { + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 24 + }, + "id": 8, + "targets": [ + { + "editorMode": "code", + "expr": "sum by (queue) (rate(who_need_help_oban_jobs_completed_total{job=\"who-need-help-worker\"}[1m]))", + "legendFormat": "completed {{queue}}", + "range": true, + "refId": "A" + }, + { + "editorMode": "code", + "expr": "sum by (queue) (rate(who_need_help_oban_jobs_failed_total{job=\"who-need-help-worker\"}[1m]))", + "legendFormat": "failed {{queue}}", + "range": true, + "refId": "B" + } + ], + "title": "Oban job outcomes by queue", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "wnh-prometheus" + }, + "fieldConfig": { + "defaults": { + "unit": "ms" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 32 + }, + "id": 9, + "targets": [ + { + "editorMode": "code", + "expr": "sum by (queue) (rate(who_need_help_oban_job_duration_microseconds_total{job=\"who-need-help-worker\"}[5m])) / sum by (queue) (rate(who_need_help_oban_jobs_completed_total{job=\"who-need-help-worker\"}[5m])) / 1000", + "legendFormat": "execution {{queue}}", + "range": true, + "refId": "A" + }, + { + "editorMode": "code", + "expr": "sum by (queue) (rate(who_need_help_oban_job_queue_duration_microseconds_total{job=\"who-need-help-worker\"}[5m])) / sum by (queue) (rate(who_need_help_oban_jobs_completed_total{job=\"who-need-help-worker\"}[5m])) / 1000", + "legendFormat": "queue wait {{queue}}", + "range": true, + "refId": "B" + } + ], + "title": "Average Oban execution and queue wait", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "wnh-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 32 + }, + "id": 10, + "targets": [ + { + "editorMode": "code", + "expr": "who_need_help_vm_run_queue_total", + "legendFormat": "{{job}} {{instance}}", + "range": true, + "refId": "A" + } + ], + "title": "BEAM scheduler run queue by replica", + "type": "timeseries" } ], "refresh": "5s", diff --git a/ops/observability/prometheus.template.yml b/ops/observability/prometheus.template.yml index 243a379..767f047 100644 --- a/ops/observability/prometheus.template.yml +++ b/ops/observability/prometheus.template.yml @@ -25,3 +25,13 @@ scrape_configs: - files: - /runtime/web-targets.json refresh_interval: __SCRAPE_INTERVAL__ + + - job_name: who-need-help-worker + metrics_path: /metrics + authorization: + type: Bearer + credentials_file: /runtime/metrics-token + file_sd_configs: + - files: + - /runtime/worker-targets.json + refresh_interval: __SCRAPE_INTERVAL__ diff --git a/ops/observability/rules.yml b/ops/observability/rules.yml index 342e70d..15f63be 100644 --- a/ops/observability/rules.yml +++ b/ops/observability/rules.yml @@ -8,3 +8,10 @@ groups: annotations: summary: "The local Prometheus scraper cannot reach one web replica" description: "The direct metrics target {{ $labels.instance }} is unavailable." + - alert: WhoNeedHelpWorkerReplicaUnavailable + expr: up{job="who-need-help-worker"} == 0 + labels: + severity: local-drill + annotations: + summary: "The local Prometheus scraper cannot reach one worker replica" + description: "The direct worker metrics target {{ $labels.instance }} is unavailable." diff --git a/scripts/clean-deploy-verify.sh b/scripts/clean-deploy-verify.sh index e1fe757..a3d66d3 100755 --- a/scripts/clean-deploy-verify.sh +++ b/scripts/clean-deploy-verify.sh @@ -136,15 +136,17 @@ secret_key_base=$(openssl rand -hex 64) handover_secret=$(openssl rand -hex 64) release_cookie=$(openssl rand -hex 64) metrics_token=$(openssl rand -hex 32) +docker_socket_gid=$(stat -c '%g' /var/run/docker.sock) cat >"$env_file" <"$target" <"$temporary" if grep -Eq '^[A-Z0-9_]+=GENERATE_' "$temporary"; then diff --git a/scripts/load-run.sh b/scripts/load-run.sh index 9aa8a4e..9558e93 100755 --- a/scripts/load-run.sh +++ b/scripts/load-run.sh @@ -18,6 +18,9 @@ set -a . "$ENV_FILE" set +a +WEB_POOL_SIZE=${WEB_POOL_SIZE:-${POOL_SIZE:-}} +WORKER_POOL_SIZE=${WORKER_POOL_SIZE:-${POOL_SIZE:-}} + duration_source=".env.load" if [[ -n "$duration_override" ]]; then LOAD_DURATION=$duration_override @@ -28,7 +31,8 @@ for name in LOAD_PROJECT LOAD_HOST LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS \ LOAD_HTTP_VUS LOAD_WS_VUS \ LOAD_DURATION LOAD_WS_HOLD_MS LOAD_WS_CONNECT_TIMEOUT_MS \ LOAD_HTTP_THINK_SECONDS LOAD_AUTH_VUS LOAD_AUTH_WS_TIMEOUT_MS \ - LOAD_AUTH_THINK_SECONDS LOAD_FIXTURE_PASSWORD POSTGRES_DB HTTP_PORT; do + LOAD_AUTH_THINK_SECONDS LOAD_FIXTURE_PASSWORD POSTGRES_DB HTTP_PORT \ + WEB_POOL_SIZE WORKER_POOL_SIZE; do if [[ -z "${!name:-}" ]]; then echo "$name is missing from .env.load" >&2 exit 1 @@ -36,7 +40,7 @@ for name in LOAD_PROJECT LOAD_HOST LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS \ done for name in LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS LOAD_HTTP_VUS LOAD_WS_VUS \ - LOAD_AUTH_VUS; do + LOAD_AUTH_VUS WEB_POOL_SIZE WORKER_POOL_SIZE; do if [[ ! "${!name}" =~ ^[1-9][0-9]*$ ]]; then echo "$name must be a positive integer." >&2 exit 1 @@ -430,17 +434,23 @@ SQL summarize_database_connections() { jq -s \ - --argjson pool_size "$POOL_SIZE" \ - --argjson app_replicas "$((LOAD_WEB_REPLICAS + LOAD_WORKER_REPLICAS))" ' + --argjson web_pool_size "$WEB_POOL_SIZE" \ + --argjson worker_pool_size "$WORKER_POOL_SIZE" \ + --argjson web_replicas "$LOAD_WEB_REPLICAS" \ + --argjson worker_replicas "$LOAD_WORKER_REPLICAS" ' . as $samples | { schema_version: 1, measurement: "read-only pg_stat_activity samples during the k6 run", thresholds_applied: false, sample_count: length, - configured_app_replicas: $app_replicas, - configured_pool_size_per_replica: $pool_size, - configured_repo_pool_slots: ($app_replicas * $pool_size), + configured_web_replicas: $web_replicas, + configured_worker_replicas: $worker_replicas, + configured_web_pool_size_per_replica: $web_pool_size, + configured_worker_pool_size_per_replica: $worker_pool_size, + configured_repo_pool_slots: + (($web_replicas * $web_pool_size) + + ($worker_replicas * $worker_pool_size)), max_connections: (map(.max_connections) | max), reserved_connections: (map(.reserved_connections) | max), superuser_reserved_connections: diff --git a/scripts/observability-run.sh b/scripts/observability-run.sh index b21a877..6ccb8a0 100755 --- a/scripts/observability-run.sh +++ b/scripts/observability-run.sh @@ -15,7 +15,7 @@ set -a . "$ENV_FILE" set +a -for name in LOAD_PROJECT LOAD_WEB_REPLICAS POSTGRES_DB METRICS_TOKEN \ +for name in LOAD_PROJECT LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS POSTGRES_DB METRICS_TOKEN \ OBSERVABILITY_PROMETHEUS_PORT OBSERVABILITY_ALERTMANAGER_PORT \ OBSERVABILITY_GRAFANA_PORT OBSERVABILITY_SCRAPE_INTERVAL \ OBSERVABILITY_EVALUATION_INTERVAL OBSERVABILITY_TIMEOUT_SECONDS \ @@ -36,10 +36,12 @@ if [[ ! "$LABEL" =~ ^[A-Za-z0-9._-]+$ ]]; then exit 1 fi -if [[ ! "$LOAD_WEB_REPLICAS" =~ ^[1-9][0-9]*$ ]]; then - echo "LOAD_WEB_REPLICAS must be a positive integer." >&2 - exit 1 -fi +for name in LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS; do + if [[ ! "${!name}" =~ ^[1-9][0-9]*$ ]]; then + echo "$name must be a positive integer." >&2 + exit 1 + fi +done for name in OBSERVABILITY_PROMETHEUS_PORT OBSERVABILITY_ALERTMANAGER_PORT \ OBSERVABILITY_GRAFANA_PORT; do @@ -216,23 +218,36 @@ fetch_receiver_events() { } wait_for_targets_up() { - local expected=$1 + local expected_web=$1 + local expected_worker=$2 local deadline=$((SECONDS + OBSERVABILITY_TIMEOUT_SECONDS)) while ((SECONDS < deadline)); do if curl --fail --silent --show-error \ "$prometheus_url/api/v1/targets?state=active" \ >"$output_dir/targets-current.json" 2>/dev/null && - jq -e --argjson expected "$expected" --arg run_id "$LABEL" ' + jq -e \ + --argjson expected_web "$expected_web" \ + --argjson expected_worker "$expected_worker" \ + --arg run_id "$LABEL" ' [ .data.activeTargets[] | select( .labels.job == "who-need-help-web" and .labels.run_id == $run_id ) - ] as $targets - | ($targets | length) == $expected - and all($targets[]; .health == "up") + ] as $web_targets + | [ + .data.activeTargets[] + | select( + .labels.job == "who-need-help-worker" and + .labels.run_id == $run_id + ) + ] as $worker_targets + | ($web_targets | length) == $expected_web + and ($worker_targets | length) == $expected_worker + and all($web_targets[]; .health == "up") + and all($worker_targets[]; .health == "up") ' "$output_dir/targets-current.json" >/dev/null; then return 0 fi @@ -240,7 +255,7 @@ wait_for_targets_up() { sleep 1 done - echo "Timed out waiting for every direct web metrics target." >&2 + echo "Timed out waiting for every direct web and worker metrics target." >&2 return 1 } @@ -349,12 +364,18 @@ cleanup() { trap cleanup EXIT HUP INT TERM mapfile -t web_ids < <(service_ids web) +mapfile -t worker_ids < <(service_ids worker) if [[ "${#web_ids[@]}" -ne "$LOAD_WEB_REPLICAS" ]]; then echo "Expected $LOAD_WEB_REPLICAS running load web replicas; observed ${#web_ids[@]}." >&2 exit 1 fi +if [[ "${#worker_ids[@]}" -ne "$LOAD_WORKER_REPLICAS" ]]; then + echo "Expected $LOAD_WORKER_REPLICAS running load worker replicas; observed ${#worker_ids[@]}." >&2 + exit 1 +fi + internal_network=$( docker network ls \ --filter "label=com.docker.compose.project=$LOAD_PROJECT" \ @@ -404,6 +425,43 @@ done jq -s '.' "$target_lines" >"$prometheus_runtime/web-targets.json" unlink "$target_lines" +target_lines="$prometheus_runtime/worker-targets.jsonl" +: >"$target_lines" + +for worker_id in "${worker_ids[@]}"; do + assert_scope "$worker_id" worker + wait_for_web "$worker_id" + worker_name=$(docker inspect --format '{{.Name}}' "$worker_id") + worker_name=${worker_name#/} + worker_ip=$( + docker inspect \ + --format "{{with index .NetworkSettings.Networks \"$internal_network\"}}{{.IPAddress}}{{end}}" \ + "$worker_id" + ) + + if [[ ! "$worker_ip" =~ ^[0-9]+([.][0-9]+){3}$ ]]; then + echo "No observed IPv4 address for $worker_name on $internal_network." >&2 + exit 1 + fi + + jq -cn \ + --arg target "$worker_ip:4000" \ + --arg instance "$worker_name" \ + --arg compose_project "$LOAD_PROJECT" \ + --arg run_id "$LABEL" \ + '{ + targets: [$target], + labels: { + instance: $instance, + compose_project: $compose_project, + run_id: $run_id + } + }' >>"$target_lines" +done + +jq -s '.' "$target_lines" >"$prometheus_runtime/worker-targets.json" +unlink "$target_lines" + sed \ -e "s/__SCRAPE_INTERVAL__/$OBSERVABILITY_SCRAPE_INTERVAL/g" \ -e "s/__EVALUATION_INTERVAL__/$OBSERVABILITY_EVALUATION_INTERVAL/g" \ @@ -413,11 +471,14 @@ printf '%s' "$METRICS_TOKEN" >"$prometheus_runtime/metrics-token" printf '%s' "$OBSERVABILITY_GRAFANA_ADMIN_PASSWORD" \ >"$grafana_runtime/admin-password" cp "$prometheus_runtime/web-targets.json" "$output_dir/generated-targets.json" +cp "$prometheus_runtime/worker-targets.json" "$output_dir/generated-worker-targets.json" chmod 600 "$prometheus_runtime/prometheus.yml" \ "$prometheus_runtime/web-targets.json" \ + "$prometheus_runtime/worker-targets.json" \ "$prometheus_runtime/metrics-token" \ "$grafana_runtime/admin-password" \ - "$output_dir/generated-targets.json" + "$output_dir/generated-targets.json" \ + "$output_dir/generated-worker-targets.json" set_runtime_owner "$prometheus_runtime" "65534:65534" set_runtime_owner "$grafana_runtime" "472:0" @@ -482,33 +543,42 @@ curl --fail --silent --show-error \ if ! jq -e ' .dashboard.uid == "wnh-overview" and .meta.provisioned == true and - (.dashboard.panels | length) == 4 + (.dashboard.panels | length) == 10 ' "$output_dir/grafana-dashboard.json" >/dev/null; then echo "The provisioned Grafana dashboard did not match the tracked dashboard." >&2 exit 1 fi -wait_for_targets_up "$LOAD_WEB_REPLICAS" +wait_for_targets_up "$LOAD_WEB_REPLICAS" "$LOAD_WORKER_REPLICAS" mv "$output_dir/targets-current.json" "$output_dir/targets-before-drill.json" jq -n \ - --slurpfile expected "$output_dir/generated-targets.json" \ + --slurpfile expected_web "$output_dir/generated-targets.json" \ + --slurpfile expected_worker "$output_dir/generated-worker-targets.json" \ --slurpfile observed "$output_dir/targets-before-drill.json" ' { - expected_instances: ($expected[0] | map(.labels.instance) | sort), - observed_instances: ( + expected_web_instances: ($expected_web[0] | map(.labels.instance) | sort), + observed_web_instances: ( $observed[0].data.activeTargets | map(select(.labels.job == "who-need-help-web") | .labels.instance) | sort + ), + expected_worker_instances: ($expected_worker[0] | map(.labels.instance) | sort), + observed_worker_instances: ( + $observed[0].data.activeTargets + | map(select(.labels.job == "who-need-help-worker") | .labels.instance) + | sort ) } | . + { - exact_instance_match: (.expected_instances == .observed_instances) + exact_instance_match: + (.expected_web_instances == .observed_web_instances and + .expected_worker_instances == .observed_worker_instances) } ' >"$output_dir/target-summary.json" if ! jq -e '.exact_instance_match' "$output_dir/target-summary.json" >/dev/null; then - echo "Prometheus did not preserve every direct web instance target." >&2 + echo "Prometheus did not preserve every direct web and worker instance target." >&2 exit 1 fi @@ -524,7 +594,7 @@ wait_for_webhook_status \ docker start "$drill_web_id" >"$output_dir/started-web.txt" wait_for_web "$drill_web_id" -wait_for_targets_up "$LOAD_WEB_REPLICAS" +wait_for_targets_up "$LOAD_WEB_REPLICAS" "$LOAD_WORKER_REPLICAS" mv "$output_dir/targets-current.json" "$output_dir/targets-after-recovery.json" wait_for_alert_clear "$drill_instance" wait_for_webhook_status \ @@ -550,9 +620,11 @@ jq -n \ --arg alertmanager_url "$alertmanager_url" \ --arg grafana_url "$grafana_url" \ --argjson web_replicas "$LOAD_WEB_REPLICAS" \ + --argjson worker_replicas "$LOAD_WORKER_REPLICAS" \ '{ run_id: $run_id, web_replicas: $web_replicas, + worker_replicas: $worker_replicas, direct_targets_up_before_and_after: true, induced_instance: $drill_instance, firing_webhook_observed: true, diff --git a/scripts/quality.sh b/scripts/quality.sh index cec7b96..97a9da0 100755 --- a/scripts/quality.sh +++ b/scripts/quality.sh @@ -22,6 +22,9 @@ backup_image="who-need-help:backup-audit-$run_id" minio_image="who-need-help:minio-audit-$run_id" mc_image="who-need-help:mc-audit-$run_id" boundary_mock_image="who-need-help:boundary-mock-audit-$run_id" +socket_proxy_image="who-need-help:socket-proxy-audit-$run_id" +postgis_image="who-need-help:postgis-audit-$run_id" +socket_proxy_container="wnh-socket-proxy-audit-$run_id" scan_dir=$(mktemp -d "${TMPDIR:-/tmp}/wnh-quality-scan.XXXXXX") scan_list="${scan_dir}.files" scan_tar="${scan_dir}.tar" @@ -30,14 +33,17 @@ umask 077 QUALITY_POSTGRES_USER="wnh_quality_$(openssl rand -hex 6)" QUALITY_POSTGRES_PASSWORD=$(openssl rand -base64 48 | tr -d '\n') export QUALITY_POSTGRES_USER QUALITY_POSTGRES_PASSWORD +QUALITY_POSTGIS_IMAGE=$postgis_image +export QUALITY_POSTGIS_IMAGE compose="docker compose -p $project -f $ROOT/compose.quality.yaml" cleanup() { $compose down --volumes --remove-orphans >/dev/null 2>&1 || true + docker rm --force "$socket_proxy_container" >/dev/null 2>&1 || true docker image rm "$quality_image" "$assets_image" "$e2e_image" "$release_image" \ "$backup_image" "$minio_image" "$mc_image" \ - "$boundary_mock_image" \ + "$boundary_mock_image" "$socket_proxy_image" "$postgis_image" \ >/dev/null 2>&1 || true rm -rf "$scan_dir" "$scan_list" "$scan_tar" } @@ -56,6 +62,7 @@ docker run --rm \ echo "Checking Dockerfiles with Hadolint 2.14.0" for dockerfile in Dockerfile Dockerfile.backup Dockerfile.minio \ + Dockerfile.postgis Dockerfile.socket-proxy \ android/Dockerfile e2e/Dockerfile ops/external-boundaries/Dockerfile; do docker run --rm --interactive "$HADOLINT_IMAGE" \ hadolint --failure-threshold warning - <"$dockerfile" @@ -71,9 +78,42 @@ echo "Rendering every Docker Compose profile" docker compose --env-file .env.example -f compose.yaml config --quiet docker compose --env-file .env.example -f compose.yaml config --format json | jq --exit-status ' - [.services.migrate, .services.web, .services.worker] - | all(.environment.ERL_ZFLAGS == "+Q 65536") + . as $root + | [$root.services.migrate, $root.services.web, $root.services.worker] + | all( + .environment.ERL_ZFLAGS == "+Q 65536" and + .read_only == true and + .cap_drop == ["ALL"] and + .security_opt == ["no-new-privileges:true"] and + (.tmpfs | index("/tmp") != null) + ) + and $root.services.web.environment.POOL_SIZE == "4" + and $root.services.worker.environment.POOL_SIZE == "2" + and $root.services.migrate.environment.POOL_SIZE == "2" + and $root.services.web.deploy.replicas == 2 + and $root.services.worker.deploy.replicas == 2 + and ($root.services.proxy.networks | keys | sort) == ["docker-api", "edge", "ingress"] + and ($root.services.web.networks | keys | sort) == ["egress", "ingress", "internal"] + and ($root.services.worker.networks | keys | sort) == ["egress", "internal"] + and ($root.services.db.networks | keys) == ["internal"] + and $root.networks.ingress.internal == true + and $root.networks.internal.internal == true + and ($root.networks.egress.internal // false) == false + and $root.services.db.security_opt == ["no-new-privileges:true"] + and $root.services.mailpit.ports[0].host_ip == "127.0.0.1" ' >/dev/null +WEB_REPLICAS=1 WORKER_REPLICAS=1 \ + docker compose --env-file .env.example -f compose.yaml config --format json | + jq --exit-status ' + .services.web.deploy.replicas == 1 and + .services.worker.deploy.replicas == 1 + ' >/dev/null +CPU_REPLAY_WEB_CPUS=1 \ +CPU_REPLAY_WORKER_CPUS=1 \ +CPU_REPLAY_WEB_SCHEDULERS=1 \ +CPU_REPLAY_WORKER_SCHEDULERS=1 \ + docker compose --env-file .env.example \ + -f compose.yaml -f compose.cpu-replay.yaml config --quiet PORTABILITY_IMAGE=who-need-help:portability-render \ docker compose --env-file .env.example \ -f compose.yaml -f compose.portability.yaml config --quiet @@ -129,6 +169,8 @@ printf '%s' 'isolated-quality-metrics-token' \ >"$scan_dir/observability-runtime/prometheus/metrics-token" printf '%s\n' '[]' \ >"$scan_dir/observability-runtime/prometheus/web-targets.json" +printf '%s\n' '[]' \ + >"$scan_dir/observability-runtime/prometheus/worker-targets.json" docker run --rm \ --user 0:0 \ --volume "$scan_dir/observability-runtime/prometheus:/runtime:ro" \ @@ -149,7 +191,7 @@ docker run --rm \ "$PYTHON_IMAGE" python -c \ 'import py_compile; py_compile.compile("/src/mock_server.py", cfile="/tmp/mock_server.pyc", doraise=True)' jq --exit-status \ - 'type == "object" and .uid == "wnh-overview" and (.panels | length) == 4' \ + 'type == "object" and .uid == "wnh-overview" and (.panels | length) == 10' \ ops/observability/grafana/dashboards/who-need-help-overview.json \ >/dev/null @@ -183,6 +225,77 @@ docker run --rm \ --exit-code 1 \ /scan +echo "Building, smoke-testing, and scanning pinned runtime infrastructure images" +docker build --tag "$socket_proxy_image" --file Dockerfile.socket-proxy . +docker build --tag "$postgis_image" --file Dockerfile.postgis . +test "$(docker image inspect --format '{{.Config.User}}' "$socket_proxy_image")" = "haproxy" +test "$(docker image inspect --format '{{.Config.User}}' "$postgis_image")" = "postgres" +docker run --rm --entrypoint sh "$postgis_image" -euc ' + test ! -e /usr/local/bin/gosu + test "$(id -u)" = 70 +' + +docker run --detach \ + --name "$socket_proxy_container" \ + --read-only \ + --tmpfs /run:uid=99,gid=99,mode=0755 \ + --tmpfs /tmp \ + --cap-drop ALL \ + --group-add "$(stat -c '%g' /var/run/docker.sock)" \ + --security-opt no-new-privileges \ + --env CONTAINERS=1 \ + --env EVENTS=1 \ + --env INFO=1 \ + --env NETWORKS=1 \ + --env PING=1 \ + --env POST=0 \ + --env VERSION=1 \ + --volume /var/run/docker.sock:/var/run/docker.sock:ro \ + --publish 127.0.0.1::2375 \ + "$socket_proxy_image" >/dev/null +socket_proxy_port=$( + docker port "$socket_proxy_container" 2375/tcp | + sed -n 's/.*://p' | + head -n 1 +) +test -n "$socket_proxy_port" +socket_proxy_ready=false +for _attempt in $(seq 1 30); do + if curl --fail --silent --show-error \ + "http://127.0.0.1:$socket_proxy_port/_ping" >/dev/null; then + socket_proxy_ready=true + break + fi + sleep 1 +done +test "$socket_proxy_ready" = true +test "$( + curl --silent --output /dev/null --write-out '%{http_code}' \ + "http://127.0.0.1:$socket_proxy_port/containers/json" +)" = "200" +test "$( + curl --silent --output /dev/null --write-out '%{http_code}' \ + --request POST \ + "http://127.0.0.1:$socket_proxy_port/containers/create" +)" = "403" +docker rm --force "$socket_proxy_container" >/dev/null + +for image in \ + "$socket_proxy_image" \ + "$postgis_image" \ + "traefik:v3.7.8@sha256:4299bbed850421258fc5448c2e0e6ad350981d4d335a68de11b92448aedbefe5" \ + "axllent/mailpit:v1.30.4@sha256:5a49a77c5bdbe7c5474450b4f46348d09949df3695257729c93a30369382d4f6"; do + docker run --rm \ + --volume /var/run/docker.sock:/var/run/docker.sock \ + --volume "$ROOT/.tools/trivy-cache:/root/.cache/trivy" \ + "$TRIVY_IMAGE" image \ + --scanners vuln \ + --severity HIGH,CRITICAL \ + --ignore-unfixed \ + --exit-code 1 \ + "$image" +done + echo "Building the pinned quality image and cached Dialyzer PLTs" docker build --target quality --tag "$quality_image" . diff --git a/scripts/upgrade-rehearsal-compose.sh b/scripts/upgrade-rehearsal-compose.sh index 69ed0b3..8941282 100755 --- a/scripts/upgrade-rehearsal-compose.sh +++ b/scripts/upgrade-rehearsal-compose.sh @@ -106,6 +106,8 @@ fi export DATABASE_URL="$database_prefix/$rehearsal_db$database_query" export HTTP_PORT=0 export MAILPIT_PORT=0 +export DOCKER_SOCKET_GID +DOCKER_SOCKET_GID=$(stat -c '%g' /var/run/docker.sock) export PHX_HOST="$rehearsal_phx_host" export PHX_SCHEME="$rehearsal_phx_scheme" export PHX_URL_PORT="$rehearsal_phx_url_port" @@ -113,7 +115,7 @@ export MAP_TILE_URL="$rehearsal_map_tile_url" export TRAEFIK_TRUSTED_IPS=127.0.0.1/32 export TRAEFIK_PROJECT_CONSTRAINT="$project" export TRAEFIK_APP_NAME="wnh-upgrade-rehearsal-$compact_id" -export TRAEFIK_DOCKER_NETWORK="${project}_internal" +export TRAEFIK_DOCKER_NETWORK="${project}_ingress" export REHEARSAL_IMAGE="who-need-help:upgrade-rehearsal-$run_id" compose() { diff --git a/test/who_need_help/accounts_test.exs b/test/who_need_help/accounts_test.exs index 4673684..f74e988 100644 --- a/test/who_need_help/accounts_test.exs +++ b/test/who_need_help/accounts_test.exs @@ -48,6 +48,29 @@ defmodule WhoNeedHelp.AccountsTest do end end + describe "public_user_query/1" do + test "does not load authentication or moderation fields" do + user = + user_fixture() + |> set_password() + |> Ecto.Changeset.change(moderation_note: "private moderator note") + |> Repo.update!() + + public_user = + Accounts.public_user_query() + |> where([candidate], candidate.id == ^user.id) + |> Repo.one!() + + assert public_user.id == user.id + assert public_user.display_name == user.display_name + assert is_nil(public_user.email) + assert is_nil(public_user.hashed_password) + assert is_nil(public_user.confirmed_at) + assert is_nil(public_user.accepted_terms_at) + assert is_nil(public_user.moderation_note) + end + end + describe "register_user/1" do test "requires email to be set" do {:error, changeset} = Accounts.register_user(%{}) diff --git a/test/who_need_help/activities_test.exs b/test/who_need_help/activities_test.exs index 32d9747..e0007e3 100644 --- a/test/who_need_help/activities_test.exs +++ b/test/who_need_help/activities_test.exs @@ -196,6 +196,11 @@ defmodule WhoNeedHelp.ActivitiesTest do assert length(Enum.uniq(actual_ids)) == 4 assert is_binary(first.next_cursor) assert second.next_cursor == nil + assert Enum.all?(first.entries ++ second.entries, &(&1.approved_participant_count == 1)) + + assert Enum.all?(first.entries ++ second.entries, fn activity -> + match?(%Ecto.Association.NotLoaded{}, activity.participants) + end) end test "my activities are derived from the organizer and participant membership rows", context do diff --git a/test/who_need_help/mutual_aid_flow_test.exs b/test/who_need_help/mutual_aid_flow_test.exs index 5f9e754..10f76b2 100644 --- a/test/who_need_help/mutual_aid_flow_test.exs +++ b/test/who_need_help/mutual_aid_flow_test.exs @@ -6,6 +6,7 @@ defmodule WhoNeedHelp.MutualAidFlowTest do alias WhoNeedHelp.{Accounts, Catalog, CatalogModeration, Help, Messaging, Tracking, Trust} alias WhoNeedHelp.Repo alias WhoNeedHelp.Tracking.Position + alias WhoNeedHelpWeb.Presence setup do category = Catalog.seed_defaults() @@ -56,6 +57,20 @@ defmodule WhoNeedHelp.MutualAidFlowTest do assert reputation.completed == 1 assert reputation.unique_people == 1 assert reputation.verified_handovers == 1 + + reputations = Trust.reputations([context.helper.id, context.requester.id]) + assert reputations[context.helper.id] == reputation + assert reputations[context.requester.id].completed == 1 + + unused_user = user_fixture(display_name: "No completed help") + + assert Trust.reputations([unused_user.id])[unused_user.id] == %{ + completed: 0, + unique_people: 0, + verified_handovers: 0, + location_supported: 0, + rating: nil + } end test "chat is durable and only visible to match participants", context do @@ -199,6 +214,47 @@ defmodule WhoNeedHelp.MutualAidFlowTest do refute Repo.get(Position, position.id) end + test "concurrent tracking stops serialize and broadcast once", context do + {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs) + {:ok, assignment} = Help.accept_request(context.helper_scope, request.id) + {:ok, _session} = Tracking.start_session(context.helper_scope, assignment) + :ok = Tracking.subscribe(assignment.id) + + results = + 1..2 + |> Task.async_stream( + fn _ -> Tracking.stop_session(context.helper_scope, assignment) end, + max_concurrency: 2, + ordered: false + ) + |> Enum.map(fn {:ok, result} -> result end) + |> Enum.sort() + + assert results == [{:ok, :already_stopped}, {:ok, :stopped}] + assert_receive {:tracking_stopped, helper_id}, 1_000 + assert helper_id == context.helper.id + refute_receive {:tracking_stopped, ^helper_id}, 100 + end + + test "stale browser cleanup cannot stop a replacement tracking session", context do + {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs) + {:ok, assignment} = Help.accept_request(context.helper_scope, request.id) + {:ok, old_session} = Tracking.start_session(context.helper_scope, assignment) + assert {:ok, :stopped} = Tracking.stop_session(context.helper_scope, assignment) + + {:ok, current_session} = Tracking.start_session(context.helper_scope, assignment) + refute current_session.id == old_session.id + + assert {:ok, :already_stopped} = + Tracking.stop_browser_session( + assignment.id, + context.helper.id, + old_session.id + ) + + assert Tracking.active_session?(context.helper_scope, assignment) + end + test "reviews remain hidden until both participants submit", context do {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs) {:ok, assignment} = Help.accept_request(context.helper_scope, request.id) @@ -239,6 +295,15 @@ defmodule WhoNeedHelp.MutualAidFlowTest do assert proposal_export["mode"] == "help" assert proposal_export["community_votes"] == 0 + [listed] = Catalog.list_proposals() + assert listed.id == proposal.id + assert listed.vote_count == 0 + assert %Ecto.Association.NotLoaded{} = listed.votes + + assert {:ok, _vote} = Catalog.vote(context.helper_scope, proposal.id) + assert [%{id: listed_id, vote_count: 1}] = Catalog.list_proposals() + assert listed_id == proposal.id + refute export =~ context.requester.email refute export =~ context.requester.display_name end @@ -331,4 +396,131 @@ defmodule WhoNeedHelp.MutualAidFlowTest do assert {:error, :not_found} = Accounts.delete_social_identity(context.helper, identity.id) assert {:ok, _identity} = Accounts.delete_social_identity(context.requester, identity.id) end + + test "request details batch public users and public social identities", context do + {:ok, requester_identity} = + Accounts.upsert_verified_social_identity(context.requester, %{ + provider: :github, + provider_uid: "requester-private-provider-id", + profile_url: "https://github.com/requester", + handle: "requester", + verified_at: DateTime.utc_now(:second) + }) + + {:ok, helper_identity} = + Accounts.upsert_verified_social_identity(context.helper, %{ + provider: :github, + provider_uid: "helper-private-provider-id", + profile_url: "https://github.com/helper", + handle: "helper", + verified_at: DateTime.utc_now(:second) + }) + + {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs) + {:ok, _assignment} = Help.accept_request(context.helper_scope, request.id) + + handler_id = "request-detail-queries-#{System.unique_integer([:positive])}" + + :ok = + :telemetry.attach( + handler_id, + [:who_need_help, :repo, :query], + fn _event, _measurements, metadata, receiver -> + send(receiver, {:request_detail_query, metadata.query}) + end, + self() + ) + + loaded = + try do + {:ok, loaded} = Help.get_request(context.requester_scope, request.id) + loaded + after + :telemetry.detach(handler_id) + end + + queries = collect_request_detail_queries() + + assert Enum.count(queries, &String.contains?(&1, ~s(FROM "users"))) == 1 + assert Enum.count(queries, &String.contains?(&1, ~s(FROM "social_identities"))) == 1 + + assert [loaded_requester_identity] = loaded.requester.social_identities + assert loaded_requester_identity.id == requester_identity.id + assert is_nil(loaded_requester_identity.provider_uid) + + assert [loaded_helper_identity] = loaded.assignment.helper.social_identities + assert loaded_helper_identity.id == helper_identity.id + assert is_nil(loaded_helper_identity.provider_uid) + end + + test "browser tracking stops only after the last tab leaves", context do + {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs) + {:ok, assignment} = Help.accept_request(context.helper_scope, request.id) + {:ok, session} = Tracking.start_session(context.helper_scope, assignment) + :ok = Tracking.subscribe(assignment.id) + + parent = self() + + start_browser = fn -> + spawn(fn -> + {:ok, key} = + Presence.track_browser(self(), assignment.id, context.helper.id, session.id) + + send(parent, {:browser_ready, self(), key}) + + receive do + :close -> :ok + end + end) + end + + first = start_browser.() + second = start_browser.() + key = Presence.tracking_key(assignment.id, context.helper.id) + + assert_receive {:browser_ready, ^first, ^key} + assert_receive {:browser_ready, ^second, ^key} + assert wait_for_presence_count(key, 2) + + first_ref = Process.monitor(first) + send(first, :close) + assert_receive {:DOWN, ^first_ref, :process, ^first, :normal} + assert wait_for_presence_count(key, 1) + assert Tracking.active_session?(context.helper_scope, assignment) + + second_ref = Process.monitor(second) + send(second, :close) + assert_receive {:DOWN, ^second_ref, :process, ^second, :normal} + assert_receive {:tracking_stopped, helper_id}, 1_000 + assert helper_id == context.helper.id + refute Tracking.active_session?(context.helper_scope, assignment) + end + + defp wait_for_presence_count(key, expected, attempts \\ 100) + + defp wait_for_presence_count(_key, _expected, 0), do: false + + defp wait_for_presence_count(key, expected, attempts) do + count = + case Presence.get_by_key(Presence.tracking_topic(), key) do + [] -> 0 + %{metas: metas} -> length(metas) + end + + if count == expected do + true + else + Process.sleep(10) + wait_for_presence_count(key, expected, attempts - 1) + end + end + + defp collect_request_detail_queries(queries \\ []) do + receive do + {:request_detail_query, query} -> + collect_request_detail_queries([query | queries]) + after + 0 -> Enum.reverse(queries) + end + end end diff --git a/test/who_need_help/workers/expire_requests_test.exs b/test/who_need_help/workers/expire_requests_test.exs new file mode 100644 index 0000000..91c238a --- /dev/null +++ b/test/who_need_help/workers/expire_requests_test.exs @@ -0,0 +1,65 @@ +defmodule WhoNeedHelp.Workers.ExpireRequestsTest do + use WhoNeedHelp.DataCase, async: false + + import WhoNeedHelp.AccountsFixtures + + alias WhoNeedHelp.{Catalog, Help, Repo} + alias WhoNeedHelp.Help.HelpRequest + alias WhoNeedHelp.Trust.AuditEvent + alias WhoNeedHelp.Workers.ExpireRequests + + test "expires an arbitrary backlog without loading it as one result set" do + category = Catalog.seed_defaults() + requester = user_fixture(display_name: "Expiration requester") + scope = user_scope_fixture(requester) + + request_ids = + for number <- 1..12 do + {:ok, request} = + Help.create_request(scope, %{ + "title" => "Expiring medicine request #{number}", + "description" => "The reserved legal medicine was not collected in time.", + "pickup_instructions" => "No payment or prescription handling is required.", + "location_label" => "Central district", + "latitude" => "50.4501", + "longitude" => "30.5234", + "urgency" => "today", + "location_visibility" => "approximate_public", + "structured_data" => %{"pickup_status" => "reserved"}, + "expires_at" => DateTime.utc_now(:second) |> DateTime.add(3, :hour), + "category_id" => category.id + }) + + request.id + end + + past = DateTime.utc_now(:second) |> DateTime.add(-1, :minute) + + Repo.update_all( + from(request in HelpRequest, where: request.id in ^request_ids), + set: [expires_at: past] + ) + + assert {:ok, %{expired: 12}} = ExpireRequests.perform(%Oban.Job{}) + + assert Repo.aggregate( + from( + request in HelpRequest, + where: request.id in ^request_ids and request.status == :expired + ), + :count + ) == 12 + + assert Repo.aggregate( + from( + event in AuditEvent, + where: + event.action == "request.expired" and + event.target_id in ^request_ids + ), + :count + ) == 12 + + assert {:ok, %{expired: 0}} = ExpireRequests.perform(%Oban.Job{}) + end +end diff --git a/test/who_need_help_web/worker_metrics_plug_test.exs b/test/who_need_help_web/worker_metrics_plug_test.exs new file mode 100644 index 0000000..1f044a5 --- /dev/null +++ b/test/who_need_help_web/worker_metrics_plug_test.exs @@ -0,0 +1,37 @@ +defmodule WhoNeedHelpWeb.WorkerMetricsPlugTest do + use WhoNeedHelp.DataCase, async: false + + import Plug.Conn + import Plug.Test + + alias WhoNeedHelpWeb.WorkerMetricsPlug + + test "liveness and readiness are available without exposing metrics" do + live = WorkerMetricsPlug.call(conn(:get, "/healthz/live"), []) + ready = WorkerMetricsPlug.call(conn(:get, "/healthz/ready"), []) + + assert live.status == 200 + assert ready.status == 200 + assert Jason.decode!(live.resp_body)["role"] == "worker" + assert Jason.decode!(ready.resp_body)["status"] == "ready" + end + + test "metrics require the independent bearer token" do + unauthorized = WorkerMetricsPlug.call(conn(:get, "/metrics"), []) + + authorized = + conn(:get, "/metrics") + |> put_req_header("authorization", "Bearer test-metrics-token") + |> WorkerMetricsPlug.call([]) + + assert unauthorized.status == 401 + assert get_resp_header(unauthorized, "www-authenticate") == ["Bearer"] + assert authorized.status == 200 + assert authorized.resp_body =~ "who_need_help_database_query_duration_seconds" + end + + test "all other methods and paths are unavailable" do + assert WorkerMetricsPlug.call(conn(:post, "/metrics"), []).status == 404 + assert WorkerMetricsPlug.call(conn(:get, "/"), []).status == 404 + end +end