diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9bc2b1b..6c08395 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,6 +1,7 @@ name: CI on: + workflow_call: push: branches: [main] pull_request: @@ -15,68 +16,61 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v6 + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 - - uses: cachix/install-nix-action@v31 + - uses: cachix/install-nix-action@a49548c11d9846ad46ecc0115273879b045f001c # v31 - - uses: erlef/setup-beam@v1 - id: beam + - name: Restore dependencies cache + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: - elixir-version: 1.20.2 - otp-version: 29.0.3 + path: deps + key: ${{ runner.os }}-mix-deps-${{ hashFiles('mix.lock') }} - - name: Restore dependencies cache - uses: actions/cache@v5 + - name: Restore build cache + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: - path: | - deps - _build - key: ${{ runner.os }}-mix-${{ hashFiles('**/mix.lock') }} - restore-keys: ${{ runner.os }}-mix- + path: _build + key: ${{ runner.os }}-mix-build-${{ hashFiles('flake.lock', 'mix.lock') }} - name: Restore PLT cache - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: priv/plts - key: ${{ runner.os }}-plt-otp${{ steps.beam.outputs.otp-version }}-${{ hashFiles('**/mix.lock') }}-${{ github.sha }} - restore-keys: | - ${{ runner.os }}-plt-otp${{ steps.beam.outputs.otp-version }}-${{ hashFiles('**/mix.lock') }}- - ${{ runner.os }}-plt-otp${{ steps.beam.outputs.otp-version }}- + key: ${{ runner.os }}-plt-${{ hashFiles('flake.lock', 'mix.lock') }} - name: Install dependencies - run: mix deps.get + run: nix develop --command mix deps.get --check-locked - name: Compile - run: mix compile --warnings-as-errors + run: nix develop --command mix compile --warnings-as-errors - name: Check formatting - run: mix format --check-formatted + run: nix develop --command mix format --check-formatted - name: Check unused dependencies - run: mix deps.unlock --check-unused + run: nix develop --command mix deps.unlock --check-unused - name: Check dependency cycles - run: mix xref graph --format cycles --fail-above 0 + run: nix develop --command mix xref graph --format cycles --fail-above 0 - name: Run unit tests without soffice - run: mix test + run: nix develop --command mix test --exclude integration + env: + URP_INTEGRATION: "0" - name: Build documentation - run: mix docs --warnings-as-errors + run: nix develop --command mix docs --warnings-as-errors - name: Audit Hex dependencies - run: mix hex.audit - - - name: Check Nix flake - run: nix flake check + run: nix develop --command mix hex.audit - name: Run dialyzer - run: mix dialyzer + run: nix develop --command mix dialyzer - - uses: docker/setup-buildx-action@v4 + - uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4 - name: Build soffice image - uses: docker/build-push-action@v7 + uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7 with: context: benchmarks file: benchmarks/Dockerfile.soffice-debian @@ -88,10 +82,12 @@ jobs: - name: Start soffice run: | docker run --detach --name soffice \ - --publish 2002:2002 \ + --publish 127.0.0.1:2002:2002 \ --volume /tmp:/tmp \ soffice timeout 30 bash -c 'until nc -z localhost 2002; do sleep 1; done' - name: Run integration tests - run: mix test --include lo26 + run: nix develop --command mix test --include integration --include lo26 + env: + URP_INTEGRATION: "1" diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 05b92f5..3b59298 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,35 +6,43 @@ on: - "v*" permissions: - contents: write + contents: read jobs: + validate: + name: Validate release commit + uses: ./.github/workflows/ci.yml + publish: name: Publish to Hex + needs: validate runs-on: ubuntu-latest + permissions: + contents: write steps: - - uses: actions/checkout@v6 + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 - - uses: erlef/setup-beam@v1 - with: - elixir-version: "1.20.2" - otp-version: "29.0.3" + - uses: cachix/install-nix-action@a49548c11d9846ad46ecc0115273879b045f001c # v31 - name: Restore dependencies cache - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: - path: | - deps - _build - key: ${{ runner.os }}-mix-${{ hashFiles('**/mix.lock') }} - restore-keys: ${{ runner.os }}-mix- + path: deps + key: ${{ runner.os }}-mix-deps-${{ hashFiles('mix.lock') }} - name: Install dependencies - run: mix deps.get + run: nix develop --command mix deps.get --check-locked + + - name: Verify tag, package version, and changelog + run: | + version="$(tr -d '[:space:]' < VERSION)" + test "${GITHUB_REF_NAME}" = "v${version}" + grep --fixed-strings --quiet "## [v${version}]" CHANGELOG.md + nix develop --command mix hex.build - name: Publish to Hex - run: mix hex.publish --yes + run: nix develop --command mix hex.publish --yes env: HEX_API_KEY: ${{ secrets.HEX_API_KEY }} diff --git a/CHANGELOG.md b/CHANGELOG.md index fe3a726..9ddebb0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,31 @@ # Changelog +## [Unreleased] + +### Changed + +- Emit complete telemetry spans with start, stop, and exception events. +- Use the Nix flake as the Elixir/OTP source of truth in CI and release validation. +- Make LibreOffice integration tests an explicit opt-in instead of probing an ambient port. +- Clarify that multiple connections to one soffice process are supported but share state + and generally do not provide useful conversion parallelism. + +### Fixed + +- Return errors from non-bang APIs when pool checkout fails and discard broken diagnostic workers. +- Scope URP TID and OID caches to individual connections and decode cached interface references. +- Preserve streamed input resources through export and bound enumerable input with demand-driven flow. +- Propagate stream export and output-sink failures instead of returning partial success. +- Run best-effort document and remote-temp cleanup after failed conversions. +- Restore per-call timeout and frame-size overrides before returning workers to the pool. +- Validate public conversion options before dispatching test stubs or protocol work. +- Harden release validation and make benchmarks reject failed or non-PDF results. + +### Security + +- Use unpredictable temporary names and run the benchmark LibreOffice image as a non-root user. +- Bind documented benchmark service ports to loopback by default. + ## [v0.10.1] - 2026-07-14 ### Changed diff --git a/PERFORMANCE.md b/PERFORMANCE.md index d238948..1fe51dc 100644 --- a/PERFORMANCE.md +++ b/PERFORMANCE.md @@ -8,7 +8,7 @@ a popular LibreOffice-based conversion service. ```sh docker compose --file benchmarks/docker-compose.yml up --detach --wait -mix run benchmarks/bench.exs +nix develop --command mix run benchmarks/bench.exs ``` The results below were recorded on July 14, 2026, on Apple M3 Max with @@ -24,39 +24,44 @@ The fixture uses Liberation fonts only — regenerate with ``` Name ips average deviation median 99th % -URP 1.37 0.73 s ±4.74% 0.72 s 0.82 s -Gotenberg 0.84 1.19 s ±5.46% 1.16 s 1.37 s +URP → Debian glibc 1.31 0.76 s ±14.66% 0.72 s 1.09 s +URP → Alpine musl 0.88 1.14 s ±8.36% 1.16 s 1.21 s +Gotenberg (HTTP) 0.85 1.18 s ±5.55% 1.16 s 1.36 s ``` **15.5 MB input → 62 MB PDF:** ``` Name ips average deviation median 99th % -URP 0.196 5.10 s ±2.12% 5.05 s 5.22 s -Gotenberg 0.135 7.39 s ±0.88% 7.36 s 7.46 s +URP → Debian glibc 0.194 5.16 s ±1.86% 5.18 s 5.24 s +URP → Alpine musl 0.143 6.98 s ±13.31% 6.74 s 8.01 s +Gotenberg (HTTP) 0.137 7.30 s ±2.32% 7.23 s 7.49 s ``` -URP had **39% lower average latency** for the small document and **31% -lower average latency** for the large document. The absolute advantage -grew from 0.46 s to 2.29 s. These measurements compare the complete -stacks, including their slightly different LibreOffice patch versions; -they do not isolate dependency or runtime upgrades individually. +The Debian URP stack had **36% lower average latency** for the small +document and **29% lower average latency** for the large document than +Gotenberg. The absolute advantage grew from 0.42 s to 2.14 s. Alpine URP +was much closer to Gotenberg and had noticeably higher variance on the +large fixture. These measurements compare complete stacks with different +LibreOffice builds, C libraries, and container packaging; they do not +isolate dependency or runtime upgrades individually. ### Process overhead sanity check `benchmarks/convert.exs` compares the persistent URP connection with a cold `soffice --convert-to` process and Gotenberg using the 33 KB -`sample3.docx` fixture. Across five timed iterations, the averages were: +`sample3.docx` fixture. Across ten timed iterations, the results were: -| Method | Average | Range | -|--------|---------|-------| -| URP | 46 ms | 43–49 ms | -| Gotenberg | 153 ms | 146–160 ms | -| LibreOffice CLI | 280 ms | 267–302 ms | +| Method | Average | Median | Range | +|--------|---------|--------|-------| +| URP | 45 ms | 45 ms | 42–49 ms | +| Gotenberg | 174 ms | 154 ms | 147–323 ms | +| LibreOffice CLI | 285 ms | 286 ms | 276–302 ms | This is a process-overhead check, not an apples-to-apples transport benchmark: URP reuses a live office process, while the CLI measurement -starts a new process for every conversion. +starts a new process for every conversion. One Gotenberg request was a +323 ms outlier; the median remained close to the previous run. ## I/O strategies @@ -64,7 +69,7 @@ URP supports two I/O transfer strategies via the `:io` option, benchmarked with `benchmarks/io_bench.exs`: ```sh -mix run benchmarks/io_bench.exs +nix develop --command mix run benchmarks/io_bench.exs ``` **File I/O** (`:file`, default) writes temp files on soffice's filesystem diff --git a/README.md b/README.md index c60f2bf..8d73e10 100644 --- a/README.md +++ b/README.md @@ -56,9 +56,13 @@ config :urp, :default, > #### pool_size {: .warning} > -> Each connection needs its own soffice instance. With `pool_size: 3`, -> run 3 soffice containers — one per connection. Concurrent operations -> on a single soffice process are not safe. +> A single URP connection handles one operation at a time. LibreOffice accepts +> multiple connections to one soffice process, but they share process-wide +> state and generally do not improve conversion throughput. Keep `pool_size: 1` +> unless you have tested your workload. For predictable parallelism and fault +> isolation, use separate soffice processes with distinct profiles. The current +> pool sends every worker to the configured host and port; distributing workers +> across containers requires separate named pools or an external TCP balancer. ### Testing @@ -69,18 +73,20 @@ URP.Test.stub(fn _input, _opts -> {:ok, "/tmp/fake.pdf"} end) assert {:ok, _} = MyApp.generate_invoice(order) ``` -When soffice is unavailable, tests tagged `:integration` are excluded automatically. -Run the complete suite, including the LibreOffice 26.2+ coverage, with: +`mix test` always runs the deterministic unit suite without probing local ports. +Run the complete suite, including the LibreOffice 26.2+ coverage, explicitly with: ```sh docker compose --file benchmarks/docker-compose.yml up --detach --wait soffice -mix test --include lo26 +URP_INTEGRATION=1 nix develop --command mix test --include lo26 ``` ### Telemetry -Every operation emits `[:urp, :call, :stop]` with queue, service, and -total time measurements. See `URP.Telemetry`. +Every operation emits a `[:urp, :call, :start]` event followed by either +`[:urp, :call, :stop]` or `[:urp, :call, :exception]`. Stop events include +queue, service, backoff, and total time. Connection retries emit +`[:urp, :connection, :retry]`. See `URP.Telemetry`. ## Performance @@ -96,7 +102,9 @@ See [PERFORMANCE.md](PERFORMANCE.md) for benchmarks and container image recommen ```sh ./release.sh patch # or minor, major -git push origin main --tags +git push origin main +# Wait for main CI, then: +git push origin "v$(cat VERSION)" ``` ## License diff --git a/benchmarks/Dockerfile.soffice-debian b/benchmarks/Dockerfile.soffice-debian index dbb6d18..7192758 100644 --- a/benchmarks/Dockerfile.soffice-debian +++ b/benchmarks/Dockerfile.soffice-debian @@ -8,12 +8,21 @@ RUN export DEBIAN_FRONTEND=noninteractive && \ apt-get install -y --no-install-recommends -t trixie-backports \ libreoffice-core-nogui libreoffice-writer-nogui libreoffice-calc-nogui libreoffice-impress-nogui && \ apt-get install -y --no-install-recommends \ - fonts-liberation fonts-crosextra-carlito && \ + adduser fonts-liberation fonts-crosextra-carlito netcat-openbsd && \ + adduser --uid 10001 --disabled-password --gecos "" soffice && \ + mkdir -p /tmp/soffice-profile && \ + chown soffice:soffice /tmp/soffice-profile && \ apt-get clean && rm -rf /var/lib/apt/lists/* ENV HOST=0.0.0.0 ENV PORT=2002 +ENV HOME=/home/soffice EXPOSE ${PORT} -CMD ["sh", "-c", "exec soffice --headless --norestore --accept=\"socket,host=${HOST},port=${PORT},tcpNoDelay=1;urp;\""] +USER soffice + +HEALTHCHECK --start-period=10s --interval=2s --timeout=1s --retries=15 \ + CMD nc -z 127.0.0.1 "$PORT" || exit 1 + +CMD ["sh", "-c", "exec soffice --headless --norestore -env:UserInstallation=file:///tmp/soffice-profile --accept=\"socket,host=${HOST},port=${PORT},tcpNoDelay=1;urp;\""] diff --git a/benchmarks/convert.exs b/benchmarks/convert.exs index b9f3bbb..0d6ab86 100644 --- a/benchmarks/convert.exs +++ b/benchmarks/convert.exs @@ -4,7 +4,7 @@ # # Prerequisites: # 1. soffice container on port 2002 (URP) -# 2. soffice-cli container with /fixtures mount (CLI — separate instance to avoid lock) +# 2. soffice-cli Compose service (CLI — separate instance to avoid profile locking) # 3. Gotenberg service from docker-compose.yml on GOTENBERG_PORT (HTTP API) # # docker compose --file benchmarks/docker-compose.yml up --detach --wait @@ -14,20 +14,50 @@ # # Environment variables: # GOTENBERG_PORT - Gotenberg HTTP port (default: 3002) -# SOFFICE_CLI - Container name for CLI (default: soffice-cli) -# FIXTURE - docx file in benchmarks/fixtures/ (default: sample3.docx) -# ITERATIONS - Number of timed runs per method (default: 5) +# SOFFICE_CLI - Compose service name for CLI (default: soffice-cli) +# FIXTURE - fixture in benchmarks/fixtures or test/fixtures (default: sample3.docx) +# ITERATIONS - Number of timed runs per method (default: 10) gotenberg_port = System.get_env("GOTENBERG_PORT", "3002") soffice_cli = System.get_env("SOFFICE_CLI", "soffice-cli") fixture = System.get_env("FIXTURE", "sample3.docx") -iterations = String.to_integer(System.get_env("ITERATIONS", "5")) +iterations = String.to_integer(System.get_env("ITERATIONS", "10")) +compose_file = Path.expand("benchmarks/docker-compose.yml") + +{docx_path, container_fixture} = + [ + {Path.expand("benchmarks/fixtures/#{fixture}"), "/benchmark-fixtures/#{fixture}"}, + {Path.expand("test/fixtures/#{fixture}"), "/test-fixtures/#{fixture}"} + ] + |> Enum.find(fn {path, _container_path} -> File.exists?(path) end) || + raise "fixture #{fixture} not found in benchmarks/fixtures or test/fixtures" -docx_path = Path.expand("benchmarks/fixtures/#{fixture}") docx_bytes = File.read!(docx_path) gotenberg_url = "http://localhost:#{gotenberg_port}/forms/libreoffice/convert" pdf_stem = Path.rootname(fixture) +run! = fn command, args -> + {output, status} = System.cmd(command, args, stderr_to_stdout: true) + + if status != 0 do + raise "command failed (#{status}): #{command} #{Enum.join(args, " ")}\n#{output}" + end + + output +end + +compose! = fn args -> + run!.("docker", ["compose", "--file", compose_file] ++ args) +end + +assert_pdf! = fn label, bytes -> + unless match?(<<"%PDF-", _::binary>>, bytes) do + raise "#{label} did not produce a PDF (#{byte_size(bytes)} bytes)" + end + + bytes +end + IO.puts("Fixture: #{fixture} (#{div(byte_size(docx_bytes), 1024)} KB)") IO.puts("Iterations: #{iterations}") IO.puts("CLI: #{soffice_cli}") @@ -57,31 +87,61 @@ end # --- URP --- bench.("URP", iterations, fn -> - {:ok, _pdf} = + {:ok, pdf} = URP.convert({:binary, docx_bytes}, filter: "writer_pdf_Export", output: :binary) + + assert_pdf!.("URP", pdf) end) # --- CLI --- bench.("CLI", iterations, fn -> - {_, 0} = - System.cmd("docker", [ - "exec", soffice_cli, "soffice", - "--headless", "--convert-to", "pdf", - "--outdir", "/tmp", - "/fixtures/#{fixture}" + script = """ + set -eu + output_dir=$(mktemp -d /tmp/urp-cli.XXXXXX) + trap 'rm -rf "$output_dir"' EXIT + + if ! soffice --headless --convert-to pdf --outdir "$output_dir" "$2" \ + >"$output_dir/convert.log" 2>&1; then + cat "$output_dir/convert.log" >&2 + exit 1 + fi + + cat "$output_dir/$1.pdf" + """ + + pdf = + compose!.([ + "exec", + "-T", + soffice_cli, + "sh", + "-c", + script, + "urp-cli", + pdf_stem, + container_fixture ]) - System.cmd("docker", ["exec", soffice_cli, "rm", "-f", "/tmp/#{pdf_stem}.pdf"]) + assert_pdf!.("LibreOffice CLI", pdf) end) # --- Gotenberg --- bench.("Gotenberg", iterations, fn -> - {_pdf, 0} = - System.cmd("curl", [ - "--silent", "--request", "POST", gotenberg_url, - "--form", "files=@#{docx_path}", - "--output", "-" + pdf = + run!.("curl", [ + "--silent", + "--show-error", + "--fail-with-body", + "--request", + "POST", + gotenberg_url, + "--form", + "files=@#{docx_path}", + "--output", + "-" ]) + + assert_pdf!.("Gotenberg", pdf) end) diff --git a/benchmarks/docker-compose.yml b/benchmarks/docker-compose.yml index 03722df..2241cd0 100644 --- a/benchmarks/docker-compose.yml +++ b/benchmarks/docker-compose.yml @@ -16,16 +16,18 @@ services: context: . dockerfile: Dockerfile.soffice-debian ports: - - "2002:2002" + - "127.0.0.1:2002:2002" soffice-cli: build: context: . dockerfile: Dockerfile.soffice-debian - container_name: soffice-cli command: ["sleep", "infinity"] + healthcheck: + disable: true volumes: - - ./fixtures:/fixtures:ro + - ./fixtures:/benchmark-fixtures:ro + - ../test/fixtures:/test-fixtures:ro soffice-alpine: image: libreofficedocker/alpine:3.23 @@ -35,9 +37,9 @@ services: - --norestore - --accept=socket,host=0.0.0.0,port=2002,tcpNoDelay=1;urp; ports: - - "2003:2002" + - "127.0.0.1:2003:2002" gotenberg: image: gotenberg/gotenberg:8.32.0 ports: - - "3002:3000" + - "127.0.0.1:3002:3000" diff --git a/flake.nix b/flake.nix index 4c4d4d8..ca1fa85 100644 --- a/flake.nix +++ b/flake.nix @@ -18,6 +18,8 @@ packages = [ beamPackages.erlang beamPackages.elixir_1_20 + beamPackages.hex + beamPackages.rebar3 pkgs.git pkgs.uv ]; diff --git a/lib/urp.ex b/lib/urp.ex index 6fc15d8..987fd69 100644 --- a/lib/urp.ex +++ b/lib/urp.ex @@ -35,7 +35,7 @@ defmodule URP do For multiple soffice instances, configure named pools: config :urp, :pools, - spreadsheets: [host: "soffice-2", port: 2002, pool_size: 3] + spreadsheets: [host: "soffice-2", port: 2002, pool_size: 1] {:ok, pdf} = URP.convert({:binary, bytes}, filter: "calc_pdf_Export", pool: :spreadsheets) @@ -224,7 +224,8 @@ defmodule URP do ## Input types - * `path` (binary) — local file path, loaded via file-backed streaming + * `path` (binary) — local file path. File mode reads it locally before + uploading it to soffice; use stream input for bounded local memory. * `{:binary, bytes}` — raw document bytes * enumerable — any `Enumerable` (e.g. `File.stream!/2`), streamed lazily @@ -238,6 +239,8 @@ defmodule URP do * `:settings` — list of `{path, property, value}` triplets to set on soffice before conversion via `ConfigurationUpdateAccess`. Useful for tuning cache limits, graphic memory, etc. Values can be booleans, integers, or strings. + These settings modify the soffice process-wide profile and persist after + the conversion; do not vary them concurrently on a shared soffice process. See [officecfg schema](https://github.com/LibreOffice/core/tree/master/officecfg/registry/schema/org/openoffice/Office) for all available settings. * `:io` — I/O transfer strategy (default `:file`): @@ -249,7 +252,8 @@ defmodule URP do overhead — soffice writes in fixed [32 767-byte chunks](https://github.com/LibreOffice/core/blob/libreoffice-26-2-0/sfx2/source/doc/docfile.cxx#L2573), so a 7 MB PDF is ~223 writeBytes calls. - No temp files and constant memory usage. + No temp files. Path and callback outputs use bounded memory; `:binary` + output necessarily accumulates the complete result in memory. * `{:file, :stream}` or `{:stream, :file}` — mix strategies independently for input and output. `{:file, :stream}` is a good defensive choice: fast file-based input with chunked stream output (no large single @@ -333,17 +337,13 @@ defmodule URP do end defp do_convert(input, opts) do + validate_convert_opts!(opts) + case URP.Test.__fetch_stub__() do {:ok, fun} -> fun.(input, opts) :error -> - if not Keyword.has_key?(opts, :filter) do - raise ArgumentError, - "URP.convert/2 requires the :filter option. " <> - "Common filters: \"writer_pdf_Export\", \"calc_pdf_Export\", \"impress_pdf_Export\", \"Markdown\"" - end - {pool, opts} = resolve_pool(opts) {output, opts} = Keyword.pop(opts, :output) @@ -361,6 +361,10 @@ defmodule URP do fun when is_function(fun, 1) -> Keyword.put(opts, :sink, fun) + + other -> + raise ArgumentError, + ":output must be a path, :binary, or a one-argument function; got: #{inspect(other)}" end case URP.Pool.convert(pool, input, pool_opts) do @@ -373,6 +377,101 @@ defmodule URP do end end + defp validate_convert_opts!(opts) do + if not Keyword.keyword?(opts) do + raise ArgumentError, "URP.convert/2 options must be a keyword list" + end + + filter = Keyword.get(opts, :filter) + + if not (is_binary(filter) and filter != "") do + raise ArgumentError, + "URP.convert/2 requires the :filter option. " <> + "Common filters: \"writer_pdf_Export\", \"calc_pdf_Export\", \"impress_pdf_Export\", \"Markdown\"" + end + + validate_io!(Keyword.get(opts, :io, :file)) + validate_timeout!(:timeout, Keyword.get(opts, :timeout, 120_000)) + validate_timeout!(:recv_timeout, Keyword.get(opts, :recv_timeout, 120_000)) + validate_output!(Keyword.get(opts, :output)) + validate_filter_data!(Keyword.get(opts, :filter_data, [])) + validate_settings!(Keyword.get(opts, :settings, [])) + + case Keyword.get(opts, :max_frame_size, 1) do + value when is_integer(value) and value > 0 -> + :ok + + value -> + raise ArgumentError, ":max_frame_size must be a positive integer; got: #{inspect(value)}" + end + + case Keyword.get(opts, :pool) do + nil -> :ok + name when is_atom(name) -> :ok + name -> raise ArgumentError, ":pool must be a configured atom; got: #{inspect(name)}" + end + end + + defp validate_io!(:file), do: :ok + defp validate_io!(:stream), do: :ok + + defp validate_io!({input, output}) + when input in [:file, :stream] and output in [:file, :stream], do: :ok + + defp validate_io!(value) do + raise ArgumentError, + ":io must be :file, :stream, or {:file | :stream, :file | :stream}; got: #{inspect(value)}" + end + + defp validate_timeout!(_name, :infinity), do: :ok + defp validate_timeout!(_name, value) when is_integer(value) and value >= 0, do: :ok + + defp validate_timeout!(name, value) do + raise ArgumentError, + ":#{name} must be a non-negative integer or :infinity; got: #{inspect(value)}" + end + + defp validate_output!(nil), do: :ok + defp validate_output!(:binary), do: :ok + defp validate_output!(path) when is_binary(path), do: :ok + defp validate_output!(fun) when is_function(fun, 1), do: :ok + + defp validate_output!(value) do + raise ArgumentError, + ":output must be a path, :binary, or a one-argument function; got: #{inspect(value)}" + end + + defp validate_filter_data!(filter_data) do + valid? = + Keyword.keyword?(filter_data) and + Enum.all?(filter_data, fn {_key, value} -> + is_boolean(value) or is_integer(value) or is_binary(value) + end) + + if not valid? do + raise ArgumentError, + ":filter_data must be a keyword list with boolean, integer, or string values" + end + end + + defp validate_settings!(settings) do + valid? = + is_list(settings) and + Enum.all?(settings, fn + {path, property, value} -> + is_binary(path) and is_binary(property) and + (is_boolean(value) or is_integer(value) or is_binary(value)) + + _other -> + false + end) + + if not valid? do + raise ArgumentError, + ":settings must contain {path, property, value} triplets with supported values" + end + end + @filter_extensions %{ "writer_pdf_Export" => ".pdf", "calc_pdf_Export" => ".pdf", @@ -396,7 +495,7 @@ defmodule URP do end ext = Map.get(@filter_extensions, filter, ".bin") - id = :erlang.unique_integer([:positive]) + id = :crypto.strong_rand_bytes(18) |> Base.url_encode64(padding: false) Path.join(System.tmp_dir!(), "#{basename}_#{id}#{ext}") end @@ -407,7 +506,24 @@ defmodule URP do end @doc false + def ensure_pool!(name) when not is_atom(name) do + raise ArgumentError, "pool name must be a configured atom; got: #{inspect(name)}" + end + def ensure_pool!(name) do + pools = Application.get_env(:urp, :pools, []) + + config = + case Keyword.fetch(pools, name) do + {:ok, config} -> + config + + :error -> + raise ArgumentError, + "pool #{inspect(name)} is not configured. " <> + "Add it to config :urp, :pools, #{name}: [host: \"...\", port: 2002]" + end + pid_name = pool_process_name(name) case GenServer.whereis(pid_name) do @@ -415,27 +531,24 @@ defmodule URP do pid_name nil -> - pools = Application.get_env(:urp, :pools, []) - - case Keyword.fetch(pools, name) do - {:ok, config} -> - opts = - [ - name: pid_name, - host: Keyword.get(config, :host, "localhost"), - port: Keyword.get(config, :port, 2002), - pool_size: Keyword.get(config, :pool_size, 1) - ] ++ Keyword.take(config, [:backoff_initial, :backoff_max]) - - case DynamicSupervisor.start_child(URP.PoolSupervisor, {URP.Pool, opts}) do - {:ok, _pid} -> pid_name - {:error, {:already_started, _pid}} -> pid_name - end - - :error -> - raise ArgumentError, - "pool #{inspect(name)} is not configured. " <> - "Add it to config :urp, :pools, #{name}: [host: \"...\", port: 2002]" + opts = + [ + name: pid_name, + host: Keyword.get(config, :host, "localhost"), + port: Keyword.get(config, :port, 2002), + pool_size: Keyword.get(config, :pool_size, 1) + ] ++ + Keyword.take(config, [ + :backoff_initial, + :backoff_max, + :connect_timeout, + :send_timeout + ]) + + case DynamicSupervisor.start_child(URP.PoolSupervisor, {URP.Pool, opts}) do + {:ok, _pid} -> pid_name + {:error, {:already_started, _pid}} -> pid_name + {:error, reason} -> raise "could not start pool #{inspect(name)}: #{inspect(reason)}" end end end diff --git a/lib/urp/bridge.ex b/lib/urp/bridge.ex index 314b077..6b64768 100644 --- a/lib/urp/bridge.ex +++ b/lib/urp/bridge.ex @@ -7,8 +7,10 @@ defmodule URP.Bridge do storing to URL, and closing. Each connection performs a handshake and bootstraps a Desktop reference - on `open/2`. A connection handles one conversion at a time (soffice is - single-threaded). Close the connection with `close!/1` when done. + on `open/2`. A connection handles one conversion at a time. LibreOffice can + accept multiple connections, but they share process-wide state and much of + the document core is serialized internally. Close the connection with + `close!/1` when done. All functions take and return `conn`, following the Plug pattern. Document OIDs and conversion state are stashed on the conn struct — no separate @@ -73,6 +75,7 @@ defmodule URP.Bridge do reply: term(), error: String.t() | nil, tid_cache: map(), + oid_cache: map(), private: map() } @@ -92,6 +95,7 @@ defmodule URP.Bridge do recv_timeout: 120_000, max_frame_size: 512 * 1024 * 1024, tid_cache: %{}, + oid_cache: %{}, private: %{} ] @@ -100,11 +104,24 @@ defmodule URP.Bridge do # Configuration provider singleton path (used by version and locale) @config_provider_path "/singletons/com.sun.star.configuration.theDefaultProvider" + @cleanup_timeout 1_000 + @default_connect_timeout 5_000 + @default_send_timeout 120_000 @doc "Connect to soffice, perform URP handshake, and bootstrap a Desktop reference." - @spec open(String.t(), non_neg_integer()) :: t() - def open(host \\ "localhost", port \\ 2002) do - case :gen_tcp.connect(String.to_charlist(host), port, [:binary, active: false]) do + @spec open(String.t(), non_neg_integer(), keyword()) :: t() + def open(host \\ "localhost", port \\ 2002, opts \\ []) do + connect_timeout = Keyword.get(opts, :connect_timeout, @default_connect_timeout) + send_timeout = Keyword.get(opts, :send_timeout, @default_send_timeout) + + socket_opts = [ + :binary, + active: false, + send_timeout: send_timeout, + send_timeout_close: true + ] + + case :gen_tcp.connect(String.to_charlist(host), port, socket_opts, connect_timeout) do {:ok, sock} -> init_connection(sock) @@ -114,10 +131,13 @@ defmodule URP.Bridge do end defp init_connection(sock) do + Process.put(:urp_tid_cache, %{}) + Process.put(:urp_oid_cache, %{}) + %__MODULE__{sock: sock} |> handshake!() |> bootstrap() - |> capture_tid_cache() + |> capture_protocol_caches() rescue e -> :gen_tcp.close(sock) @@ -180,6 +200,71 @@ defmodule URP.Bridge do %{conn | doc_oid: nil} end + @doc false + @spec cleanup(t()) :: t() + def cleanup(%__MODULE__{} = conn) do + original_reply = conn.reply + original_timeout = conn.recv_timeout + conn = %{conn | recv_timeout: min(original_timeout, @cleanup_timeout)} + + conn = + if conn.sock && conn.doc_oid do + attempt_cleanup(conn, &close_document/1) + else + %{conn | doc_oid: nil} + end + + cleanup_url = conn.cleanup_url + + conn = + if conn.sock && cleanup_url do + attempt_cleanup(conn, &delete_file(&1, cleanup_url)) + else + conn + end + + conn = release_input(conn) + + %{ + conn + | doc_oid: nil, + cleanup_url: nil, + input_ctx: nil, + reply: original_reply, + recv_timeout: original_timeout + } + end + + @doc false + @spec release_input(t()) :: t() + def release_input(%__MODULE__{input_ctx: nil} = conn), do: conn + + def release_input(%__MODULE__{input_ctx: %{source: source}} = conn) do + release_result = + case source do + {:file, fd, _size} -> File.close(fd) + {:enum, _buffer, reader} -> URP.Stream.stop_enum_reader(reader) + {:enum, _buffer, reader, _timeout} -> URP.Stream.stop_enum_reader(reader) + _other -> :ok + end + + conn = %{conn | input_ctx: nil} + + case {conn.error, release_result} do + {nil, {:error, reason}} -> + %{conn | error: "could not release input stream: #{:file.format_error(reason)}"} + + _ -> + conn + end + end + + defp attempt_cleanup(conn, fun) do + primary_error = conn.error + cleaned = fun.(%{conn | error: nil, reply: nil}) + %{cleaned | error: primary_error || cleaned.error} + end + @doc """ Query the soffice version string over URP. @@ -190,12 +275,11 @@ defmodule URP.Bridge do def version(%__MODULE__{error: e} = conn) when not is_nil(e), do: conn def version(%__MODULE__{} = conn) do - conn = call(conn, C.get_value_by_name(conn.ctx_oid, @config_provider_path), :qi) - conn = call(conn, C.qi_msf(conn.reply), :qi) - conn = call(conn, C.create_config_access(), :interface) - conn - |> call(C.qi_name_access(conn.reply), :qi) + |> call(C.get_value_by_name(conn.ctx_oid, @config_provider_path), :qi) + |> call_reply(&C.qi_msf/1, :qi) + |> call(C.create_config_access(), :interface) + |> call_reply(&C.qi_name_access/1, :qi) |> call(C.get_version(), :string) |> stash_private(:version) end @@ -233,7 +317,7 @@ defmodule URP.Bridge do ) conn - |> call(C.qi_ff_name_access(conn.reply), :qi) + |> call_reply(&C.qi_ff_name_access/1, :qi) |> call(C.get_filter_element_names(), :strings) |> stash_private(:filters) end @@ -256,7 +340,7 @@ defmodule URP.Bridge do ) conn - |> call(C.qi_td_name_access(conn.reply), :qi) + |> call_reply(&C.qi_td_name_access/1, :qi) |> call(C.get_type_element_names(), :strings) |> stash_private(:types) end @@ -271,12 +355,11 @@ defmodule URP.Bridge do def locale(%__MODULE__{error: e} = conn) when not is_nil(e), do: conn def locale(%__MODULE__{} = conn) do - conn = call(conn, C.get_value_by_name(conn.ctx_oid, @config_provider_path), :qi) - conn = call(conn, C.qi_locale_msf(conn.reply), :qi) - conn = call(conn, C.create_locale_config_access(), :interface) - conn - |> call(C.qi_locale_na(conn.reply), :qi) + |> call(C.get_value_by_name(conn.ctx_oid, @config_provider_path), :qi) + |> call_reply(&C.qi_locale_msf/1, :qi) + |> call(C.create_locale_config_access(), :interface) + |> call_reply(&C.qi_locale_na/1, :qi) |> call(C.get_locale(), :string) |> stash_private(:locale) end @@ -313,17 +396,21 @@ defmodule URP.Bridge do |> then(&call(&1, C.qi_settings_msf(&1.reply), :qi)) |> then(&call(&1, C.create_config_update_access(nodepath), :interface)) - update_oid = conn.reply + if conn.error do + conn + else + update_oid = conn.reply - conn - |> call(C.qi_name_replace(update_oid), :qi) - |> then(fn conn -> - Enum.reduce(props, conn, fn {_path, name, value}, conn -> - call(conn, C.replace_by_name(name, value), :void) + conn + |> call(C.qi_name_replace(update_oid), :qi) + |> then(fn conn -> + Enum.reduce(props, conn, fn {_path, name, value}, conn -> + call(conn, C.replace_by_name(name, value), :void) + end) end) - end) - |> call(C.qi_changes_batch(update_oid), :qi) - |> call(C.commit_changes(), :void) + |> call(C.qi_changes_batch(update_oid), :qi) + |> call(C.commit_changes(), :void) + end end @doc """ @@ -355,11 +442,8 @@ defmodule URP.Bridge do def load_document_file_stream(%__MODULE__{} = conn, path) when is_binary(path) do with {:ok, %{size: size}} <- File.stat(path), {:ok, fd} <- File.open(path, [:read, :binary, :raw]) do - try do - load_from_input_source(conn, {:file, fd, size}) - after - File.close(fd) - end + conn = load_from_input_source(conn, {:file, fd, size}) + if conn.error, do: release_input(conn), else: conn else {:error, reason} -> %{conn | error: "#{path}: #{:file.format_error(reason)}"} @@ -370,8 +454,8 @@ defmodule URP.Bridge do Load a document from an enumerable via XInputStream. Like `load_document_stream/2` but pulls chunks lazily from any `Enumerable` - (e.g. `File.stream!/2`, an S3 download stream). The enumerable is iterated - in a linked process; chunks are buffered and fed to soffice on demand. + (e.g. `File.stream!/2`, an S3 download stream). A demand-driven reader keeps + at most the requested data buffered and feeds soffice on demand. Stashes the document OID on `conn.doc_oid`. """ @spec load_document_enum_stream(t(), Enumerable.t()) :: t() @@ -381,13 +465,8 @@ defmodule URP.Bridge do def load_document_enum_stream(%__MODULE__{} = conn, enumerable) do reader = URP.Stream.start_enum_reader(enumerable) - - try do - load_from_input_source(conn, {:enum, <<>>, reader}) - after - Process.unlink(reader) - Process.exit(reader, :kill) - end + conn = load_from_input_source(conn, {:enum, <<>>, reader, conn.recv_timeout}) + if conn.error, do: release_input(conn), else: conn end @doc """ @@ -405,21 +484,17 @@ defmodule URP.Bridge do def load_document_write(%__MODULE__{error: e} = conn, _bytes) when not is_nil(e), do: conn def load_document_write(%__MODULE__{} = conn, bytes) when is_binary(bytes) do - conn = seed_tid_cache(conn) + conn = seed_protocol_caches(conn) conn = ensure_sfa(conn) - id = :erlang.unique_integer([:positive]) - url = "file:///tmp/urp_in_#{id}" + url = temp_url("urp_in") - conn = call(conn, C.sfa_open_file_write(conn.sfa_oid, url), :interface) - - conn = - conn - |> call(C.qi_sfa_output(conn.reply), :qi) - |> call(C.write_bytes(bytes), :void) - |> call(C.close_output(), :void) - |> load_document(url) - - %{conn | cleanup_url: url} + conn + |> call(C.sfa_open_file_write(conn.sfa_oid, url), :interface) + |> call_reply(&C.qi_sfa_output/1, :qi) + |> call(C.write_bytes(bytes), :void) + |> call(C.close_output(), :void) + |> load_document(url) + |> then(&%{&1 | cleanup_url: url}) end @doc "Delete a temp file on soffice's filesystem via XSimpleFileAccess.kill()." @@ -430,6 +505,12 @@ defmodule URP.Bridge do call(conn, C.sfa_kill(sfa_oid, url), :void) end + defp delete_file_preserving_error(conn, url) do + reply = conn.reply + conn = attempt_cleanup(conn, &delete_file(&1, url)) + %{conn | reply: reply} + end + @doc """ Store a document to soffice's filesystem and read back the result. @@ -444,15 +525,14 @@ defmodule URP.Bridge do def store_document_write(%__MODULE__{doc_oid: doc_oid} = conn, opts) when is_binary(doc_oid) do filter = Keyword.fetch!(opts, :filter) filter_data = Keyword.get(opts, :filter_data, []) - id = :erlang.unique_integer([:positive]) - url = "file:///tmp/urp_out_#{id}" + url = temp_url("urp_out") conn = conn |> store_to_url(url, filter, filter_data) |> read_file(url) - %{delete_file(conn, url) | reply: conn.reply} + delete_file_preserving_error(conn, url) end @doc """ @@ -466,23 +546,26 @@ defmodule URP.Bridge do def read_file(%__MODULE__{} = conn, url) do conn = ensure_sfa(conn) - conn = call(conn, C.sfa_open_file_read(conn.sfa_oid, url), :interface) conn = conn - |> call(C.qi_sfa_input(conn.reply), :qi) + |> call(C.sfa_open_file_read(conn.sfa_oid, url), :interface) + |> call_reply(&C.qi_sfa_input/1, :qi) |> call(C.available(), :int32) + |> call_reply(&C.read_bytes/1, :read_bytes) - size = conn.reply - conn = call(conn, C.read_bytes(size), :read_bytes) - bytes = conn.reply - - %{call(conn, C.close_input(), :void) | reply: bytes} + if conn.error do + conn + else + bytes = conn.reply + %{call(conn, C.close_input(), :void) | reply: bytes} + end end defp load_from_input_source(conn, source) do - conn = seed_tid_cache(conn) + conn = seed_protocol_caches(conn) stream_oid = "elixir-in-#{:erlang.unique_integer([:positive])}" + conn = %{conn | input_ctx: %{source: source, seekable_cache: nil, input_cache: nil}} conn = conn @@ -498,11 +581,15 @@ defmodule URP.Bridge do conn else # soffice will call readBytes/available/closeInput/seek/getPosition/getLength on our stream - {reply, conn} = URP.Stream.recv_handling_input(conn, source, stream_oid) - - case P.parse_interface_reply(reply) do - {:ok, doc_oid} -> %{conn | doc_oid: doc_oid} - {:error, message} -> %{conn | error: message} + try do + {reply, conn} = URP.Stream.recv_handling_input(conn, source, stream_oid) + + case P.parse_interface_reply(reply) do + {:ok, doc_oid} -> %{conn | doc_oid: doc_oid} + {:error, message} -> %{conn | error: message} + end + rescue + error -> %{conn | error: "input stream failed: #{Exception.message(error)}"} end end end @@ -542,8 +629,26 @@ defmodule URP.Bridge do if conn.error do conn else - {_reply, result, conn} = URP.Stream.recv_handling_output(conn, sink) - %{conn | reply: result} + try do + {reply, result, conn} = URP.Stream.recv_handling_output(conn, sink) + + case {P.parse_exception(reply), result} do + {message, _result} when is_binary(message) -> + remove_partial_sink(sink) + %{conn | error: message, reply: nil} + + {nil, {:error, message}} -> + remove_partial_sink(sink) + %{conn | error: message, reply: nil} + + {nil, result} -> + %{conn | reply: result} + end + rescue + error -> + remove_partial_sink(sink) + %{conn | error: "output stream failed: #{Exception.message(error)}", reply: nil} + end end end @@ -559,11 +664,9 @@ defmodule URP.Bridge do :interface ) - sfa_oid = conn.reply - conn |> stash(:sfa_oid) - |> call(C.qi_sfa(sfa_oid), :qi) + |> call_reply(&C.qi_sfa/1, :qi) end ## Handshake @@ -606,6 +709,18 @@ defmodule URP.Bridge do props ++ extra end + defp temp_url(prefix) do + id = :crypto.strong_rand_bytes(18) |> Base.url_encode64(padding: false) + "file:///tmp/#{prefix}_#{id}" + end + + defp remove_partial_sink({:path, path}) do + File.rm(path) + :ok + end + + defp remove_partial_sink(_sink), do: :ok + ## Send + receive + parse defp call(%__MODULE__{} = conn, frame) do @@ -622,26 +737,47 @@ defmodule URP.Bridge do |> parse_reply(parser) end + defp call_reply(%__MODULE__{error: e} = conn, _builder, _parser) when not is_nil(e), do: conn + + defp call_reply(%__MODULE__{} = conn, builder, parser), + do: call(conn, builder.(conn.reply), parser) + defp parse_reply(%__MODULE__{error: e} = conn, _parser) when not is_nil(e), do: conn - defp parse_reply(conn, :qi), do: handle_parsed(conn, P.parse_qi_reply(conn.reply)) - defp parse_reply(conn, :interface), do: handle_parsed(conn, P.parse_interface_reply(conn.reply)) - defp parse_reply(conn, :string), do: handle_parsed(conn, P.parse_any_string_reply(conn.reply)) + defp parse_reply(conn, :qi), do: parse_safely(conn, :qi, &P.parse_qi_reply/1) + + defp parse_reply(conn, :interface), + do: parse_safely(conn, :interface, &P.parse_interface_reply/1) + + defp parse_reply(conn, :string), + do: parse_safely(conn, :string, &P.parse_any_string_reply/1) defp parse_reply(conn, :strings), - do: handle_parsed(conn, P.parse_string_sequence_reply(conn.reply)) + do: parse_safely(conn, :strings, &P.parse_string_sequence_reply/1) - defp parse_reply(conn, :int32), do: handle_parsed(conn, P.parse_int32_reply(conn.reply)) + defp parse_reply(conn, :int32), do: parse_safely(conn, :int32, &P.parse_int32_reply/1) defp parse_reply(conn, :read_bytes), - do: handle_parsed(conn, P.parse_read_bytes_reply(conn.reply)) + do: parse_safely(conn, :read_bytes, &P.parse_read_bytes_reply/1) defp parse_reply(conn, :void) do - case P.parse_exception(conn.reply) do - nil -> conn - message -> %{conn | error: message, reply: nil} + try do + case P.parse_exception(conn.reply) do + nil -> conn + message -> %{conn | error: message, reply: nil} + end + rescue + error -> + %{conn | error: "malformed URP void reply: #{Exception.message(error)}", reply: nil} end end + defp parse_safely(conn, parser, fun) do + handle_parsed(conn, fun.(conn.reply)) + rescue + error -> + %{conn | error: "malformed URP #{parser} reply: #{Exception.message(error)}", reply: nil} + end + defp handle_parsed(conn, {:ok, value}), do: %{conn | reply: value} defp handle_parsed(conn, {:error, msg}), do: %{conn | error: msg, reply: nil} @@ -680,22 +816,19 @@ defmodule URP.Bridge do e -> %{conn | error: Exception.message(e)} end - # TID cache transfer: protocol parsing during bootstrap populates TIDs in the - # process dictionary. capture_tid_cache saves them onto conn so they survive - # the process exit. seed_tid_cache restores them in the checkout caller's - # process so reply parsing works. - defp capture_tid_cache(conn) do - case Process.get(:urp_tid_cache) do - nil -> conn - cache -> %{conn | tid_cache: cache} - end + # Protocol parsing uses process-local read caches. Pool checkout scopes these + # values to one connection and persists the updated maps back onto the worker. + defp capture_protocol_caches(conn) do + %{ + conn + | tid_cache: Process.get(:urp_tid_cache, conn.tid_cache), + oid_cache: Process.get(:urp_oid_cache, conn.oid_cache) + } end - defp seed_tid_cache(%__MODULE__{tid_cache: cache} = conn) when cache == %{}, do: conn - - defp seed_tid_cache(%__MODULE__{tid_cache: cache} = conn) do - existing = Process.get(:urp_tid_cache, %{}) - Process.put(:urp_tid_cache, Map.merge(cache, existing)) + defp seed_protocol_caches(%__MODULE__{} = conn) do + if is_nil(Process.get(:urp_tid_cache)), do: Process.put(:urp_tid_cache, conn.tid_cache) + if is_nil(Process.get(:urp_oid_cache)), do: Process.put(:urp_oid_cache, conn.oid_cache) conn end diff --git a/lib/urp/call.ex b/lib/urp/call.ex index deb9c71..3e3545c 100644 --- a/lib/urp/call.ex +++ b/lib/urp/call.ex @@ -497,7 +497,7 @@ defmodule URP.Call do P.enc_str(url), P.enc_str("_blank"), @frame_search_default, - <>, + P.enc_count(length(props)), props ] end @@ -509,7 +509,7 @@ defmodule URP.Call do [ @store_to_url_prefix, P.enc_str(url), - <>, + P.enc_count(length(props)), props ] end @@ -688,7 +688,7 @@ defmodule URP.Call do [ <<@cache_filter_data_seq::16>>, P.enc_str("[]com.sun.star.beans.PropertyValue"), - <>, + P.enc_count(length(filter_data)), inner ] ) diff --git a/lib/urp/pool.ex b/lib/urp/pool.ex index 4980e20..60e8d83 100644 --- a/lib/urp/pool.ex +++ b/lib/urp/pool.ex @@ -71,7 +71,7 @@ defmodule URP.Pool do if is_nil(conn.error) do {{:ok, conn.private[key]}, {:ok, reset_conversion_state(conn)}} else - {{:error, conn.error}, {:ok, reset_conversion_state(conn)}} + {{:error, conn.error}, :closed} end end) end @@ -90,8 +90,12 @@ defmodule URP.Pool do store_opts = Keyword.take(opts, [:filter, :filter_data]) meta = %{operation: :convert, pool: pool} + stream_input? = io_in == :stream or enumerable_input?(input) do_checkout(pool, timeout, meta, fn conn -> + default_max_frame_size = conn.max_frame_size + default_recv_timeout = conn.recv_timeout + # Clear stale reply from bootstrap (desktop OID) on the first conversion. # Subsequent conversions are already clean via reset_conversion_state. conn = %{conn | reply: nil} @@ -105,21 +109,14 @@ defmodule URP.Pool do |> store_output(store_opts, sink, io_out) result = conn.reply - conn = Bridge.close_document(conn) - conn = safe_cleanup(conn) - - # Persist accumulated TID cache entries back to conn before clearing. - # soffice's type cache is per-connection and doesn't reset between - # operations — our side must keep up to avoid type desync. - tid_cache = Process.get(:urp_tid_cache, conn.tid_cache) - conn = %{conn | tid_cache: tid_cache} - Process.delete(:urp_tid_cache) + conn = Bridge.cleanup(conn) + conn = %{conn | max_frame_size: default_max_frame_size, recv_timeout: default_recv_timeout} # Stream-based input registers an XInputStream at a fixed OID cache slot. # soffice's URP cache doesn't fully reset on reuse, producing truncated # documents on subsequent stream loads. Discard the connection to force # a fresh handshake. File-based I/O reuses connections normally. - reusable = io_in == :file and is_nil(conn.error) + reusable = not stream_input? and is_nil(conn.error) has_result = is_binary(result) or result == :ok cond do @@ -139,7 +136,19 @@ defmodule URP.Pool do defp normalize_io(:file), do: {:file, :file} defp normalize_io(:stream), do: {:stream, :stream} - defp normalize_io({in_mode, out_mode}), do: {in_mode, out_mode} + + defp normalize_io({in_mode, out_mode}) + when in_mode in [:file, :stream] and out_mode in [:file, :stream], + do: {in_mode, out_mode} + + defp normalize_io(other) do + raise ArgumentError, + ":io must be :file, :stream, or {:file | :stream, :file | :stream}; got: #{inspect(other)}" + end + + defp enumerable_input?({:binary, bytes}) when is_binary(bytes), do: false + defp enumerable_input?(path) when is_binary(path), do: false + defp enumerable_input?(_input), do: true ## Input routing @@ -176,8 +185,15 @@ defmodule URP.Pool do defp store_output(conn, store_opts, sink, :file) do conn = Bridge.store_document_write(conn, store_opts) - result = if conn.reply, do: apply_sink(conn.reply, sink) - %{conn | reply: result} + + if conn.error do + conn + else + case apply_sink(conn.reply, sink) do + {:ok, result} -> %{conn | reply: result} + {:error, message} -> %{conn | error: message, reply: nil} + end + end end defp store_output(conn, store_opts, sink, :stream) do @@ -187,45 +203,106 @@ defmodule URP.Pool do defp do_checkout(pool, timeout, meta, fun, attempt \\ 1) do t0 = System.monotonic_time() + :telemetry.execute([:urp, :call, :start], %{system_time: System.system_time()}, meta) + + try do + {result, queue_time, service_time, attempts} = + do_checkout_attempt(pool, timeout, fun, attempt, 0, 0) + + total_time = System.monotonic_time() - t0 + + :telemetry.execute( + [:urp, :call, :stop], + %{ + total_time: total_time, + queue_time: queue_time, + service_time: service_time, + backoff_time: max(total_time - queue_time - service_time, 0) + }, + meta |> Map.put(:result, result_tag(result)) |> Map.put(:attempts, attempts) + ) + + result + catch + kind, reason -> + stacktrace = __STACKTRACE__ + + :telemetry.execute( + [:urp, :call, :exception], + %{duration: System.monotonic_time() - t0}, + Map.merge(meta, %{kind: kind, reason: reason, stacktrace: stacktrace}) + ) + + case kind do + :exit -> {:error, "pool checkout failed: #{Exception.format_exit(reason)}"} + _ -> :erlang.raise(kind, reason, stacktrace) + end + end + end + + defp do_checkout_attempt(pool, timeout, fun, attempt, queue_acc, service_acc) do + attempt_started = System.monotonic_time() - result = + {result, queue_time, service_time} = NimblePool.checkout!( pool, :checkout, fn _from, conn -> - t1 = System.monotonic_time() - {result, checkin} = fun.(conn) - t2 = System.monotonic_time() - - :telemetry.execute( - [:urp, :call, :stop], - %{ - total_time: t2 - t0, - queue_time: t1 - t0, - service_time: t2 - t1 - }, - Map.put(meta, :result, result_tag(result)) - ) - - {result, checkin} + checked_out = System.monotonic_time() + + {result, checkin} = + with_protocol_caches(conn, fn -> + fun.(conn) + end) + + finished = System.monotonic_time() + {{result, checked_out - attempt_started, finished - checked_out}, checkin} end, timeout ) - case result do - {:error, message} when is_binary(message) and attempt < @disposed_max_retries -> - if String.contains?(message, @bridge_disposed) do - Process.sleep(@disposed_retry_interval_ms * attempt) - do_checkout(pool, timeout, meta, fun, attempt + 1) - else - result - end + queue_acc = queue_acc + queue_time + service_acc = service_acc + service_time - _ -> - result + if disposed_error?(result) and attempt < @disposed_max_retries do + Process.sleep(@disposed_retry_interval_ms * attempt) + do_checkout_attempt(pool, timeout, fun, attempt + 1, queue_acc, service_acc) + else + {result, queue_acc, service_acc, attempt} end end + defp disposed_error?({:error, message}) when is_binary(message), + do: String.contains?(message, @bridge_disposed) + + defp disposed_error?(_result), do: false + + defp with_protocol_caches(conn, fun) do + missing = make_ref() + previous_tid = Process.get(:urp_tid_cache, missing) + previous_oid = Process.get(:urp_oid_cache, missing) + Process.put(:urp_tid_cache, conn.tid_cache) + Process.put(:urp_oid_cache, conn.oid_cache) + + try do + {result, checkin} = fun.() + tid_cache = Process.get(:urp_tid_cache, %{}) + oid_cache = Process.get(:urp_oid_cache, %{}) + {result, persist_protocol_caches(checkin, tid_cache, oid_cache)} + after + restore_process_value(:urp_tid_cache, previous_tid, missing) + restore_process_value(:urp_oid_cache, previous_oid, missing) + end + end + + defp persist_protocol_caches({:ok, conn}, tid_cache, oid_cache), + do: {:ok, %{conn | tid_cache: tid_cache, oid_cache: oid_cache}} + + defp persist_protocol_caches(checkin, _tid_cache, _oid_cache), do: checkin + + defp restore_process_value(key, missing, missing), do: Process.delete(key) + defp restore_process_value(key, value, _missing), do: Process.put(key, value) + defp result_tag({:error, _}), do: :error defp result_tag(_), do: :ok @@ -242,6 +319,7 @@ defmodule URP.Pool do port = Keyword.get(config, :port, 2002) backoff_initial = Keyword.get(config, :backoff_initial, @default_backoff_initial) backoff_max = Keyword.get(config, :backoff_max, @default_backoff_max) + bridge_opts = Keyword.take(config, [:connect_timeout, :send_timeout]) # Use {:async, ...} so start_link returns immediately even if soffice is # down and open_with_retry loops for a while. We must transfer socket @@ -251,14 +329,14 @@ defmodule URP.Pool do {:async, fn -> - conn = open_with_retry(host, port, backoff_initial, backoff_max) + conn = open_with_retry(host, port, backoff_initial, backoff_max, bridge_opts) if conn.sock, do: :gen_tcp.controlling_process(conn.sock, pool_pid) conn end, config} end - defp open_with_retry(host, port, backoff_initial, backoff_max, attempt \\ 1) do - conn = Bridge.open(host, port) + defp open_with_retry(host, port, backoff_initial, backoff_max, bridge_opts, attempt \\ 1) do + conn = Bridge.open(host, port, bridge_opts) cond do is_nil(conn.error) -> @@ -266,7 +344,7 @@ defmodule URP.Pool do is_nil(conn.sock) -> # TCP or handshake failed — soffice unreachable. Retry with backoff. - delay = min(backoff_initial * Integer.pow(2, attempt - 1), backoff_max) + delay = backoff_delay(backoff_initial, backoff_max, attempt) :telemetry.execute( [:urp, :connection, :retry], @@ -281,7 +359,7 @@ defmodule URP.Pool do ) Process.sleep(delay) - open_with_retry(host, port, backoff_initial, backoff_max, attempt + 1) + open_with_retry(host, port, backoff_initial, backoff_max, bridge_opts, attempt + 1) true -> # Connected but bootstrap failed — don't retry forever. @@ -290,6 +368,11 @@ defmodule URP.Pool do end end + defp backoff_delay(initial, maximum, attempt) do + capped_attempt = min(attempt - 1, 30) + min(initial * Integer.pow(2, capped_attempt), maximum) + end + @impl NimblePool def handle_checkout(:checkout, _from, conn, pool_state) do {:ok, conn, conn, pool_state} @@ -314,21 +397,27 @@ defmodule URP.Pool do {:ok, pool_state} end - defp apply_sink(bytes, nil), do: bytes + defp apply_sink(bytes, nil), do: {:ok, bytes} defp apply_sink(bytes, {:path, path}) do - File.write!(path, bytes) - :ok + case File.write(path, bytes) do + :ok -> + {:ok, :ok} + + {:error, reason} -> + {:error, "could not write output #{path}: #{:file.format_error(reason)}"} + end end defp apply_sink(bytes, fun) when is_function(fun, 1) do - fun.(bytes) - :ok + try do + fun.(bytes) + {:ok, :ok} + rescue + error -> {:error, "output sink failed: #{Exception.message(error)}"} + end end - defp safe_cleanup(%{cleanup_url: nil} = conn), do: conn - defp safe_cleanup(conn), do: Bridge.delete_file(conn, conn.cleanup_url) - defp reset_conversion_state(conn) do %{ conn diff --git a/lib/urp/protocol.ex b/lib/urp/protocol.ex index 7f38d6f..9967c56 100644 --- a/lib/urp/protocol.ex +++ b/lib/urp/protocol.ex @@ -72,6 +72,8 @@ defmodule URP.Protocol do # Read in chunks and reassemble. @recv_chunk_size 4 * 1024 * 1024 + defp recv_exact(_sock, 0, _timeout), do: <<>> + defp recv_exact(sock, size, timeout) when size <= @recv_chunk_size do {:ok, payload} = :gen_tcp.recv(sock, size, timeout) payload @@ -164,6 +166,11 @@ defmodule URP.Protocol do def enc_str_iodata(s) when byte_size(s) < 0xFF, do: [<>, s] def enc_str_iodata(s), do: [<<0xFF, byte_size(s)::32>>, s] + @doc "Encode a URP compressed sequence count." + @spec enc_count(non_neg_integer()) :: binary() + def enc_count(count) when count < 0xFF, do: <> + def enc_count(count), do: <<0xFF, count::32>> + @doc "Decode a compressed string, returning `{string, rest}`." @spec dec_str(binary()) :: {binary(), binary()} def dec_str(<<0xFF, len::32, s::binary-size(len), rest::binary>>), do: {s, rest} @@ -325,7 +332,7 @@ defmodule URP.Protocol do <> -> rest = if (tc &&& @tc_new) != 0, do: elem(dec_str(rest), 1), else: rest - {:ok, elem(dec_str(rest), 0)} + decode_oid_reference(rest) end end end @@ -338,8 +345,7 @@ defmodule URP.Protocol do if (flags &&& @exception) != 0 do {:error, parse_exception(payload)} else - {oid, _} = dec_str(rest) - if oid == "", do: {:error, "empty OID"}, else: {:ok, oid} + decode_oid_reference(rest) end end @@ -425,7 +431,7 @@ defmodule URP.Protocol do {message, _} = dec_str(rest) message rescue - MatchError -> "UNO exception (could not parse message)" + _error -> "UNO exception (could not parse message)" end end end @@ -453,6 +459,32 @@ defmodule URP.Protocol do defp dec_count(<<0xFF, count::32, rest::binary>>), do: {count, rest} defp dec_count(<>), do: {count, rest} + defp decode_oid_reference(rest) do + {oid, rest} = dec_str(rest) + <> = rest + + cond do + oid != "" -> + if cache != 0xFFFF do + oid_cache = Process.get(:urp_oid_cache, %{}) + Process.put(:urp_oid_cache, Map.put(oid_cache, cache, oid)) + end + + {:ok, oid} + + cache != 0xFFFF -> + case Process.get(:urp_oid_cache, %{}) do + %{^cache => cached_oid} -> {:ok, cached_oid} + _cache -> {:error, "unknown cached OID index #{cache}"} + end + + true -> + {:error, "empty OID"} + end + rescue + _error -> {:error, "malformed interface reference"} + end + defp dec_strings(_rest, 0, acc), do: Enum.reverse(acc) defp dec_strings(rest, n, acc) do diff --git a/lib/urp/stream.ex b/lib/urp/stream.ex index 1994c7a..fdb749e 100644 --- a/lib/urp/stream.ex +++ b/lib/urp/stream.ex @@ -31,11 +31,14 @@ defmodule URP.Stream do import Bitwise - @type input_source :: binary() | {:file, pid(), non_neg_integer()} | {:enum, binary(), pid()} + @type input_source :: + binary() + | {:file, File.io_device(), non_neg_integer()} + | {:enum, binary(), pid(), timeout()} @typep source :: {:mem, binary(), non_neg_integer()} - | {:file, pid(), non_neg_integer()} - | {:enum, binary(), pid() | :eof} + | {:file, File.io_device(), non_neg_integer()} + | {:enum, binary(), pid() | :eof, timeout()} @type sink :: nil | {:path, Path.t()} | (binary() -> any()) @type connection :: map() @@ -210,7 +213,7 @@ defmodule URP.Stream do @ole2_magic <<0xD0, 0xCF, 0x11, 0xE0>> @magic_bytes_len max(byte_size(@zip_magic), byte_size(@ole2_magic)) - defp seekable_source?({:enum, _, _}), do: false + defp seekable_source?({:enum, _, _, _}), do: false defp seekable_source?({:mem, data, _pos}) do match?(@zip_magic <> _, data) or match?(@ole2_magic <> _, data) @@ -258,7 +261,8 @@ defmodule URP.Stream do end defp handle_input(6, _body, src) do - {P.reply(<>), src} + count = available(src) |> max(0) |> min(2_147_483_647) + {P.reply(<>), src} end defp handle_input(7, _body, src), do: {P.reply(), src} @@ -286,7 +290,8 @@ defmodule URP.Stream do * `{:path, path}` — write chunks to file as they arrive, returns `{reply, :ok, conn}` * `fun/1` — call with each chunk, returns `{reply, :ok, conn}` """ - @spec recv_handling_output(connection(), sink()) :: {binary(), binary() | :ok, connection()} + @spec recv_handling_output(connection(), sink()) :: + {binary(), binary() | :ok | {:error, String.t()}, connection()} def recv_handling_output(conn, sink \\ nil) def recv_handling_output(conn, nil) do @@ -294,12 +299,19 @@ defmodule URP.Stream do end def recv_handling_output(conn, {:path, path}) do - fd = File.open!(path, [:write, :binary, :raw]) + case File.open(path, [:write, :binary, :raw]) do + {:ok, fd} -> + try do + do_recv_output(conn, {:file, fd}) + after + File.close(fd) + end - try do - do_recv_output(conn, {:file, fd}) - after - File.close(fd) + {:error, reason} -> + do_recv_output( + conn, + {:error, "could not open output #{path}: #{:file.format_error(reason)}"} + ) end end @@ -353,18 +365,27 @@ defmodule URP.Stream do defp write_sink({:mem, chunks}, data), do: {:mem, [data | chunks]} defp write_sink({:file, fd} = sink, data) do - IO.binwrite(fd, data) - sink + case :file.write(fd, data) do + :ok -> sink + {:error, reason} -> {:error, "could not write output: #{:file.format_error(reason)}"} + end end defp write_sink({:fun, fun} = sink, data) do - fun.(data) - sink + try do + fun.(data) + sink + rescue + error -> {:error, "output sink failed: #{Exception.message(error)}"} + end end + defp write_sink({:error, _message} = sink, _data), do: sink + defp finalize_sink({:mem, chunks}), do: chunks |> Enum.reverse() |> IO.iodata_to_binary() defp finalize_sink({:file, _fd}), do: :ok defp finalize_sink({:fun, _fun}), do: :ok + defp finalize_sink({:error, message}), do: {:error, message} ## Source-polymorphic read helpers ## @@ -396,11 +417,11 @@ defmodule URP.Stream do end # Enumerable-backed - defp read_chunk({:enum, buffer, reader}, n) do - {buffer, reader} = fill_buffer(buffer, reader, n) + defp read_chunk({:enum, buffer, reader, timeout}, n) do + {buffer, reader} = fill_buffer(buffer, reader, n, timeout) to_read = min(n, byte_size(buffer)) <> = buffer - {chunk, {:enum, rest, reader}} + {chunk, {:enum, rest, reader, timeout}} end defp skip_chunk({:mem, data, pos}, n) do @@ -415,11 +436,11 @@ defmodule URP.Stream do {:file, fd, size} end - defp skip_chunk({:enum, buffer, reader}, n) do - {buffer, reader} = fill_buffer(buffer, reader, n) + defp skip_chunk({:enum, buffer, reader, timeout}, n) do + {buffer, reader} = fill_buffer(buffer, reader, n, timeout) skip = min(n, byte_size(buffer)) <<_::binary-size(^skip), rest::binary>> = buffer - {:enum, rest, reader} + {:enum, rest, reader, timeout} end defp available({:mem, data, pos}), do: byte_size(data) - pos @@ -429,11 +450,11 @@ defmodule URP.Stream do size - pos end - defp available({:enum, buffer, :eof}), do: byte_size(buffer) + defp available({:enum, buffer, :eof, _timeout}), do: byte_size(buffer) # Unknown remaining size — return a conservative estimate so soffice # keeps reading without over-allocating (it may use this to size buffers). - defp available({:enum, buffer, _reader}), do: max(byte_size(buffer), 65_536) + defp available({:enum, buffer, _reader, _timeout}), do: max(byte_size(buffer), 65_536) ## XSeekable source helpers @@ -457,36 +478,124 @@ defmodule URP.Stream do defp get_length({:file, _fd, size}), do: size @doc """ - Spawn a linked process that iterates `enumerable`, sending chunks to the caller. + Spawn a demand-driven reader for `enumerable`. - The reader sends `{pid, {:chunk, data}}` for each element and `{pid, :eof}` - when the enumerable is exhausted. The caller receives these in `fill_buffer/3`. + The enumerable advances only after a `{:next, caller}` request, preventing + producer data from accumulating unboundedly in the caller's mailbox. """ @spec start_enum_reader(Enumerable.t()) :: pid() def start_enum_reader(enumerable) do - parent = self() - - spawn_link(fn -> - Enum.each(enumerable, fn chunk -> - data = IO.iodata_to_binary(chunk) - send(parent, {self(), {:chunk, data}}) - end) + owner = self() - send(parent, {self(), :eof}) + spawn(fn -> + owner_ref = Process.monitor(owner) + enum_reader_loop({:new, enumerable}, owner, owner_ref) end) end - defp fill_buffer(buffer, :eof, _needed), do: {buffer, :eof} + @doc false + @spec stop_enum_reader(pid() | :eof) :: :ok + def stop_enum_reader(:eof), do: :ok + + def stop_enum_reader(reader) when is_pid(reader) do + ref = Process.monitor(reader) + send(reader, :stop) + + receive do + {:DOWN, ^ref, :process, ^reader, _reason} -> :ok + after + 1_000 -> + Process.exit(reader, :kill) + + receive do + {:DOWN, ^ref, :process, ^reader, _reason} -> :ok + end + end + + flush_reader_messages(reader) + :ok + end + + defp enum_reader_loop(state, owner, owner_ref) do + receive do + {:next, caller} -> + case enum_next(state) do + {:chunk, data, continuation} -> + send(caller, {self(), {:chunk, data}}) + enum_reader_loop({:continuation, continuation}, owner, owner_ref) + + :eof -> + send(caller, {self(), :eof}) + enum_reader_loop(:eof, owner, owner_ref) + + {:error, message} -> + send(caller, {self(), {:error, message}}) + end + + :stop -> + halt_enum(state) + :ok - defp fill_buffer(buffer, reader, needed) when byte_size(buffer) >= needed, + {:DOWN, ^owner_ref, :process, ^owner, _reason} -> + halt_enum(state) + :ok + end + end + + defp enum_next(:eof), do: :eof + + defp enum_next({:new, enumerable}) do + reduce_one(fn command -> Enumerable.reduce(enumerable, command, &suspend_chunk/2) end) + end + + defp enum_next({:continuation, continuation}) do + reduce_one(continuation) + end + + defp reduce_one(reducer) do + case reducer.({:cont, nil}) do + {:suspended, data, continuation} -> {:chunk, data, continuation} + {:done, _acc} -> :eof + {:halted, _acc} -> :eof + end + rescue + error -> {:error, Exception.message(error)} + catch + kind, reason -> {:error, Exception.format(kind, reason, __STACKTRACE__)} + end + + defp suspend_chunk(chunk, _acc), do: {:suspend, IO.iodata_to_binary(chunk)} + + defp halt_enum({:continuation, continuation}) do + continuation.({:halt, nil}) + :ok + catch + _kind, _reason -> :ok + end + + defp halt_enum(_state), do: :ok + + defp flush_reader_messages(reader) do + receive do + {^reader, _message} -> flush_reader_messages(reader) + after + 0 -> :ok + end + end + + defp fill_buffer(buffer, :eof, _needed, _timeout), do: {buffer, :eof} + + defp fill_buffer(buffer, reader, needed, _timeout) when byte_size(buffer) >= needed, do: {buffer, reader} - defp fill_buffer(buffer, reader, needed) do + defp fill_buffer(buffer, reader, needed, timeout) do # Accumulate as iodata to avoid O(n²) binary concatenation - fill_buffer_acc([buffer], byte_size(buffer), reader, needed) + fill_buffer_acc([buffer], byte_size(buffer), reader, needed, timeout) end - defp fill_buffer_acc(acc, acc_size, reader, needed) do + defp fill_buffer_acc(acc, acc_size, reader, needed, timeout) do + send(reader, {:next, self()}) + receive do {^reader, {:chunk, data}} -> acc = [acc, data] @@ -495,11 +604,17 @@ defmodule URP.Stream do if acc_size >= needed do {IO.iodata_to_binary(acc), reader} else - fill_buffer_acc(acc, acc_size, reader, needed) + fill_buffer_acc(acc, acc_size, reader, needed, timeout) end {^reader, :eof} -> {IO.iodata_to_binary(acc), :eof} + + {^reader, {:error, message}} -> + raise "enumerable input failed: #{message}" + after + timeout -> + raise "enumerable input timed out after #{timeout}ms" end end diff --git a/lib/urp/telemetry.ex b/lib/urp/telemetry.ex index c0381db..a96a88b 100644 --- a/lib/urp/telemetry.ex +++ b/lib/urp/telemetry.ex @@ -4,10 +4,15 @@ defmodule URP.Telemetry do ## Events + ### `[:urp, :call, :start]` + + Emitted before requesting a connection from the pool. Measurements contain + `:system_time`; metadata contains `:operation` and `:pool`. + ### `[:urp, :call, :stop]` - Emitted after every pool checkout completes (all operations go through - `URP.Pool.do_checkout/5`). + Emitted when an operation completes normally, including operations that + return an error tuple. #### Measurements (in `:native` time units) @@ -16,6 +21,7 @@ defmodule URP.Telemetry do | `:total_time` | Wall time from checkout request to work completion | | `:queue_time` | Time spent waiting for a pool connection | | `:service_time` | Time spent doing work on the connection | + | `:backoff_time` | Time spent outside queue/service, primarily disposed-bridge retry backoff | Use `System.convert_time_unit/3` to convert to milliseconds or microseconds: @@ -29,6 +35,20 @@ defmodule URP.Telemetry do | `:operation` | `atom` | The operation performed (e.g. `:convert`, `:version`, `:filters`) | | `:pool` | `term` | The pool name or pid | | `:result` | `:ok \\| :error` | Whether the operation succeeded | + | `:attempts` | `pos_integer` | Number of checkout attempts | + + ### `[:urp, :call, :exception]` + + Emitted when checkout exits or operation code raises, throws, or exits. + Measurements contain `:duration`. Metadata contains `:operation`, `:pool`, + `:kind`, `:reason`, and `:stacktrace`. Expected pool exits are converted into + `{:error, message}` by non-bang APIs after this event is emitted. + + ### `[:urp, :connection, :retry]` + + Emitted when a worker cannot connect to soffice and schedules another + attempt. Measurements contain `:attempt` and `:delay` in milliseconds. + Metadata contains `:host`, `:port`, and `:reason`. ## Example handler diff --git a/lib/urp/test.ex b/lib/urp/test.ex index b3f4927..e8b63f7 100644 --- a/lib/urp/test.ex +++ b/lib/urp/test.ex @@ -38,8 +38,8 @@ defmodule URP.Test do ## Process allowances Stubs are scoped to the process that called `stub/1`. For processes - started with `Task` or `GenServer`, `$callers` propagation handles - this automatically. For other processes, use `allow/2`: + started with `Task`, `$callers` propagation handles this automatically. + GenServers and other independently started processes must use `allow/2`: test "async worker" do URP.Test.stub(fn _, _ -> {:ok, "pdf"} end) @@ -81,8 +81,8 @@ defmodule URP.Test do @doc """ Allow `allowed_pid` to use the stub registered by `owner_pid`. - Usually not needed — `$callers` propagation handles `Task` and `GenServer` - automatically. Use this for processes that don't propagate `$callers`. + Usually not needed for `Task`, which propagates `$callers`. Use this for + GenServers and other processes that don't propagate `$callers`. """ @spec allow(pid(), pid() | (-> pid() | [pid()])) :: :ok def allow(owner_pid \\ self(), allowed_pid) do diff --git a/release.sh b/release.sh index 177fcff..9a0a62e 100755 --- a/release.sh +++ b/release.sh @@ -2,15 +2,33 @@ set -euo pipefail usage() { - echo "Usage: $0 " + echo "Usage: $0 " >&2 + exit 1 +} + +fail() { + echo "Error: $*" >&2 exit 1 } [[ $# -eq 1 ]] || usage +[[ -z "$(git status --porcelain)" ]] || fail "working tree must be clean" +[[ "$(git branch --show-current)" == "main" ]] || fail "releases must be prepared from main" + +upstream=$(git rev-parse --abbrev-ref '@{upstream}' 2>/dev/null) || + fail "main has no upstream; fetch and configure origin/main first" +read -r behind ahead < <(git rev-list --left-right --count "${upstream}...HEAD") +[[ "$behind" == "0" && "$ahead" == "0" ]] || + fail "main must match ${upstream}; fetch, pull, and push pending commits first" bump_type="$1" -current=$(cat VERSION | tr -d '[:space:]') -IFS='.' read -r major minor patch <<< "$current" +current=$(tr -d '[:space:]' < VERSION) +[[ "$current" =~ ^([0-9]+)\.([0-9]+)\.([0-9]+)$ ]] || + fail "VERSION is not valid semantic version: ${current}" + +major=${BASH_REMATCH[1]} +minor=${BASH_REMATCH[2]} +patch=${BASH_REMATCH[3]} case "$bump_type" in patch) new="$major.$minor.$((patch + 1))" ;; @@ -19,29 +37,46 @@ case "$bump_type" in *) usage ;; esac -today=$(date -u +%Y-%m-%d) +grep --fixed-strings --quiet '## [Unreleased]' CHANGELOG.md || + fail "CHANGELOG.md has no [Unreleased] section" -echo "Bumping version: $current -> $new" +unreleased=$( + awk ' + /^## \[Unreleased\]$/ { in_section = 1; next } + in_section && /^## / { exit } + in_section { print } + ' CHANGELOG.md +) -# Update VERSION -echo "$new" > VERSION +grep --quiet '[^[:space:]]' <<< "$unreleased" || + fail "CHANGELOG.md [Unreleased] section is empty" -# Stamp changelog — replace [Unreleased] with version + date -if ! grep -q '## \[Unreleased\]' CHANGELOG.md; then - echo "Error: no [Unreleased] section in CHANGELOG.md" >&2 - git checkout VERSION - exit 1 -fi +git rev-parse --verify --quiet "refs/tags/v${new}" >/dev/null && + fail "tag v${new} already exists" + +today=$(date -u +%Y-%m-%d) tmp=$(mktemp) -sed "s/## \[Unreleased\]/## [v$new] - $today/" CHANGELOG.md > "$tmp" && mv "$tmp" CHANGELOG.md +trap 'rm -f "$tmp"' EXIT + +awk -v version="$new" -v date="$today" ' + /^## \[Unreleased\]$/ { + print + print "" + print "## [v" version "] - " date + next + } + { print } +' CHANGELOG.md > "$tmp" + +printf '%s\n' "$new" > VERSION +mv "$tmp" CHANGELOG.md +trap - EXIT -# Commit and tag git add VERSION CHANGELOG.md -git commit -m "Release v$new" -git tag -m "v$new" "v$new" - -echo "" -echo "Release v$new prepared!" -echo "" -echo "Run this to publish:" -echo " git push origin main v$new" +git commit -m "Release v${new}" +git tag --annotate --message "v${new}" "v${new}" + +echo +echo "Release v${new} prepared. Publish only after main CI succeeds:" +echo " git push origin main" +echo " git push origin v${new}" diff --git a/test/test_helper.exs b/test/test_helper.exs index 781e72c..d55e90b 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -1,19 +1,33 @@ -soffice_available? = - case :gen_tcp.connect(~c"localhost", 2002, [:binary], 1000) do - {:ok, sock} -> - :gen_tcp.close(sock) - true +integration? = System.get_env("URP_INTEGRATION") in ["1", "true"] - {:error, _} -> - false - end +excluded_tags = + if integration? do + {:ok, _apps} = Application.ensure_all_started(:urp) + + version = + case URP.version(timeout: 10_000) do + {:ok, version} -> + version + + {:error, message} -> + raise "URP_INTEGRATION is enabled but soffice is unavailable: #{message}" + end -# Tests tagged :lo26 require LibreOffice 26.2+. Excluded by default for -# local development with older soffice. CI includes them (--include lo26). -excluded_tags = if soffice_available?, do: [:lo26], else: [:integration, :lo26] + lo26? = + case Regex.run(~r/^(\d+)\.(\d+)/, version) do + [_, major, minor] -> + {major, minor} = {String.to_integer(major), String.to_integer(minor)} + major > 26 or (major == 26 and minor >= 2) -unless soffice_available? do - IO.puts("soffice not reachable on localhost:2002 — excluding integration tests") -end + _other -> + false + end + + IO.puts("running integration tests against LibreOffice #{version}") + if lo26?, do: [], else: [:lo26] + else + IO.puts("URP_INTEGRATION is not enabled — excluding integration tests") + [:integration, :lo26] + end ExUnit.start(exclude: excluded_tags) diff --git a/test/urp/pool_reconnect_test.exs b/test/urp/pool_reconnect_test.exs index fb5bf94..6db940f 100644 --- a/test/urp/pool_reconnect_test.exs +++ b/test/urp/pool_reconnect_test.exs @@ -76,4 +76,43 @@ defmodule URP.PoolReconnectTest do assert_receive {:telemetry_event, [:urp, :connection, :retry], %{attempt: 3, delay: 30}, _}, 5_000 end + + test "checkout timeout returns an error and emits start plus exception" do + port = unused_port!() + pool_name = :checkout_timeout_test + handler_id = "#{@handler_id}-call-span" + + :telemetry.attach_many( + handler_id, + [[:urp, :call, :start], [:urp, :call, :exception]], + &__MODULE__.handle_event/4, + self() + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + + _pool = + start_supervised!( + {URP.Pool, + name: pool_name, + host: "localhost", + port: port, + pool_size: 1, + backoff_initial: 10, + backoff_max: 30} + ) + + assert {:error, message} = URP.Pool.version(pool_name, timeout: 20) + assert message =~ "pool checkout failed" + + assert_receive {:telemetry_event, [:urp, :call, :start], %{system_time: system_time}, + metadata} + + assert is_integer(system_time) + assert metadata.operation == :version + + assert_receive {:telemetry_event, [:urp, :call, :exception], %{duration: duration}, metadata} + assert duration > 0 + assert metadata.kind == :exit + end end diff --git a/test/urp/protocol_test.exs b/test/urp/protocol_test.exs index 9866a2c..79bf001 100644 --- a/test/urp/protocol_test.exs +++ b/test/urp/protocol_test.exs @@ -177,6 +177,32 @@ defmodule URP.ProtocolTest do end end + describe "interface OID cache" do + setup do + previous = Process.get(:urp_oid_cache) + Process.delete(:urp_oid_cache) + + on_exit(fn -> + if previous, + do: Process.put(:urp_oid_cache, previous), + else: Process.delete(:urp_oid_cache) + end) + end + + test "stores and resolves cached interface references" do + assert P.parse_interface_reply(<<0x80>> <> P.enc_str("document-1") <> <<7::16>>) == + {:ok, "document-1"} + + assert P.parse_interface_reply(<<0x80>> <> P.enc_str("") <> <<7::16>>) == + {:ok, "document-1"} + end + + test "reports an unknown cached interface reference" do + assert P.parse_interface_reply(<<0x80>> <> P.enc_str("") <> <<9::16>>) == + {:error, "unknown cached OID index 9"} + end + end + describe "parse_string_sequence_reply/1" do test "parses empty sequence" do payload = <<0x80, 0>> @@ -229,6 +255,18 @@ defmodule URP.ProtocolTest do end end + describe "enc_count/1" do + test "uses the compact form through 254" do + assert P.enc_count(0) == <<0>> + assert P.enc_count(254) == <<254>> + end + + test "uses the extended form from 255" do + assert P.enc_count(255) == <<0xFF, 255::32>> + assert P.enc_count(65_536) == <<0xFF, 65_536::32>> + end + end + describe "request/2" do test "minimal request — no type, oid, tid" do header = P.request(5) diff --git a/test/urp/stream_test.exs b/test/urp/stream_test.exs new file mode 100644 index 0000000..b5808d5 --- /dev/null +++ b/test/urp/stream_test.exs @@ -0,0 +1,67 @@ +defmodule URP.StreamTest do + use ExUnit.Case, async: true + + alias URP.Stream, as: URPStream + + test "enumerable reader advances only when requested" do + owner = self() + + enumerable = + Stream.resource( + fn -> 0 end, + fn value -> + send(owner, {:produced, value}) + {[Integer.to_string(value)], value + 1} + end, + fn _value -> send(owner, :reader_closed) end + ) + + reader = URPStream.start_enum_reader(enumerable) + + refute_receive {:produced, _} + + send(reader, {:next, self()}) + assert_receive {:produced, 0} + assert_receive {^reader, {:chunk, "0"}} + refute_receive {:produced, 1} + + send(reader, {:next, self()}) + assert_receive {:produced, 1} + assert_receive {^reader, {:chunk, "1"}} + + assert :ok = URPStream.stop_enum_reader(reader) + assert_receive :reader_closed + refute Process.alive?(reader) + end + + test "enumerable reader reports producer failures without exiting the caller" do + enumerable = Stream.map([:chunk], fn _ -> raise "reader exploded" end) + reader = URPStream.start_enum_reader(enumerable) + + send(reader, {:next, self()}) + + assert_receive {^reader, {:error, "reader exploded"}} + refute Process.alive?(reader) + assert Process.alive?(self()) + end + + test "enumerable reader exits when its owner dies" do + parent = self() + + owner = + spawn(fn -> + reader = URPStream.start_enum_reader(Stream.cycle(["chunk"])) + send(parent, {:reader, reader}) + + receive do + :stop_owner -> :ok + end + end) + + assert_receive {:reader, reader} + ref = Process.monitor(reader) + send(owner, :stop_owner) + + assert_receive {:DOWN, ^ref, :process, ^reader, :normal} + end +end diff --git a/test/urp/telemetry_test.exs b/test/urp/telemetry_test.exs index ce2fd92..1f7fd4d 100644 --- a/test/urp/telemetry_test.exs +++ b/test/urp/telemetry_test.exs @@ -25,10 +25,14 @@ defmodule URP.TelemetryTest do assert is_integer(measurements.total_time) assert is_integer(measurements.queue_time) assert is_integer(measurements.service_time) + assert is_integer(measurements.backoff_time) assert measurements.total_time > 0 assert measurements.queue_time >= 0 assert measurements.service_time > 0 - assert measurements.total_time == measurements.queue_time + measurements.service_time + assert measurements.backoff_time >= 0 + + assert measurements.total_time == + measurements.queue_time + measurements.service_time + measurements.backoff_time assert metadata.operation == :version assert metadata.pool == URP.Pool.Default diff --git a/test/urp_test.exs b/test/urp_test.exs index e18c11c..8d06dc5 100644 --- a/test/urp_test.exs +++ b/test/urp_test.exs @@ -202,17 +202,23 @@ defmodule URPTest do describe "temp file cleanup" do test "no urp_in temp files linger after conversion" do + before = MapSet.new(Path.wildcard("/tmp/urp_in_*")) + assert {:ok, _pdf} = URP.convert({:binary, build_test_docx()}, filter: @pdf, output: :binary) - assert Enum.empty?(Path.wildcard("/tmp/urp_in_*")) + after_conversion = MapSet.new(Path.wildcard("/tmp/urp_in_*")) + assert MapSet.difference(after_conversion, before) == MapSet.new() end test "no urp_out temp files linger after conversion" do + before = MapSet.new(Path.wildcard("/tmp/urp_out_*")) + assert {:ok, _pdf} = URP.convert({:binary, build_test_docx()}, filter: @pdf, output: :binary) - assert Enum.empty?(Path.wildcard("/tmp/urp_out_*")) + after_conversion = MapSet.new(Path.wildcard("/tmp/urp_out_*")) + assert MapSet.difference(after_conversion, before) == MapSet.new() end end @@ -334,6 +340,70 @@ defmodule URPTest do end end + describe "stream I/O through the public API" do + test "file-backed stream input remains available through export" do + path = write_test_file!("docx", build_test_docx("Streamed file input")) + on_exit(fn -> File.rm(path) end) + + assert {:ok, pdf} = + URP.convert(path, + filter: @pdf, + output: :binary, + io: {:stream, :file} + ) + + assert "%PDF-" <> _ = pdf + end + + test "consecutive enumerable inputs get independent stream connections" do + docx = build_test_docx("Enumerable input") + + for _iteration <- 1..2 do + assert {:ok, pdf} = + URP.convert(to_chunks(docx, 1_024), + filter: @pdf, + output: :binary + ) + + assert "%PDF-" <> _ = pdf + end + end + + test "stream output writes a complete path result" do + path = tmp_path("pdf") + on_exit(fn -> File.rm(path) end) + + assert {:ok, ^path} = + URP.convert({:binary, build_test_docx()}, + filter: @pdf, + output: path, + io: {:file, :stream} + ) + + assert "%PDF-" <> _ = File.read!(path) + end + + test "stream output propagates sink and UNO export failures" do + assert {:error, message} = + URP.convert({:binary, build_test_docx()}, + filter: @pdf, + output: fn _chunk -> raise "sink exploded" end, + io: {:file, :stream} + ) + + assert message =~ "sink exploded" + + assert {:error, message} = + URP.convert({:binary, build_test_docx()}, + filter: "not-a-real-export-filter", + output: :binary, + io: {:file, :stream} + ) + + assert is_binary(message) + end + end + describe "pool queuing" do test "two concurrent callers queue on pool_size 1 and both succeed" do docx1 = build_test_docx("Document One") @@ -389,6 +459,26 @@ defmodule URPTest do assert {:ok, "parent stub"} = Task.await(task) end + + test "production option validation still runs when a stub is installed" do + URP.Test.stub(fn _input, _opts -> flunk("invalid calls must not reach the stub") end) + + assert_raise ArgumentError, ~r/requires the :filter option/, fn -> + URP.convert({:binary, "bytes"}) + end + + assert_raise ArgumentError, ~r/:io must be/, fn -> + URP.convert({:binary, "bytes"}, filter: @pdf, io: :invalid) + end + + assert_raise ArgumentError, ~r/:output must be/, fn -> + URP.convert({:binary, "bytes"}, filter: @pdf, output: {:invalid, :output}) + end + + assert_raise ArgumentError, ~r/:pool must be/, fn -> + URP.convert({:binary, "bytes"}, filter: @pdf, pool: "dynamic-name") + end + end end defp collect_chunks(acc \\ []) do