feat(storage): S3-backed log adapter, tenant auth boundary, per-worktree dev ports - #2
Merged
Merged
Conversation
Step 2 of the storage plan: append/query now round-trip logs through an S3-compatible endpoint via the existing Rustler object_store NIF. Objects are one NDJSON per append batch, keyed by tenant-scoped prefix so isolation holds by construction. Memory adapter stays wired as the test-only default. docker-compose.yml swaps MinIO for RustFS since MinIO's community edition is no longer maintained and RustFS is a drop-in S3-compatible replacement. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Fixes the concrete correctness/config/deploy issues Codex flagged; documents
the concerns that are deliberately deferred to step 3 (idempotency,
unbounded query work, cross-tenant auth).
- Validate tenant name even on an empty batch, so an adversarial name is
rejected on the first attempt rather than only once a record survives OTLP
decoding.
- Return {:error, {:encode_failed, _}} on Jason encode failure (e.g. a body
with non-UTF-8 bytes) instead of raising and killing the ingest process.
- Rewrite fetch_records to prepend batches and flatten once, avoiding
O(n^2) list concatenation on tenants with many objects.
- Reject empty PULSO_S3_* env values in prod (fetch_env! only guarded
against missing keys, not empty strings).
- docker-compose: add the mandatory RUSTFS_VOLUMES env var (RustFS refused
to start without it) and bind the published ports to 127.0.0.1 so the
well-known dev credentials cannot be reached from another host on the LAN.
- Add non-integration unit tests for tenant validation (including empty
batch) and the encode-failure path, so `mix test` protects against
regressions without needing RustFS running.
- Rewrite the module docstring to name the known limits of the step-2
layout (not idempotent, unbounded query, no auth here, no columnar
layout) so a future reader knows what step 3 has to fix.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Follow-up on the earlier Codex adversarial review. Every finding is either
fixed here or documented as fundamentally step-3 shaped in the module doc.
Correctness / idempotency
- Object keys are now content-addressed
(tenants/<tenant>/logs/<sort_ns>-<sha256_prefix>.ndjson). A retry with an
identical batch lands on the same key, so a lost-response retry does not
duplicate records. Truncated SHA-256 (64 bits) also drops the collision
risk of the old random 8-byte suffix.
- Attribute and resource map keys are coerced to strings on append via
`Pulso.Storage.S3.sanitize_map/1`, so a caller that hands us atom or int
keys (a future path where OTLP is not the only producer) does not silently
collide on the JSON round trip.
- Query order is now delegated to `Pulso.Storage.SortOrder` and shared by
Memory and S3. Ties break on observed_timestamp_ns, then trace_id,
span_id, body — deterministic across adapters so a client that switches
backends never sees the limit response reshuffle.
- The Rust NIF distinguishes NotFound from other errors (returns the atom
`:not_found`). `Pulso.Storage.S3.query/2` treats it as "raced with a
delete" and skips; other errors halt so an outage is never hidden.
Auth boundary
- New `Pulso.Auth` behavior with two impls:
* `Pulso.Auth.Open` (default; accepts everything, matches prior behavior
for dev/test)
* `Pulso.Auth.SharedSecret` (bearer token per tenant, hashed with
sha256, compared in constant time; required in prod)
- `PulsoWeb.OTLPController` verifies the caller before appending, returning
401 for auth failures, 400 for invalid tenant names, 500 otherwise.
- Prod runtime.exs requires `PULSO_TENANT_TOKENS` (JSON) and rejects empty
values for every PULSO_S3_* env var — same treatment as the S3 config.
Per-worktree port scoping
- Ports adapted from tuist/tuist. `mise/utilities/dev_instance_env.sh`
runs on every `mise` invocation, persists a suffix into
`.git/worktrees/<name>/pulso-dev-instance` (falling back to
`.pulso-dev-instance` in the checkout root), and exports:
* PULSO_DEV_INSTANCE
* PORT (4000 + suffix; Phoenix)
* PULSO_RUSTFS_API_PORT (9095 + suffix)
* PULSO_RUSTFS_CONSOLE_PORT (9098 + suffix)
* PULSO_S3_ENDPOINT (http://localhost:$PULSO_RUSTFS_API_PORT)
Two worktrees now run their own Phoenix and RustFS side by side without
the second one binding on top of the first.
- docker-compose.yml interpolates the RustFS host ports from these vars
(with the previous defaults as fallback).
- runtime.exs dev reads PULSO_S3_ENDPOINT — mise sets it, and the fallback
9195 matches the compose default port.
What's still deferred, deliberately
- Unbounded query work when `limit` is set. Segment min/max metadata (step
3) is required to safely early-exit; short-circuiting on key order alone
would break sort semantics because a recently-written batch can carry
old-timestamp records. The module docstring names this explicitly.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Second adversarial pass on top of 5a628e3. Highest-severity findings: CRITICAL — MCP read path bypassed the auth boundary - Pulso.MCP.dispatch/2 now takes a context; PulsoWeb.MCPController seeds it with {conn: conn} so tools that reference a tenant can call Pulso.Auth.verify/2. Pulso.MCP.Tools.call/3 does this for query_logs and returns {:error, {:unauthorized, reason}} on failure, which the MCP layer surfaces as an errored tool result. Ping, tools/list, and initialize still work without a conn (no tenant to authorize). HIGH — Auth fell open when config was missing - config/config.exs now sets an explicit default (module: Pulso.Auth.Open) for dev/test. - config/prod.exs sets a sentinel (:must_configure_at_runtime). - runtime.exs :prod overrides with Pulso.Auth.SharedSecret and its tokens. - Pulso.Auth.module/0 raises loudly on the sentinel or a missing config — a release that skips runtime.exs will not silently accept traffic. HIGH — Content-hash keys collapsed distinct-but-identical batches - append/3 now takes opts. When idempotency_key is present, the object key is deterministic (SHA-256 of tenant || key, mirroring Stripe / RFC 9457 idempotency semantics). When it is absent, the key includes a fresh random suffix so two producers with identical bytes never collide. - PulsoWeb.OTLPController reads `Idempotency-Key` and threads it through. - The Storage behaviour callback is now append/3; Memory is a no-op on the opt. MEDIUM — Invalid tenant returned 401 under shared-secret auth - Tenant name is validated in the OTLP controller BEFORE Auth.verify, so a `bad/name` tenant returns 400 regardless of the auth outcome. MEDIUM — sanitize_map silently collapsed colliding keys - Now returns {:error, {:attribute_key_collision, [key]}} on collision. OTLP-produced attributes are unaffected (string keys throughout); a hand-built Log with `%{1 => a, "1" => b}` gets a diagnostic instead of a dropped value. MEDIUM — Port suffixes 3 apart could collide - Ranges widened from 9095/9098 + suffix (overlap for suffixes 3 apart) to 9000 + suffix / 10000 + suffix (non-overlapping across the whole 100..999 range). ClickHouse on 9000 stays clear of the API range's 9100 floor. LOW — Sort ties conflated nil and empty string - Pulso.Storage.SortOrder wraps every string tiebreaker in `{presence_flag, value}` so nil sorts after every real string, including "". New test coverage: MCP query_logs auth (3 cases), OTLP invalid-tenant 400, OTLP Idempotency-Key passthrough, S3 idempotency (with vs without key), sanitize_map collision → error, nil-vs-"" sort distinction, auth misconfiguration crashes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — MCP.dispatch/1 with no context could bypass auth
- The "no conn = allow" fallback in Pulso.MCP.Tools was a bypass. Now a
missing conn falls through to an empty %Plug.Conn{}, which fails
Pulso.Auth.SharedSecret.verify/2 as :missing_token. Under
Pulso.Auth.Open (dev/test default) the empty conn still passes, so
existing tests still work.
HIGH — Retries with no observed timestamp still duplicated
- Pulso.Storage.S3.batch_sort_ns/1 now keys on `timestamp_ns` (caller-
provided, stable per request) instead of `observed_timestamp_ns` (set
fresh-now on every call). Two retries under the same Idempotency-Key
now produce the same sort_ns and land on the same object.
HIGH — SortOrder.presence_pair crashed on non-string log bodies
- Pulso.Storage.SortOrder.presence_pair/1 now coerces non-string, non-nil
bodies via inspect/1. OTLP AnyValue can produce int/bool/list bodies;
the sort layer must not crash on those. Coverage in sort_order_test.
MEDIUM — Reused idempotency key + different content silently overwrote
- Pulso.Storage.S3.object_key/4 now hashes tenant || key || content_hash
when idempotency_key is present. Same content + same key still
collapses (idempotent). Different content + same key now diverges,
which surfaces the client bug rather than losing the earlier write.
LOW — Port range collided with Node.js debugger (9229) and others
- Shifted RustFS host ports away from the crowded 9xxx band:
* API: 11000 + suffix (was 9000 + suffix)
* Console: 12000 + suffix (was 10000 + suffix)
Avoids Node debugger (9229), ClickHouse (9000), Prometheus (9090),
gRPC dev (50051), etc.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — Retries with a missing observed timestamp still duplicated - Previous fix put the payload content hash into the idempotent object key, but `normalize/1` fills `observed_timestamp_ns` with `now` before encoding, so the payload varied between retries and the key drifted. - Introduce `Pulso.Storage.S3.caller_content_hash/1`: hashes a canonical view of the pre-normalization records (`timestamp_ns`, `severity_*`, `service`, `body`, `trace_id`, `span_id`, `attributes`, `resource` — every field the caller controls, none the normalizer will set). - `object_key/4` now takes this fingerprint directly. The stored NDJSON payload still carries `observed_timestamp_ns` as ingest metadata; only the key derivation is stabilized. MEDIUM — Key format is not a public API - Add a "Key format stability" section to the S3 module docstring: any future change to hashing, delimiters, or sort-key width invalidates cross-version idempotency and needs a migration plan. LOW — Integration tests pointed at the pre-shift port - Update the PULSO_S3_ENDPOINT fallback in `test/pulso/object_store_test.exs` and `test/pulso/storage/s3_test.exs` from 9000 to 11100 to match the new docker-compose default. New unit tests: identical records produce identical fingerprints; observed_timestamp_ns does not influence the fingerprint; changing any caller-controlled field does; a non-UTF-8 body surfaces as an encode error rather than a crash. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — Caller-supplied observed_timestamp_ns was silently ignored
- The prior canonicalization excluded observed_timestamp_ns entirely so
retries whose observed_ts was filled in by the normalizer would
dedupe. That over-corrected: a caller who explicitly sets an observed
timestamp is declaring it as part of the record. Excluding it meant
two writes with different observed_ts but the same idempotency key
overwrote each other.
- caller_content_hash/1 now includes observed_timestamp_ns as the caller
provided it (nil when they didn't set it). Retries with nil-on-both
still dedupe. A distinct observed_ts produces a distinct object.
MEDIUM — Hash was not canonical across Elixir/Jason versions
- caller_content_hash/1 no longer hashes Jason-encoded bytes. Jason
serializes map keys in Map.to_list/1 order, which can shift when a
small map promotes to a hash map or a runtime upgrade changes map
layout. Instead, hash `:erlang.term_to_binary(term, [:deterministic])`
bytes — the BEAM guarantees stable key ordering with that flag from
OTP 24.1 onward. Fingerprint is now identical across runtimes, GC
cycles, and library versions.
- New test proves attributes inserted in a different order produce the
same fingerprint.
Cleanup: caller_content_hash/1 no longer returns {:error, _}. The
:erlang encoder handles any Elixir term including invalid UTF-8. Encode
errors on the stored payload still surface via `encode/1`.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Both flagged the same theme — key format is not stable across releases — so the fix is one architectural change that closes both. HIGH — Cross-commit key format changes would orphan retries after deploy - The whole branch has been iterating the key format (fingerprint algo, suffix layout, schema). If Pulso were already running in prod, each iteration would have orphaned prior objects. Since this is step 2 with no prior deployment, no migration is needed today, but the risk needs to be structurally closed. MEDIUM — `:erlang.term_to_binary(_, [:deterministic])` is stable within an OTP release, not across major OTP upgrades. A future OTP major bump could change the fingerprint bytes and thus the object keys. Fix: bake a schema version into every object key. - New path: `tenants/<tenant>/v1/logs/<sort_ns>-<suffix>.ndjson` - @schema_version constant in Pulso.Storage.S3 - Any future change to the fingerprint algorithm, sort-key width, or delimiter bumps to v2/. Old objects live at v1/, new at v2/. A reader can be taught to look at both during a migration window, and a compaction job re-keys at its own pace. - Module docstring's "Key format stability" section rewritten to reflect the versioning contract and to correct the earlier overstated cross-version stability claim on term_to_binary. - Tests: integration cleanup paths updated; unit test asserts the v1/ segment is present so a future removal is caught immediately. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — OTLP ingest silently dropped records without timeUnixNano
- Pulso.OTLP.Logs.decode/1 previously returned only the surviving records,
and PulsoWeb.OTLPController answered 200 with an empty body. A sender
whose batch contained a record with a missing/malformed timeUnixNano
received success and never retried the dropped record — silent data
loss.
- decode/1 now returns {records, rejected_count}. The controller wraps a
non-zero rejected count in an OTLP-native `partialSuccess` object per
the ExportLogsServiceResponse schema. The sender can then decide
whether to retry the batch (with client-side fixups on the rejected
records) or accept the loss.
- Empty-batch and fully-successful responses still send `{}`.
Test coverage: logs_test.exs updated to the new tuple shape and asserts
the reject counter; otlp_controller_test.exs adds a case for the
partialSuccess response.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — Logs with only observedTimeUnixNano were dropped and marked rejected
- Per the OTLP data model, `time_unix_nano` MAY be absent on a
LogRecord. The receiver should fall back to `observed_time_unix_nano`
and, if that is also absent, to the current wall clock. Rejecting
those valid records via `partialSuccess` was silent data loss —
clients are asked NOT to retry a partial-success batch.
- Pulso.OTLP.Logs.decode_log_record/3 now:
* uses `time_unix_nano` when present,
* falls back to `observed_time_unix_nano`,
* falls back to `System.system_time(:nanosecond)`.
- Only genuinely malformed entries (a `logRecords` element that is not
a JSON object) count as rejected.
Tests: two new decode cases (observed-only, both absent), one for the
non-map reject path; controller test updated to trigger reject with a
malformed entry instead of a missing timestamp.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — Zero timestamps sent every record to the Unix epoch - Per the OTLP spec, `*_unix_nano: 0` means "unknown", identical to the field being absent. The prior decoder passed the 0 through, so a record with only observedTimeUnixNano set was stored at ts=0 and vanished from time-bounded queries. - Pulso.OTLP.Logs.nano_or_nil/1 now folds 0 (integer or "0" string) to nil. HIGH — Wall-clock fallback at decode broke idempotency - The previous fix stuffed `now` into `timestamp_ns` whenever both OTLP timestamps were missing. But that value flowed into caller_content_hash AND batch_sort_ns, so two retries under the same Idempotency-Key produced different fingerprints AND different sort_ns prefixes — the second PUT landed at a distinct key and duplicated. - Split the responsibilities. The decoder preserves the caller's intent (nil for absent). Pulso.Record.Log's `timestamp_ns` is now nilable. - Storage adapters (Memory and S3) backfill on ingest: prefer the observed timestamp, then the wall clock. This runs AFTER caller_content_hash and caller_sort_ns compute their fingerprints on the pre-normalization records, so retries stay deterministic. - Rename Pulso.Storage.S3.batch_sort_ns/1 to caller_sort_ns/1 and compute it from pre-normalization records with a 0 fallback for nil timestamps. New test coverage: - OTLP.Logs: decoder preserves nil ts (both absent, observed-only, time_unix_nano=0 folds to nil). - Memory adapter: backfill from observed, backfill from wall clock. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
HIGH — Retries silently mutated the stored timestamp - `normalize/1` in both storage adapters was injecting `now` for a nil timestamp on every append. A retry under the same idempotency_key therefore landed on the same object key but overwrote the first record with a later ts. An already-acknowledged log could disappear from its original time range and re-emerge in a later one. - Storage no longer backfills timestamps at all. Records arrive verbatim; retries are truly idempotent because the payload is identical byte-for-byte. If a caller needs a wall-clock timestamp, they set it themselves at ingest. Step 3's conditional PUT will allow first-write-wins backfill safely. Bonus catch while reviewing the query path: - `filter_by_time` compared `ts >= start_ts` — Elixir's term ordering puts atoms above numbers, so `nil >= 5` was true, and nil-ts records leaked through every start_ts filter. Added an `is_integer(ts)` guard in both adapters. Tests: Memory & S3 tests updated to verify records stored verbatim; new test explicitly guards the nil-ts / time-filter behavior. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
pepicrft
marked this pull request as ready for review
September 24, 2026 05:13
Credo failed CI on two "nested too deep" refactor findings and one "prefer comparing against an empty list" warning. - Pulso.Storage.S3.sanitize_map: extract per-entry logic into `insert_sanitized/2` + `put_sanitized/3` + `sanitize_value/1` so the cond block collapses to a single-level dispatch. - Pulso.OTLP.Logs.decode_resource_logs: extract the inner scope-logs and per-record reducers into named helpers. - test/pulso/storage/s3_test.exs: replace `length(remaining) >= 1` with the cheaper `remaining != []`. Ran `mix credo` (no issues) and `mix test` (73 passed) locally.
Elixir 1.18+ ships a `JSON` module with encode!/1, encode_to_iodata!/1,
decode/1, and decode!/1. It covers everything Pulso needs from a JSON
codec, so drop the direct Jason dep and route Phoenix's `:json_library`
to `JSON`. Jason may still show up as a transitive optional dep of
Phoenix or other libraries; that is fine as long as no code in this
repo references it directly.
Callsite changes:
- lib/pulso/mcp/tools.ex, test/pulso/mcp/tools_test.exs,
lib/pulso/storage/s3.ex: swap `Jason.encode!/1`, `Jason.decode!/1` for
the JSON equivalents.
- lib/pulso/storage/s3.ex `encode/1`: Elixir's JSON only exposes bang
encoders, so wrap the call in a try/rescue to keep the
`{:ok, iodata} | {:error, {:encode_failed, reason}}` contract.
- config/runtime.exs: `Jason.decode/1` -> `JSON.decode/1` (same
`{:ok, term} | {:error, reason}` shape).
- config/config.exs: `config :phoenix, :json_library, JSON`.
- mix.exs: drop the direct `:jason` dep.
The Integration job started failing after the branch swapped MinIO for
RustFS and renamed PULSO_MINIO_* env vars to PULSO_S3_*: the workflow
was still starting MinIO on port 9000 and setting PULSO_MINIO_*
variables that no test reads, so every integration test tried to
connect to http://localhost:11100 (the docker-compose default host
port) and got an "error sending request" failure.
- Replace the ad-hoc `docker run minio` step with
`docker compose up -d rustfs`, which reuses the checked-in
docker-compose.yml and therefore stays in sync with local dev.
- Set PULSO_RUSTFS_API_PORT=11100 / PULSO_RUSTFS_CONSOLE_PORT=12100 so
the compose port interpolation picks them up without depending on the
mise-derived per-worktree suffix.
- Set PULSO_S3_ENDPOINT/BUCKET/REGION/ACCESS_KEY_ID/SECRET_ACCESS_KEY to
the values the tests actually read, using RustFS's default admin
credentials.
- Health-check loop probes the RustFS S3 endpoint on the mapped port
(accepting 200/403/404 as "listener up").
- Bucket seed uses the AWS CLI with the same env vars, pointing at
http://localhost:${PULSO_RUSTFS_API_PORT}.
Jason still shows up as a transitive optional dep of Phoenix and a few dev tools, so `mix deps.get` cannot fully remove it. Enforce "no direct Jason usage in our code" at CI time instead. - .credo.exs: enable Credo.Check.Warning.ForbiddenModule with `Jason` and a message pointing the reader at AGENTS.md. Verified locally with a throwaway `lib/jason_probe.ex` that referenced `Jason.encode!/1` -- Credo flagged it as expected. - AGENTS.md: replace the "JSON: Jason" convention line with a concrete "use Elixir's built-in JSON, direct Jason.* fails CI" note.
RustFS refuses to start when the volumes in `RUSTFS_VOLUMES` resolve to the same underlying st_dev (safety check against erasure coding across one physical disk). Local laptops and GitHub Actions runners both hit this: every `/data/rustfsN` subdirectory sits on the same block device, so the 4-volume layout that RustFS ships in its own compose example fails immediately. Switch to single-volume `/data/rustfs0`. That matches how Pulso uses RustFS in dev and CI (single-node, single disk), avoids the erasure path, and does not require the RUSTFS_UNSAFE_BYPASS_DISK_CHECK escape hatch.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changed
Step 2 of the storage plan: Pulso now round-trips logs through a real S3-compatible object store, not just in-process memory. The change also introduces the ingest-side authentication boundary, per-worktree port scoping, and a bunch of correctness plumbing needed to keep the write path idempotent.
Concrete additions:
Pulso.Storage.S3writes one NDJSON object per append batch under a tenant-scoped, schema-versioned prefix (tenants/<tenant>/v1/logs/<sort_ns>-<suffix>.ndjson).Pulso.Storage.Memorystays wired as the test-only default; the OTP supervisor only starts it when the configured adapter is Memory (or nothing is configured, as inmix test).Pulso.Storage.append/3acceptsidempotency_key, andPulsoWeb.OTLPControllerforwards theIdempotency-Keyheader. Same key with the same content deduplicates; same key with different content produces a distinct object rather than overwriting; no key means retries are free to duplicate. The fingerprint is a deterministic hash of the caller's pre-normalization records via:erlang.term_to_binary(_, [:deterministic]), so retries stay byte-identical.Pulso.Authbehaviour with two shipping implementations:Pulso.Auth.Open(default for dev and test) andPulso.Auth.SharedSecret(Bearer token per tenant, sha256-hashed, constant-time compared).config/prod.exssets a:must_configure_at_runtimesentinel so a release that skipsruntime.exsraises loudly rather than falling open.Pulso.MCP.dispatch/2andPulso.MCP.Tools.call/3.query_logscallsPulso.Auth.verify/2before touching storage, closing what would otherwise have been a read-side bypass of the ingest auth boundary.Pulso.Storage.SortOrderis a shared canonical order (timestamp_ns desc, then observed_timestamp_ns desc, then trace_id, span_id, body) used by both adapters. Nil sorts distinctly from empty string via a presence-flag pair.mise/utilities/dev_instance_env.shis adapted from tuist/tuist to scope development ports per git worktree. It persists a 3-digit suffix inside.git/worktrees/<name>/pulso-dev-instance(or.pulso-dev-instanceat the checkout root as a fallback) and exportsPORT,PULSO_RUSTFS_API_PORT,PULSO_RUSTFS_CONSOLE_PORT, andPULSO_S3_ENDPOINT. Two worktrees can now run Phoenix and RustFS side by side without fighting over the same TCP ports.docker-compose.ymlswaps MinIO for RustFS. MinIO's community edition is no longer actively maintained, and RustFS is a drop-in that keeps the S3 API surface identical. Host ports are interpolated from the mise-derived env vars, bound to127.0.0.1only, and the mandatoryRUSTFS_VOLUMESenv var is set.native/pulso_object_store) now distinguishesNotFoundfrom other errors on the way back to Elixir, returning the atom:not_foundfor the former.Pulso.Storage.S3.query/2treats it as "raced with a delete" and skips the object rather than aborting the query.time_unix_nanoandobserved_time_unix_nanomay be absent, and a value of 0 explicitly means "unknown". The decoder folds 0 to nil, preserves the caller's intent verbatim, and the OTLP receiver surfaces malformed entries (alogRecordselement that is not a JSON object) viaExportLogsPartialSuccess.rejected_log_recordsper the OTLP HTTP spec.nowinto a nil timestamp would drift across retries and defeat idempotent writes. Time-bounded queries explicitly guard against nil timestamps to avoid Elixir's term-ordering trap wherenil >= 5returns true.Why
Step 1 landed the ingest and query spine end to end against an in-memory adapter to prove the shape of
%Pulso.Record.Log{}, thePulso.Storagebehaviour, and the MCP transport. That shape is now stable enough to be worth writing against real object storage, which is the point of this branch.Step 2 is deliberately not the final storage engine. The next step, step 3, replaces the flat NDJSON layout with columnar segments and a manifest that supports conditional writes. What step 2 needs to prove is only that Pulso can round-trip logs through S3 via the existing Rustler NIF, and that the surface can grow the idempotency, auth, and observability contracts that step 3 will build on. Doing the storage adapter and the auth boundary in the same branch was necessary because the write path and the read path share the same tenant surface, and adding auth after the storage adapter had bedded in would have left a bypass window in the MCP query tool.
Approach
A few choices worth explaining, since they trade off against alternatives:
Idempotency-Key, and shipping the header without a matching guarantee in the storage layer would have been dishonest. The chosen shape gives real retry safety today without needing conditional PUT, at the cost of accepting that two byte-identical batches from distinct producers under the same key are treated as the same write.:erlang.term_to_binary(_, [:deterministic])for the fingerprint.Jason.encode/1serializes maps inMap.to_list/1order, which is not canonical and can shift when a small map promotes to a hash map. The Erlang canonical form sorts map keys before encoding and is stable within an OTP release. It is not guaranteed stable across a major OTP upgrade, which is why the object key path carries av1/schema-version segment: a future format bump goes tov2/, and a compaction job migrates at its own pace.Pulso.Auth.SharedSecretrather than a real accounts service. Full tenant identity belongs at a higher layer that does not exist yet.SharedSecretgives us a real auth surface (constant-time compared, per-tenant, hashed at rest) that closes the biggest correctness gap without inventing infrastructure Pulso will eventually rip out.nowdiffers between the first call and the retry, so the second PUT overwrites the same object with a later timestamp, and an already-acknowledged log migrates to a different time range. Step 3's conditional PUT will allow first-write-wins backfill safely; until then, storing what the caller sent is the honest option.Impact
Pulso.Storage.S3is the active adapter outsidemix test.runtime.exspicks up config fromPULSO_S3_*env vars, and prod fails fast if any of them are missing or empty.PULSO_TENANT_TOKENS, a JSON object mapping tenant name to"sha256$<hex>". Deployments compute the digest offline and never store plaintext tokens in env.docker compose up -dfor the RustFS container. The host ports come from the per-worktree env vars, so plaindocker compose uppicks them up as long as mise has been sourced..git/worktrees/<name>/pulso-dev-instance, and Phoenix, the RustFS S3 API, and the RustFS console all bind to distinct ports.partialSuccessblock whenever the batch contained a malformed record. Compliant senders will not retry the batch on this response, which is exactly what we want because the drop was permanent and retrying would flood the receiver with the same rejected records.Validation
Ran locally, from the branch tip:
mix compile --warnings-as-errorsclean.mix precommitclean (formatter, unused-deps check, tests).mix test: 73 passed, 15 excluded. The excluded ones are tagged:integrationand need a running RustFS viadocker compose up -d. They cover the S3 adapter and the object-store NIF end to end.mix phx.serverand confirmed the Phoenix endpoint binds to the mise-derived port (4XXXwhere XXX is the worktree suffix).POST /mcpresponds to a JSON-RPCping.GET /returns 404 as designed.mise env:PORT,PULSO_RUSTFS_API_PORT, andPULSO_RUSTFS_CONSOLE_PORTall derive from the same suffix and never overlap.Integration tests were not run in this session because the local Docker daemon is not installed on the development machine. The Elixir to Rust NIF path was smoke-tested against an unrelated local S3 API that returned structured errors, which confirmed the request path is wired correctly. To run the integration tests locally:
docker compose up -d && PULSO_INTEGRATION=1 mix test --only integration.Known limits, deliberately deferred to step 3
Called out explicitly in the
Pulso.Storage.S3module docstring so a future reader knows what step 3 has to close:limitis set. Without per-batch time metadata this adapter cannot safely skip objects, solimit: 1on a large tenant still lists and downloads every object before filtering. Step 3's segment manifest carries min/maxtimestamp_nsper segment and lets the query short-circuit.