From f89917ffcf4dcff297b1e7be717e3aad09a1270f Mon Sep 17 00:00:00 2001 From: AlbertoV Date: Mon, 19 Oct 2026 22:37:13 +0200 Subject: [PATCH] feat(laravel)!: deliver audit events through Queen only Audit events are pushed to a Queen MQ queue that the Traceflow worker reads. The transactional outbox from 2.8.0 is removed with its migrations, the publish/status/retry commands, the require_transaction setting and the synchronous HTTP delivery. AuditPublisher pushes one partition per subject with the event ID as transaction ID. Inside a transaction the push waits for the commit and a rollback sends nothing; a failure after the write is reported, not thrown. A short timeout and a cooldown after a failure keep an outage from slowing every request. Co-Authored-By: Claude Sonnet 4.5 --- README.md | 4 +- php/laravel/README.md | 126 +++----- php/laravel/composer.json | 5 +- php/laravel/config/traceflow.php | 29 +- .../create_traceflow_audit_outbox.php.stub | 23 -- .../index_traceflow_audit_outbox.php.stub | 24 -- php/laravel/src/Audit/AuditClient.php | 74 +---- php/laravel/src/Audit/AuditOutbox.php | 40 --- php/laravel/src/Audit/AuditPublisher.php | 149 +++++++++ php/laravel/src/Audit/LogsAudit.php | 6 +- .../src/Console/AuditStatusCommand.php | 34 -- .../src/Console/PublishAuditCommand.php | 62 ---- php/laravel/src/Console/RetryAuditCommand.php | 23 -- php/laravel/src/TraceFlowSDK.php | 7 - php/laravel/src/TraceFlowServiceProvider.php | 11 +- php/laravel/tests/Unit/AuditOutboxTest.php | 226 ------------- php/laravel/tests/Unit/AuditPublisherTest.php | 300 ++++++++++++++++++ php/laravel/tests/Unit/AuditTest.php | 67 ---- .../tests/audit-installed-framework-smoke.php | 58 +++- 19 files changed, 570 insertions(+), 698 deletions(-) delete mode 100644 php/laravel/database/migrations/create_traceflow_audit_outbox.php.stub delete mode 100644 php/laravel/database/migrations/index_traceflow_audit_outbox.php.stub delete mode 100644 php/laravel/src/Audit/AuditOutbox.php create mode 100644 php/laravel/src/Audit/AuditPublisher.php delete mode 100644 php/laravel/src/Console/AuditStatusCommand.php delete mode 100644 php/laravel/src/Console/PublishAuditCommand.php delete mode 100644 php/laravel/src/Console/RetryAuditCommand.php delete mode 100644 php/laravel/tests/Unit/AuditOutboxTest.php create mode 100644 php/laravel/tests/Unit/AuditPublisherTest.php diff --git a/README.md b/README.md index 46d241c..a726315 100644 --- a/README.md +++ b/README.md @@ -10,8 +10,8 @@ Clients for [Traceflow](https://github.com/smartpricing/traceflow-service). | Java | See module configuration | [java/core/README.md](java/core/README.md) | Native audit logging is implemented in the Node and Laravel SDKs. Laravel includes -model capture and a transactional outbox. See each package's README for setup, -supported transports and tests. +model capture and delivery through a Queen MQ queue. See each package's README for +setup, supported transports and tests. The Laravel package is published from the [split repository](https://github.com/smartpricing/traceflow-laravel). diff --git a/php/laravel/README.md b/php/laravel/README.md index de52846..cb21524 100644 --- a/php/laravel/README.md +++ b/php/laravel/README.md @@ -30,16 +30,22 @@ See [config/traceflow.php](config/traceflow.php) for all settings. ## Audit model changes -Deploy the backend audit API and apply its CQL schema first. Publish the outbox -migration and run it on the same database connection as the audited models: +Audit events travel through a [Queen MQ](https://queenmq.com) queue: the SDK +pushes each event, and the Traceflow worker reads the queue and stores it. Enable +the consumer on the service first (`AUDIT_QUEEN_ENABLED`, see the +[service README](https://github.com/smartpricing/traceflow-service#delivery-through-queen)) +and install the Queen client in the application: ```sh -php artisan vendor:publish --tag=traceflow-audit-migrations -php artisan migrate +composer require queen-mq/php-client ``` -For a named connection, use `php artisan migrate --database=business`. No migration -runs automatically when the package is installed. +```dotenv +TRACEFLOW_AUDIT_QUEEN_URL=http://queen-mq.queen.svc.cluster.local:6632 +``` + +The URL and token default to the application's `QUEEN_URL` and +`QUEEN_BEARER_TOKEN`. No table or scheduled command is needed. Add `LogsAudit` and an explicit field allowlist to a model: @@ -68,7 +74,7 @@ Free text (`description`, `actor.label`, `redacted_fields`) and nested objects a validated against the backend contract when the event is created, so an invalid envelope fails the write instead of the later publish. -For this model on the default connection, update it inside a transaction: +Update the model as usual; inside a transaction, the event leaves after the commit: ```php use Illuminate\Support\Facades\DB; @@ -87,75 +93,30 @@ app(AuditContext::class)->run([ }); ``` -The trait captures create, update, delete and restore events. Saves, deletes and -restores require an already-open transaction on the model's connection. The model -write and outbox insert roll back together. Updates with no changes to selected -fields are skipped; before/after images are captured before publishing. +The trait captures create, update, delete and restore events. Updates with no +changes to selected fields are skipped; before/after images are captured at the +write. + +Inside a transaction, the events are pushed once the outermost transaction on the +model's connection commits; a rollback pushes nothing. Outside a transaction, they +are pushed right after the write. Each subject has its own partition, so Traceflow +stores its events in order, and the event ID is the push's transaction ID, so a +repeated push is stored once. + +The push happens after the write has committed. If Queen does not accept it, the +SDK reports an `AuditDeliveryException` to the application's exception handler and +the write stays: that event is lost. Watch those reports. A push waits at most +`TRACEFLOW_AUDIT_QUEEN_TIMEOUT_MS` (default `2000`) per attempt, with +`TRACEFLOW_AUDIT_QUEEN_RETRY_ATTEMPTS` (default `1`) attempts. After a failure, pushes +from the same process fail at once for `TRACEFLOW_AUDIT_QUEEN_COOLDOWN_SECONDS` +(default `30`, `0` disables it), so an outage does not slow every request or job item; +those events are reported and lost too. `AuditContext::run()` scopes context to the callback and restores the previous context even on an exception. The trait uses the authenticated actor when none is supplied. Pass trusted context explicitly into queued jobs. Override `customizeAudit(array $event): array` to redact or enrich an event. -### Writes outside transactions - -An application that saves models outside transactions can relax the requirement, -for example while it replaces an activity log written from model events: - -```dotenv -TRACEFLOW_AUDIT_REQUIRE_TRANSACTION=false -``` - -Writes inside a transaction stay atomic. A write outside a transaction stores the -outbox row right after the model write, so a crash or database error between the -two writes loses that event. An outbox error is then raised after the model write -has already been committed. `AuditOutbox::record()` follows the same setting, and -`AuditOutbox::requireTransaction()` always enforces a transaction. - -### Publish the outbox - -Run this command periodically from the application's scheduler or job runner: - -```sh -php artisan traceflow:audit-publish --limit=100 --max-seconds=30 --json -``` - -For a named database, add `--connection=business`. Each invocation processes at -most the requested number of ready rows and exits. Choose the frequency and -batch size for the expected volume. `--max-seconds` bounds delivery attempts and -HTTP timeouts for one invocation (default 30, maximum 300); database queries still -use the application's database timeout. A transient delivery failure stops the -batch to avoid retrying every row during an outage. Remaining rows stay pending. -`--json` reports published, retry-scheduled and failed counts. A delivery failure -or exhausted time budget returns a nonzero exit code. - -A row is deleted only after Traceflow confirms persistence. Transient failures -retain the same event and set a later `available_at`. Missing credentials and -HTTP `401`/`403`/`404` also retain pending rows: fixing publisher configuration or -deploying the route must not require reconstructing audit events. Those HTTP -responses end the current delivery attempt without immediate HTTP retries. Permanent failures set -`failed_at` and retain the row for review. After fixing the cause, run -`php artisan traceflow:audit-retry EVENT_ID`. It preserves the original payload, -ID and attempt history. Investigate `409` conflicts before resending: the same -event ID cannot describe different data. - -Concurrent publishers may send duplicates; Traceflow deduplicates by source and -event ID. Audit history is read from Traceflow. The outbox holds pending or failed -deliveries, not the history shown to users. - -Monitor the publisher separately from application request health: - -```sh -php artisan traceflow:audit-status --max-age=300 --max-pending=10000 -``` - -The command emits JSON with pending/ready/failed counts and oldest pending age. -It exits nonzero when there are failed rows or the age/backlog thresholds are -exceeded. Feed this result and publisher exit codes into your scheduler's alerting. -Use `--connection` consistently for capture, publishing, status and retries. -Publish and apply the additional `index_traceflow_audit_outbox` migration when -upgrading an existing installation; it indexes ready rows and pending age. - ### Manual capture Bulk updates/deletes, raw SQL, pivot changes and muted Eloquent events bypass @@ -166,7 +127,7 @@ For example, with an existing `quotes` table on the default connection: use Illuminate\Support\Facades\DB; use Smartness\TraceFlow\Audit\AuditClient; use Smartness\TraceFlow\Audit\AuditEvent; -use Smartness\TraceFlow\Audit\AuditOutbox; +use Smartness\TraceFlow\Audit\AuditPublisher; $db = DB::connection(); $db->transaction(function () use ($db) { @@ -190,19 +151,19 @@ $db->transaction(function () use ($db) { 'changes' => $changes, 'metadata' => ['reason' => 'Discount agreed'], ]); - app(AuditOutbox::class)->record($event, $db); + app(AuditPublisher::class)->publishAfterCommit($event, $db); }); ``` Manual events take their context from the supplied payload. Obtain the actual before/after values within the operation; account for locking, casts and database -triggers in the integrating application. Do not reconstruct past values in the -publisher by reloading the current model. +triggers in the integrating application. Do not reconstruct past values later by +reloading the current model. -`AuditClient::record($event)` sends directly and returns a persistence receipt or -throws `AuditDeliveryException`. `recordAudit($input)` creates and sends an event -in one call; it is also available on `TraceFlowSDK`. Direct sending alone does -not recover an event lost between business commit and a process crash. +`publishAfterCommit()` follows the trait's rules: it waits for the commit and +reports failures. `publish($event)` pushes at once and throws +`AuditDeliveryException` when Queen does not accept the event, for callers that +handle the failure themselves. Both do nothing while `TRACEFLOW_ENABLED=false`. Subject types and IDs are strings. Missing fields differ from `null`; use decimal strings for exact money values. Creation uses an empty before image, deletion an @@ -248,14 +209,13 @@ its own middleware. This propagates trace context, not `AuditContext`. | `TRACEFLOW_TRANSPORT=http` | Trace events use HTTP; `log` writes to Laravel's log channel. Kafka is not implemented in the PHP SDK. | | `TRACEFLOW_ASYNC_HTTP=true` | Default for tracing. Pending requests are flushed during shutdown; abrupt termination can lose them. | | `TRACEFLOW_SILENT_ERRORS=true` | Default for tracing delivery failures. Configuration and input errors can still throw. | -| `TRACEFLOW_ENABLED=false` | Drops trace events. Audit delivery throws a retryable error, leaving outbox rows pending. | +| `TRACEFLOW_ENABLED=false` | Drops trace events and audit events. Audit envelopes are still validated. | | `TRACEFLOW_TIMEOUT` | HTTP timeout in seconds; default `5`. | | `TRACEFLOW_MAX_RETRIES` | Default `3`. | | `TRACEFLOW_RETRY_DELAY` | Initial retry delay in milliseconds; default `1000`. | -Audit always uses synchronous HTTP and explicit errors, regardless of the tracing -transport, `async_http` or `silent_errors`. Disabling delivery does not disable -model capture or its transaction requirement. +Audit events always go through Queen, regardless of the tracing transport, +`async_http` or `silent_errors`. For local trace logs, set `TRACEFLOW_TRANSPORT=log` and choose `TRACEFLOW_LOG_CHANNEL` / `TRACEFLOW_LOG_LEVEL`. These settings do not redirect @@ -276,7 +236,7 @@ composer format:check Run the audit tests without an application database: ```sh -vendor/bin/phpunit tests/Unit/AuditTest.php tests/Unit/AuditOutboxTest.php +vendor/bin/phpunit tests/Unit/AuditTest.php tests/Unit/AuditPublisherTest.php ``` Optional compatibility check against installed Laravel classes, using isolated diff --git a/php/laravel/composer.json b/php/laravel/composer.json index bc22999..b32e6be 100644 --- a/php/laravel/composer.json +++ b/php/laravel/composer.json @@ -2,7 +2,7 @@ "name": "smartness/traceflow-laravel", "description": "Production-ready distributed tracing SDK for Laravel with event-sourced architecture, async HTTP transport, and cross-service context propagation", "type": "library", - "version": "2.8.0", + "version": "2.9.0", "keywords": [ "laravel", "tracing", @@ -46,7 +46,8 @@ "mockery/mockery": "^1.6" }, "suggest": { - "enqueue/rdkafka": "Required for Kafka transport support (^0.10)" + "enqueue/rdkafka": "Required for Kafka transport support (^0.10)", + "queen-mq/php-client": "Required to deliver audit events to Traceflow (^2.4)" }, "autoload": { "psr-4": { diff --git a/php/laravel/config/traceflow.php b/php/laravel/config/traceflow.php index 386a6bf..8727fc7 100644 --- a/php/laravel/config/traceflow.php +++ b/php/laravel/config/traceflow.php @@ -106,22 +106,29 @@ /* |-------------------------------------------------------------------------- - | Audit Configuration + | Audit Delivery |-------------------------------------------------------------------------- | - | require_transaction: when true (default), audited model writes and manual - | outbox records must run inside an explicit transaction, so the business - | write and its outbox row commit or roll back together. + | Audit events are pushed to a Queen MQ queue that the Traceflow worker + | reads (AUDIT_QUEEN_ENABLED on the service). Requires queen-mq/php-client. | - | When false, a write outside a transaction stores the outbox row right - | after the model write. A crash or database error between the two writes - | loses that event. Writes inside a transaction stay atomic in both modes. - | Use false to adopt auditing in an application that saves models outside - | transactions, with the same guarantee as an activity log written from - | model events. + | A model write inside a transaction publishes after the commit, and a + | rolled-back write publishes nothing. A push that fails after the write + | has committed is reported to the exception handler: that event is lost. + | The URL and token default to the application's QUEEN_URL and + | QUEEN_BEARER_TOKEN. A push waits at most timeout_ms per attempt. After a + | failure, pushes from the same process fail at once for cooldown_seconds + | (0 disables it), so an outage does not slow every request. | */ 'audit' => [ - 'require_transaction' => env('TRACEFLOW_AUDIT_REQUIRE_TRANSACTION', true), + 'queen' => [ + 'url' => env('TRACEFLOW_AUDIT_QUEEN_URL', env('QUEEN_URL')), + 'bearer_token' => env('TRACEFLOW_AUDIT_QUEEN_BEARER_TOKEN', env('QUEEN_BEARER_TOKEN')), + 'queue' => env('TRACEFLOW_AUDIT_QUEEN_QUEUE', 'traceflow.audit'), + 'timeout_ms' => env('TRACEFLOW_AUDIT_QUEEN_TIMEOUT_MS', 2000), + 'retry_attempts' => env('TRACEFLOW_AUDIT_QUEEN_RETRY_ATTEMPTS', 1), + 'cooldown_seconds' => env('TRACEFLOW_AUDIT_QUEEN_COOLDOWN_SECONDS', 30), + ], ], ]; diff --git a/php/laravel/database/migrations/create_traceflow_audit_outbox.php.stub b/php/laravel/database/migrations/create_traceflow_audit_outbox.php.stub deleted file mode 100644 index 248d9c2..0000000 --- a/php/laravel/database/migrations/create_traceflow_audit_outbox.php.stub +++ /dev/null @@ -1,23 +0,0 @@ -uuid('event_id')->primary(); - $table->longText('payload'); - $table->unsignedInteger('attempts')->default(0); - $table->timestamp('created_at'); - $table->timestamp('available_at')->index(); - $table->timestamp('failed_at')->nullable(); - $table->text('last_error')->nullable(); - }); - } - - public function down(): void { Schema::dropIfExists('traceflow_audit_outbox'); } -}; diff --git a/php/laravel/database/migrations/index_traceflow_audit_outbox.php.stub b/php/laravel/database/migrations/index_traceflow_audit_outbox.php.stub deleted file mode 100644 index 10f9ab9..0000000 --- a/php/laravel/database/migrations/index_traceflow_audit_outbox.php.stub +++ /dev/null @@ -1,24 +0,0 @@ -index(['failed_at', 'available_at'], 'audit_outbox_ready'); - $table->index(['failed_at', 'created_at'], 'audit_outbox_age'); - }); - } - - public function down(): void - { - Schema::table('traceflow_audit_outbox', function (Blueprint $table) { - $table->dropIndex('audit_outbox_ready'); - $table->dropIndex('audit_outbox_age'); - }); - } -}; diff --git a/php/laravel/src/Audit/AuditClient.php b/php/laravel/src/Audit/AuditClient.php index a82a612..6ec091a 100644 --- a/php/laravel/src/Audit/AuditClient.php +++ b/php/laravel/src/Audit/AuditClient.php @@ -2,83 +2,13 @@ namespace Smartness\TraceFlow\Audit; -use GuzzleHttp\Client; -use GuzzleHttp\ClientInterface; -use GuzzleHttp\Exception\GuzzleException; - final class AuditClient { - public function __construct(private readonly array $config, private readonly ?ClientInterface $http = null) {} + public function __construct(private readonly array $config) {} + /** A validated envelope for this application's source; deliver it with AuditPublisher. */ public function createEvent(array $input): AuditEvent { return AuditEvent::create($this->config['source'] ?? '', $input); } - - public function recordAudit(array $input): array - { - return $this->record($this->createEvent($input)); - } - - /** Synchronous durable receipt; never uses the tracing silent_errors or async transport. */ - public function record(AuditEvent $event, ?float $budgetSeconds = null): array - { - if (($this->config['enabled'] ?? true) === false) { - throw new AuditDeliveryException('Audit delivery is disabled', $event->id(), true); - } - if (empty($this->config['endpoint']) || empty($this->config['api_key'])) { - throw new AuditDeliveryException('Audit requires endpoint and API key', $event->id(), true); - } - if ($event->source() !== ($this->config['source'] ?? null)) { - throw new AuditDeliveryException('Audit event source does not match publisher source', $event->id(), false); - } - $deadline = $budgetSeconds === null ? null : hrtime(true) / 1e9 + max(0, $budgetSeconds); - $http = $this->http ?? new Client; - $retries = min(10, max(0, (int) ($this->config['max_retries'] ?? 3))); - for ($attempt = 0; ; $attempt++) { - $remaining = $deadline === null ? INF : $deadline - hrtime(true) / 1e9; - if ($remaining <= 0) { - throw new AuditDeliveryException('Audit delivery deadline exceeded', $event->id(), true); - } - $delay = min(30000, ($this->config['retry_delay'] ?? 250) * (2 ** $attempt)); - try { - $response = $http->request('POST', rtrim($this->config['endpoint'], '/').'/api/v1/audit-events', [ - 'headers' => ['X-API-Key' => $this->config['api_key'], 'Content-Type' => 'application/json'], - 'body' => $event->toJson(), - 'timeout' => min(max(0.001, (float) ($this->config['timeout'] ?? 10)), $remaining), - 'connect_timeout' => min(5.0, $remaining), - 'allow_redirects' => false, - 'http_errors' => false, - ]); - $status = $response->getStatusCode(); - if ($status >= 200 && $status < 300) { - $receipt = json_decode((string) $response->getBody(), true); - if (($receipt['status'] ?? null) !== 'persisted' || ($receipt['event_id'] ?? null) !== $event->id()) { - throw new AuditDeliveryException('Invalid audit persistence receipt', $event->id(), true); - } - - return $receipt; - } - $retryAfter = $response->getHeaderLine('Retry-After'); - if ($retryAfter !== '') { - $seconds = is_numeric($retryAfter) ? (float) $retryAfter : max(0, (strtotime($retryAfter) ?: time()) - time()); - $delay = max($delay, min(30000, $seconds * 1000)); - } - throw new AuditDeliveryException("Audit request failed ({$status})", $event->id(), $status >= 500 || in_array($status, [401, 403, 404, 408, 429], true), $status); - } catch (GuzzleException $exception) { - $failure = new AuditDeliveryException('Audit request failed or timed out', $event->id(), true); - } catch (AuditDeliveryException $exception) { - $failure = $exception; - } - // Authentication/routing can recover after configuration or deployment changes. - // Retain the outbox row, but do not repeat identical requests in this invocation. - if (! $failure->retryable || $attempt >= $retries || in_array($failure->statusCode, [401, 403, 404], true)) { - throw $failure; - } - if ($deadline !== null && hrtime(true) / 1e9 + $delay / 1000 >= $deadline) { - throw $failure; - } - usleep((int) ($delay * 1000)); - } - } } diff --git a/php/laravel/src/Audit/AuditOutbox.php b/php/laravel/src/Audit/AuditOutbox.php deleted file mode 100644 index ed4b15f..0000000 --- a/php/laravel/src/Audit/AuditOutbox.php +++ /dev/null @@ -1,40 +0,0 @@ -transactionLevel() < 1) { - throw new LogicException('Audit requires an explicit transaction on the subject/outbox connection'); - } - } - - /** Whether audited writes must run inside a transaction (traceflow.audit.require_transaction). */ - public static function transactionRequired(): bool - { - return (bool) config('traceflow.audit.require_transaction', true); - } - - /** Enforces the configured policy. Without the requirement, an open transaction is still joined. */ - public static function guardTransaction(ConnectionInterface $connection): void - { - if (self::transactionRequired()) { - self::requireTransaction($connection); - } - } - - public function record(AuditEvent $event, ConnectionInterface $connection): void - { - self::guardTransaction($connection); - $connection->table('traceflow_audit_outbox')->insert([ - 'event_id' => $event->id(), 'payload' => $event->toJson(), - 'created_at' => now(), 'available_at' => now(), 'attempts' => 0, - ]); - } -} diff --git a/php/laravel/src/Audit/AuditPublisher.php b/php/laravel/src/Audit/AuditPublisher.php new file mode 100644 index 0000000..9f0dd20 --- /dev/null +++ b/php/laravel/src/Audit/AuditPublisher.php @@ -0,0 +1,149 @@ + $config the traceflow configuration + * @param (Closure(array): object)|null $clientFactory builds the Queen client + */ + public function __construct(private readonly array $config, private readonly ?Closure $clientFactory = null) {} + + /** + * Push the event now. Throws AuditDeliveryException when Queen does not accept it. + * Does nothing while Traceflow is disabled (TRACEFLOW_ENABLED=false). + * + * After a failure, pushes from this process fail at once for cooldown_seconds, so an + * outage does not hold every request or job item for the push timeout. + */ + public function publish(AuditEvent $event): void + { + if (($this->config['enabled'] ?? true) === false) { + return; + } + $settings = $this->settings(); + if ($this->unavailableUntil !== null && microtime(true) < $this->unavailableUntil) { + throw new AuditDeliveryException('Queen failed a recent audit push; this event was not sent', $event->id(), true); + } + try { + $client = $this->client($settings); + } catch (Throwable $error) { + throw new AuditDeliveryException($error->getMessage(), $event->id(), false); + } + try { + $client->queue((string) ($settings['queue'] ?? 'traceflow.audit')) + ->partition(self::partitionFor($event)) + ->push([['transactionId' => $event->id(), 'data' => $event->jsonSerialize()]]) + ->execute(); + } catch (Throwable $error) { + $cooldown = max(0, (int) ($settings['cooldown_seconds'] ?? 30)); + $this->unavailableUntil = $cooldown > 0 ? microtime(true) + $cooldown : null; + + throw new AuditDeliveryException('Queen did not accept the audit event: '.$error->getMessage(), $event->id(), true); + } + $this->unavailableUntil = null; + } + + /** + * Push the event once the open transaction on the connection commits, or now + * without one. A rolled-back transaction publishes nothing. The business write + * has already committed when the push runs, so a failure is reported to the + * exception handler instead of thrown: that event is lost. + */ + public function publishAfterCommit(AuditEvent $event, ?ConnectionInterface $connection = null): void + { + $connection ??= DB::connection(); + $deliver = function () use ($event): void { + try { + $this->publish($event); + } catch (Throwable $error) { + self::report($error); + } + }; + if ($connection instanceof Connection && $connection->transactionLevel() > 0) { + $connection->afterCommit($deliver); + + return; + } + $deliver(); + } + + /** One partition per subject; long keys are hashed to fit the broker's name budget. */ + public static function partitionFor(AuditEvent $event): string + { + $subject = $event->jsonSerialize()->subject ?? null; + if (! is_object($subject)) { + return 'Default'; + } + $key = $subject->type.':'.$subject->id; + + return strlen($key) <= self::MAX_PARTITION_BYTES ? $key : 'sha256:'.hash('sha256', $key); + } + + /** @return array */ + private function settings(): array + { + $settings = $this->config['audit']['queen'] ?? []; + + return is_array($settings) ? $settings : []; + } + + /** @param array $settings */ + private function client(array $settings): object + { + if ($this->client !== null) { + return $this->client; + } + if ($this->clientFactory !== null) { + return $this->client = ($this->clientFactory)($settings); + } + if (! class_exists(Queen::class)) { + throw new RuntimeException('Audit delivery requires the queen-mq/php-client package'); + } + $urls = array_values(array_filter(array_map('trim', explode(',', (string) ($settings['url'] ?? ''))))); + if ($urls === []) { + throw new RuntimeException('Audit delivery requires TRACEFLOW_AUDIT_QUEEN_URL or QUEEN_URL'); + } + + return $this->client = new Queen([ + 'urls' => $urls, + 'bearerToken' => ($settings['bearer_token'] ?? null) ?: null, + // A push runs on the request path: fail fast rather than hold the request. + 'timeoutMillis' => max(100, (int) ($settings['timeout_ms'] ?? 2000)), + 'retryAttempts' => max(1, (int) ($settings['retry_attempts'] ?? 1)), + ]); + } + + private static function report(Throwable $error): void + { + if (function_exists('report')) { + report($error); + + return; + } + error_log('[TraceFlow] '.$error->getMessage()); + } +} diff --git a/php/laravel/src/Audit/LogsAudit.php b/php/laravel/src/Audit/LogsAudit.php index 1d621cf..f87cfdc 100644 --- a/php/laravel/src/Audit/LogsAudit.php +++ b/php/laravel/src/Audit/LogsAudit.php @@ -28,11 +28,9 @@ public function customizeAudit(array $event): array public static function bootLogsAudit(): void { static::saving(function ($model) { - AuditOutbox::guardTransaction($model->getConnection()); $model->traceflowAuditBefore = $model->auditSnapshot(true); }); static::deleting(function ($model) { - AuditOutbox::guardTransaction($model->getConnection()); $model->traceflowAuditBefore = $model->auditSnapshot(); }); static::created(fn ($model) => $model->writeAudit('created', [], $model->auditSnapshot())); @@ -44,7 +42,6 @@ public static function bootLogsAudit(): void static::deleted(fn ($model) => $model->writeAudit('deleted', $model->traceflowAuditBefore, [])); if (method_exists(static::class, 'restoring')) { static::restoring(function ($model) { - AuditOutbox::guardTransaction($model->getConnection()); $model->traceflowAuditRestoring = true; }); static::restored(function ($model) { @@ -92,6 +89,7 @@ private function writeAudit(string $action, array $before, array $after): void 'action' => $action, 'category' => $this->getTable(), 'changes' => $changes, ])); $event = app(AuditClient::class)->createEvent($payload); - app(AuditOutbox::class)->record($event, $this->getConnection()); + // Inside a transaction the event leaves after the commit; a rollback drops it. + app(AuditPublisher::class)->publishAfterCommit($event, $this->getConnection()); } } diff --git a/php/laravel/src/Console/AuditStatusCommand.php b/php/laravel/src/Console/AuditStatusCommand.php deleted file mode 100644 index 8fa7f10..0000000 --- a/php/laravel/src/Console/AuditStatusCommand.php +++ /dev/null @@ -1,34 +0,0 @@ -option('connection') ?: null)->table('traceflow_audit_outbox'); - $pending = (clone $table)->whereNull('failed_at'); - $oldest = (clone $pending)->min('created_at'); - $age = $oldest === null ? 0 : max(0, now()->getTimestamp() - Carbon::parse($oldest)->getTimestamp()); - $result = [ - 'pending' => (clone $pending)->count(), - 'ready' => (clone $pending)->where('available_at', '<=', now())->count(), - 'failed' => (clone $table)->whereNotNull('failed_at')->count(), - 'oldest_pending_seconds' => $age, - ]; - $result['healthy'] = $result['failed'] === 0 - && $age <= max(0, (int) $this->option('max-age')) - && $result['pending'] <= max(0, (int) $this->option('max-pending')); - $this->line(json_encode($result, JSON_THROW_ON_ERROR)); - - return $result['healthy'] ? self::SUCCESS : self::FAILURE; - } -} diff --git a/php/laravel/src/Console/PublishAuditCommand.php b/php/laravel/src/Console/PublishAuditCommand.php deleted file mode 100644 index eac2b74..0000000 --- a/php/laravel/src/Console/PublishAuditCommand.php +++ /dev/null @@ -1,62 +0,0 @@ -option('max-seconds'))); - $stats = ['published' => 0, 'retry_scheduled' => 0, 'failed' => 0, 'budget_exhausted' => false]; - $db = DB::connection($this->option('connection') ?: null); - $rows = $db->table('traceflow_audit_outbox')->whereNull('failed_at')->where('available_at', '<=', now()) - ->orderBy('created_at')->limit(min(1000, max(1, (int) $this->option('limit'))))->get(); - $failed = false; - foreach ($rows as $row) { - $remaining = $deadline - hrtime(true) / 1e9; - if ($remaining <= 0) { - $stats['budget_exhausted'] = true; - break; - } - try { - $client->record(AuditEvent::fromJson($row->payload), $remaining); - // Concurrent publishers are safe: identical event IDs are idempotent in Traceflow. - $db->table('traceflow_audit_outbox')->where('event_id', $row->event_id)->delete(); - $stats['published']++; - } catch (\Throwable $error) { - $failed = true; - $permanent = $error instanceof AuditDeliveryException && ! $error->retryable || $error instanceof \InvalidArgumentException || $error instanceof \JsonException; - $db->table('traceflow_audit_outbox')->where('event_id', $row->event_id)->update([ - 'attempts' => $row->attempts + 1, - 'available_at' => now()->addSeconds(min(3600, 2 ** min(12, $row->attempts + 1))), - 'failed_at' => $permanent ? now() : null, - 'last_error' => $error instanceof AuditDeliveryException ? $error->getMessage() : 'Audit publishing failed; inspect the envelope and publisher configuration', - ]); - $stats[$permanent ? 'failed' : 'retry_scheduled']++; - if (! $this->option('json')) { - $this->error("Audit {$row->event_id}: ".($permanent ? 'requires attention' : 'scheduled for retry')); - } - // A service outage must not trigger retries for every row in the batch. - if (! $permanent) { - break; - } - } - } - - if ($this->option('json')) { - $this->line(json_encode($stats, JSON_THROW_ON_ERROR)); - } - - return $failed || $stats['budget_exhausted'] ? self::FAILURE : self::SUCCESS; - } -} diff --git a/php/laravel/src/Console/RetryAuditCommand.php b/php/laravel/src/Console/RetryAuditCommand.php deleted file mode 100644 index 8029404..0000000 --- a/php/laravel/src/Console/RetryAuditCommand.php +++ /dev/null @@ -1,23 +0,0 @@ -option('connection') ?: null)->table('traceflow_audit_outbox') - ->where('event_id', $this->argument('event_id'))->whereNotNull('failed_at') - ->update(['failed_at' => null, 'available_at' => now(), 'last_error' => null]); - $this->line($changed ? 'Audit event requeued with its original ID and payload.' : 'Failed audit event not found.'); - - return $changed ? self::SUCCESS : self::FAILURE; - } -} diff --git a/php/laravel/src/TraceFlowSDK.php b/php/laravel/src/TraceFlowSDK.php index 5ed80ef..e068c3f 100644 --- a/php/laravel/src/TraceFlowSDK.php +++ b/php/laravel/src/TraceFlowSDK.php @@ -17,11 +17,6 @@ class TraceFlowSDK { - /** Audit is synchronous and reports failures regardless of silent_errors. */ - public function recordAudit(array $input): array - { - return $this->auditClient->recordAudit($input); - } private const UUID_REGEX = '/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i'; private static function ensureValidUuid(?string $value, string $fieldName): string @@ -37,7 +32,6 @@ private static function ensureValidUuid(?string $value, string $fieldName): stri } private TransportInterface $transport; - private \Smartness\TraceFlow\Audit\AuditClient $auditClient; private string $source; @@ -54,7 +48,6 @@ private static function ensureValidUuid(?string $value, string $fieldName): stri public function __construct(array $config) { - $this->auditClient = new \Smartness\TraceFlow\Audit\AuditClient($config); // Master kill switch. When false, the SDK keeps its full public surface // (startTrace/startStep/finish/fail/log all keep working) but events go // to a NullTransport — no HTTP, no error_log noise, no required config. diff --git a/php/laravel/src/TraceFlowServiceProvider.php b/php/laravel/src/TraceFlowServiceProvider.php index 669a9ab..09e2f8e 100644 --- a/php/laravel/src/TraceFlowServiceProvider.php +++ b/php/laravel/src/TraceFlowServiceProvider.php @@ -5,10 +5,8 @@ use Illuminate\Support\ServiceProvider; use Smartness\TraceFlow\Audit\AuditClient; use Smartness\TraceFlow\Audit\AuditContext; -use Smartness\TraceFlow\Console\AuditStatusCommand; +use Smartness\TraceFlow\Audit\AuditPublisher; use Smartness\TraceFlow\Console\BattleTestCommand; -use Smartness\TraceFlow\Console\PublishAuditCommand; -use Smartness\TraceFlow\Console\RetryAuditCommand; use Smartness\TraceFlow\Console\TestCommand; class TraceFlowServiceProvider extends ServiceProvider @@ -23,6 +21,7 @@ public function register(): void ); $this->app->scoped(AuditContext::class, fn () => new AuditContext); $this->app->singleton(AuditClient::class, fn () => new AuditClient(config('traceflow'))); + $this->app->singleton(AuditPublisher::class, fn () => new AuditPublisher(config('traceflow'))); // Register SDK as singleton $this->app->singleton(TraceFlowSDK::class, function ($app) { @@ -60,11 +59,7 @@ public function boot(): void ], 'traceflow-config'); // Register commands - $this->commands([TestCommand::class, BattleTestCommand::class, PublishAuditCommand::class, AuditStatusCommand::class, RetryAuditCommand::class]); - $this->publishes([ - __DIR__.'/../database/migrations/create_traceflow_audit_outbox.php.stub' => database_path('migrations/2026_10_16_000000_create_traceflow_audit_outbox.php'), - __DIR__.'/../database/migrations/index_traceflow_audit_outbox.php.stub' => database_path('migrations/2026_10_17_000000_index_traceflow_audit_outbox.php'), - ], 'traceflow-audit-migrations'); + $this->commands([TestCommand::class, BattleTestCommand::class]); } // Register shutdown handler to close active handles and flush async events. diff --git a/php/laravel/tests/Unit/AuditOutboxTest.php b/php/laravel/tests/Unit/AuditOutboxTest.php deleted file mode 100644 index 383c864..0000000 --- a/php/laravel/tests/Unit/AuditOutboxTest.php +++ /dev/null @@ -1,226 +0,0 @@ -set('database.default', 'testing'); - $app['config']->set('database.connections.testing', ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '']); - $app['config']->set('traceflow.enabled', false); - $app['config']->set('traceflow.source', 'audit-test'); - } - - protected function setUp(): void - { - parent::setUp(); - (require __DIR__.'/../../database/migrations/create_traceflow_audit_outbox.php.stub')->up(); - (require __DIR__.'/../../database/migrations/index_traceflow_audit_outbox.php.stub')->up(); - Schema::create('audit_test_subjects', function (Blueprint $table) { - $table->id(); - $table->string('status')->nullable(); - $table->string('note')->nullable(); - $table->string('secret')->nullable(); - $table->softDeletes(); - }); - } - - public function test_rollback_removes_both_subject_and_audit(): void - { - try { - DB::transaction(function () { - AuditTestSubject::create(['status' => 'draft']); - throw new \RuntimeException('rollback'); - }); - } catch (\RuntimeException) { - } - $this->assertSame(0, AuditTestSubject::count()); - $this->assertSame(0, DB::table('traceflow_audit_outbox')->count()); - } - - public function test_unwrapped_writes_are_rejected_before_the_subject_is_saved(): void - { - try { - AuditTestSubject::create(['status' => 'draft']); - $this->fail('Expected transaction requirement'); - } catch (\LogicException $error) { - $this->assertStringContainsString('transaction', $error->getMessage()); - } - $this->assertSame(0, AuditTestSubject::count()); - } - - public function test_optional_transactions_record_unwrapped_model_writes(): void - { - config(['traceflow.audit.require_transaction' => false]); - $model = AuditTestSubject::create(['status' => 'draft']); - $model->update(['status' => 'sent']); - $model->delete(); - $model->restore(); - $events = DB::table('traceflow_audit_outbox')->get()->map(fn ($row) => json_decode($row->payload)); - $actions = $events->pluck('action')->sort()->values()->all(); - $this->assertSame(['created', 'deleted', 'restored', 'updated'], $actions); - $updated = $events->firstWhere('action', 'updated'); - $this->assertSame('draft', $updated->changes->before->status); - $this->assertSame('sent', $updated->changes->after->status); - } - - public function test_optional_transactions_keep_wrapped_writes_atomic(): void - { - config(['traceflow.audit.require_transaction' => false]); - try { - DB::transaction(function () { - AuditTestSubject::create(['status' => 'draft']); - throw new \RuntimeException('rollback'); - }); - } catch (\RuntimeException) { - } - $this->assertSame(0, AuditTestSubject::count()); - $this->assertSame(0, DB::table('traceflow_audit_outbox')->count()); - } - - public function test_manual_records_follow_the_configured_policy(): void - { - $event = app(AuditClient::class)->createEvent(['subject' => ['type' => 'quote', 'id' => '1'], 'action' => 'exported']); - try { - app(AuditOutbox::class)->record($event, DB::connection()); - $this->fail('Expected transaction requirement'); - } catch (\LogicException) { - } - $this->assertSame(0, DB::table('traceflow_audit_outbox')->count()); - config(['traceflow.audit.require_transaction' => false]); - app(AuditOutbox::class)->record($event, DB::connection()); - $this->assertSame(1, DB::table('traceflow_audit_outbox')->count()); - // The explicit check keeps enforcing a transaction for callers that rely on it. - $this->expectException(\LogicException::class); - AuditOutbox::requireTransaction(DB::connection()); - } - - public function test_model_changes_are_frozen_and_only_selected_fields_are_audited(): void - { - $model = DB::transaction(fn () => AuditTestSubject::create(['status' => 'draft', 'secret' => 'example-excluded-value'])); - app(AuditContext::class)->run(['actor' => ['type' => 'user', 'id' => '42'], 'origin' => ['channel' => 'mobile_app']], fn () => DB::transaction(fn () => $model->update(['status' => 'sent', 'note' => null]))); - DB::transaction(fn () => $model->update(['secret' => 'example-updated-excluded-value'])); - $events = DB::table('traceflow_audit_outbox')->get()->map(fn ($row) => json_decode($row->payload)); - $this->assertCount(2, $events); - $updated = $events->first(fn ($event) => $event->action === 'updated'); - $this->assertSame('draft', $updated->changes->before->status); - $this->assertSame('sent', $updated->changes->after->status); - $this->assertSame('42', $updated->actor->id); - $this->assertSame('mobile_app', $updated->origin->channel); - foreach ($events as $event) { - $this->assertStringNotContainsString('secret', json_encode($event)); - } - $this->assertSame([], app(AuditContext::class)->values()); - DB::transaction(fn () => $model->delete()); - DB::transaction(fn () => $model->restore()); - $actions = DB::table('traceflow_audit_outbox')->get()->map(fn ($row) => json_decode($row->payload)->action)->all(); - sort($actions); - $this->assertSame(['created', 'deleted', 'restored', 'updated'], $actions); - } - - public function test_publisher_keeps_failed_delivery_then_deletes_only_after_persistence(): void - { - DB::transaction(fn () => AuditTestSubject::create(['status' => 'draft'])); - $id = DB::table('traceflow_audit_outbox')->value('event_id'); - $mock = new MockHandler([new Response(503), new Response(201, [], json_encode(['status' => 'persisted', 'event_id' => $id]))]); - $client = new AuditClient(['source' => 'audit-test', 'endpoint' => 'http://audit.test', 'api_key' => 'key', 'max_retries' => 0], new Client(['handler' => HandlerStack::create($mock)])); - $this->app->instance(AuditClient::class, $client); - $this->artisan('traceflow:audit-publish')->assertExitCode(1); - $this->assertSame(1, DB::table('traceflow_audit_outbox')->count()); - $this->assertSame(1, DB::table('traceflow_audit_outbox')->value('attempts')); - DB::table('traceflow_audit_outbox')->update(['available_at' => now()->subMinute()]); - $this->artisan('traceflow:audit-publish')->assertExitCode(0); - $this->assertSame(0, DB::table('traceflow_audit_outbox')->count()); - } - - public function test_outage_stops_the_batch_without_discarding_other_events(): void - { - DB::transaction(function () { - for ($i = 0; $i < 3; $i++) { - AuditTestSubject::create(['status' => 'draft']); - } - }); - $mock = new MockHandler([new Response(503)]); - $this->app->instance(AuditClient::class, new AuditClient(['source' => 'audit-test', 'endpoint' => 'http://audit.test', 'api_key' => 'key', 'max_retries' => 0], new Client(['handler' => HandlerStack::create($mock)]))); - $this->artisan('traceflow:audit-publish', ['--json' => true])->assertExitCode(1); - $this->assertSame(3, DB::table('traceflow_audit_outbox')->count()); - $this->assertSame(1, (int) DB::table('traceflow_audit_outbox')->sum('attempts')); - $this->assertSame(0, DB::table('traceflow_audit_outbox')->whereNotNull('failed_at')->count()); - } - - public function test_status_detects_backlog_and_failed_events_can_be_requeued_without_mutation(): void - { - $this->artisan('traceflow:audit-status')->assertExitCode(0); - DB::transaction(fn () => AuditTestSubject::create(['status' => 'draft'])); - $row = DB::table('traceflow_audit_outbox')->first(); - DB::table('traceflow_audit_outbox')->update(['created_at' => now()->subMinutes(10)]); - $this->artisan('traceflow:audit-status', ['--max-age' => 300])->assertExitCode(1); - $this->artisan('traceflow:audit-status', ['--max-age' => 1000])->assertExitCode(0); - $this->artisan('traceflow:audit-status', ['--max-age' => 1000, '--max-pending' => 0])->assertExitCode(1); - DB::table('traceflow_audit_outbox')->update(['failed_at' => now(), 'attempts' => 2]); - $this->artisan('traceflow:audit-status')->assertExitCode(1); - $this->artisan('traceflow:audit-retry', ['event_id' => $row->event_id])->assertExitCode(0); - $retry = DB::table('traceflow_audit_outbox')->first(); - $this->assertNull($retry->failed_at); - $this->assertSame($row->payload, $retry->payload); - $this->assertSame(2, $retry->attempts); - $this->artisan('traceflow:audit-retry', ['event_id' => $row->event_id])->assertExitCode(1); - } - - public function test_permanent_failure_does_not_block_valid_events_and_can_be_recovered(): void - { - DB::transaction(function () { - AuditTestSubject::create(['status' => 'invalid']); - AuditTestSubject::create(['status' => 'valid']); - }); - $rows = DB::table('traceflow_audit_outbox')->orderBy('created_at')->get(); - $mock = new MockHandler([new Response(400), new Response(201, [], json_encode(['status' => 'persisted', 'event_id' => $rows[1]->event_id]))]); - $this->app->instance(AuditClient::class, new AuditClient(['source' => 'audit-test', 'endpoint' => 'http://audit.test', 'api_key' => 'key', 'max_retries' => 0], new Client(['handler' => HandlerStack::create($mock)]))); - $this->artisan('traceflow:audit-publish')->assertExitCode(1); - $this->assertSame(1, DB::table('traceflow_audit_outbox')->count()); - $this->assertNotNull(DB::table('traceflow_audit_outbox')->value('failed_at')); - } -} - -class AuditTestSubject extends Model -{ - use LogsAudit, SoftDeletes; - - protected $table = 'audit_test_subjects'; - - protected $guarded = []; - - public $timestamps = false; - - public function auditFields(): array - { - return ['status', 'note']; - } - - public function auditType(): string - { - return 'custom-subject'; - } -} diff --git a/php/laravel/tests/Unit/AuditPublisherTest.php b/php/laravel/tests/Unit/AuditPublisherTest.php new file mode 100644 index 0000000..25af842 --- /dev/null +++ b/php/laravel/tests/Unit/AuditPublisherTest.php @@ -0,0 +1,300 @@ + */ + private array $reported = []; + + protected function getPackageProviders($app): array + { + return [TraceFlowServiceProvider::class]; + } + + protected function defineEnvironment($app): void + { + $app['config']->set('database.default', 'testing'); + $app['config']->set('database.connections.testing', ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '']); + $app['config']->set('traceflow.enabled', true); + $app['config']->set('traceflow.source', 'audit-test'); + } + + protected function setUp(): void + { + parent::setUp(); + $this->queen = new FakeQueen; + // No cooldown here: test_after_a_failure_pushes_fail_at_once_until_the_cooldown_ends covers it. + $config = config('traceflow'); + $config['audit']['queen']['cooldown_seconds'] = 0; + $this->app->instance(AuditPublisher::class, new AuditPublisher($config, fn () => $this->queen)); + $handler = $this->app->make(ExceptionHandler::class); + $this->app->instance(ExceptionHandler::class, new class($handler, $this->reported) implements ExceptionHandler + { + public function __construct(private ExceptionHandler $inner, private array &$reported) {} + + public function report(Throwable $e) + { + $this->reported[] = $e; + } + + public function shouldReport(Throwable $e) + { + return true; + } + + public function render($request, Throwable $e) + { + return $this->inner->render($request, $e); + } + + public function renderForConsole($output, Throwable $e) + { + $this->inner->renderForConsole($output, $e); + } + }); + Schema::create('audit_test_subjects', function (Blueprint $table) { + $table->id(); + $table->string('status')->nullable(); + $table->string('note')->nullable(); + $table->string('secret')->nullable(); + $table->softDeletes(); + }); + } + + public function test_each_model_change_is_pushed_to_the_subject_partition_with_its_event_id(): void + { + $model = AuditTestSubject::create(['status' => 'draft']); + $model->update(['status' => 'sent']); + $model->delete(); + $model->restore(); + + $this->assertSame(['created', 'updated', 'deleted', 'restored'], array_map(fn ($push) => $push['data']->action, $this->queen->pushes)); + foreach ($this->queen->pushes as $push) { + $this->assertSame('traceflow.audit', $push['queue']); + $this->assertSame('custom-subject:'.$model->id, $push['partition']); + $this->assertSame($push['data']->event_id, $push['transactionId']); + $this->assertSame('audit-test', $push['data']->source); + } + // Empty images stay JSON objects on the wire, as the backend contract requires. + $this->assertStringContainsString('"before":{}', $this->queen->pushes[0]['json']); + $this->assertStringContainsString('"after":{}', $this->queen->pushes[2]['json']); + $updated = $this->queen->pushes[1]['data']; + $this->assertSame('draft', $updated->changes->before->status); + $this->assertSame('sent', $updated->changes->after->status); + } + + public function test_a_transaction_publishes_after_the_commit_and_a_rollback_publishes_nothing(): void + { + DB::transaction(function () { + AuditTestSubject::create(['status' => 'draft']); + DB::transaction(fn () => AuditTestSubject::create(['status' => 'nested'])); + $this->assertSame([], $this->queen->pushes, 'nothing leaves before the outermost commit'); + }); + $this->assertCount(2, $this->queen->pushes); + + try { + DB::transaction(function () { + AuditTestSubject::create(['status' => 'rolled back']); + throw new \RuntimeException('rollback'); + }); + } catch (\RuntimeException) { + } + $this->assertCount(2, $this->queen->pushes); + $this->assertSame(2, AuditTestSubject::count()); + } + + public function test_a_failed_push_after_the_write_is_reported_and_does_not_undo_the_write(): void + { + $this->queen->failure = new \RuntimeException('connect ECONNREFUSED'); + AuditTestSubject::create(['status' => 'draft']); + DB::transaction(fn () => AuditTestSubject::create(['status' => 'sent'])); + + $this->assertSame(2, AuditTestSubject::count()); + $this->assertCount(2, $this->reported); + $this->assertInstanceOf(AuditDeliveryException::class, $this->reported[0]); + $this->assertTrue($this->reported[0]->retryable); + $this->assertStringContainsString('ECONNREFUSED', $this->reported[0]->getMessage()); + } + + public function test_manual_publish_throws_when_queen_refuses_and_does_nothing_while_disabled(): void + { + $event = app(AuditClient::class)->createEvent(['subject' => ['type' => 'quote', 'id' => '1'], 'action' => 'exported']); + $this->queen->failure = new \RuntimeException('Push failed'); + try { + app(AuditPublisher::class)->publish($event); + $this->fail('Expected a delivery failure'); + } catch (AuditDeliveryException $error) { + $this->assertSame($event->id(), $error->eventId); + } + + $this->queen->failure = null; + (new AuditPublisher(array_merge(config('traceflow'), ['enabled' => false]), fn () => $this->queen))->publish($event); + $this->assertSame([], $this->queen->pushes); + app(AuditPublisher::class)->publish($event); + $this->assertSame('quote:1', $this->queen->pushes[0]['partition']); + } + + public function test_after_a_failure_pushes_fail_at_once_until_the_cooldown_ends(): void + { + $event = fn () => AuditEvent::create('audit-test', ['action' => 'exported']); + $publisher = new AuditPublisher(config('traceflow'), fn () => $this->queen); + $this->queen->failure = new \RuntimeException('Operation timed out'); + $this->expectDeliveryFailure($publisher, $event(), 'timed out'); + $this->queen->failure = null; + $this->expectDeliveryFailure($publisher, $event(), 'recent audit push'); + $this->assertSame(1, $this->queen->calls, 'the broker is not called during the cooldown'); + + $config = config('traceflow'); + $config['audit']['queen']['cooldown_seconds'] = 0; + $publisher = new AuditPublisher($config, fn () => $this->queen); + $this->queen->failure = new \RuntimeException('Operation timed out'); + $this->expectDeliveryFailure($publisher, $event(), 'timed out'); + $this->queen->failure = null; + $publisher->publish($event()); + $this->assertCount(1, $this->queen->pushes); + } + + private function expectDeliveryFailure(AuditPublisher $publisher, AuditEvent $event, string $reason): void + { + try { + $publisher->publish($event); + $this->fail('Expected a delivery failure'); + } catch (AuditDeliveryException $error) { + $this->assertStringContainsString($reason, $error->getMessage()); + $this->assertTrue($error->retryable); + } + } + + public function test_a_missing_broker_url_is_a_configuration_error(): void + { + $config = config('traceflow'); + $config['audit']['queen']['url'] = ''; + $event = AuditEvent::create('audit-test', ['action' => 'exported']); + try { + (new AuditPublisher($config))->publish($event); + $this->fail('Expected a configuration error'); + } catch (AuditDeliveryException $error) { + $this->assertFalse($error->retryable); + $this->assertMatchesRegularExpression('/queen-mq\/php-client|QUEEN_URL/', $error->getMessage()); + } + } + + public function test_events_without_subject_and_long_subjects_get_valid_partitions(): void + { + $this->assertSame('Default', AuditPublisher::partitionFor(AuditEvent::create('test', ['action' => 'exported']))); + $long = AuditEvent::create('test', ['action' => 'exported', 'subject' => ['type' => str_repeat('t', 150), 'id' => str_repeat('i', 150)]]); + $partition = AuditPublisher::partitionFor($long); + $this->assertStringStartsWith('sha256:', $partition); + $this->assertSame($partition, AuditPublisher::partitionFor($long)); + $this->assertLessThanOrEqual(200, strlen($partition)); + } + + public function test_model_changes_are_frozen_and_only_selected_fields_are_audited(): void + { + $model = AuditTestSubject::create(['status' => 'draft', 'secret' => 'example-excluded-value']); + app(AuditContext::class)->run(['actor' => ['type' => 'user', 'id' => '42'], 'origin' => ['channel' => 'mobile_app']], fn () => DB::transaction(fn () => $model->update(['status' => 'sent', 'note' => null]))); + $model->update(['secret' => 'example-updated-excluded-value']); + + $this->assertCount(2, $this->queen->pushes); + $updated = $this->queen->pushes[1]['data']; + $this->assertSame('draft', $updated->changes->before->status); + $this->assertSame('sent', $updated->changes->after->status); + $this->assertSame('42', $updated->actor->id); + $this->assertSame('mobile_app', $updated->origin->channel); + foreach ($this->queen->pushes as $push) { + $this->assertStringNotContainsString('secret', $push['json']); + } + $this->assertSame([], app(AuditContext::class)->values()); + } +} + +/** Records pushes as the Queen PHP client would send them. */ +final class FakeQueen +{ + /** @var list */ + public array $pushes = []; + + public ?Throwable $failure = null; + + public int $calls = 0; + + public function queue(string $name): object + { + $this->calls++; + + return new class($this, $name) + { + private string $partition = 'Default'; + + public function __construct(private FakeQueen $queen, private string $name) {} + + public function partition(string $partition): static + { + $this->partition = $partition; + + return $this; + } + + public function push(array $items): object + { + return new class($this->queen, $this->name, $this->partition, $items) + { + public function __construct(private FakeQueen $queen, private string $name, private string $partition, private array $items) {} + + public function execute(): array + { + if ($this->queen->failure !== null) { + throw $this->queen->failure; + } + foreach ($this->items as $item) { + $json = json_encode($item['data'], JSON_THROW_ON_ERROR); + $this->queen->pushes[] = ['queue' => $this->name, 'partition' => $this->partition, 'transactionId' => $item['transactionId'], 'json' => $json, 'data' => json_decode($json)]; + } + + return [['status' => 'queued']]; + } + }; + } + }; + } +} + +class AuditTestSubject extends Model +{ + use LogsAudit, SoftDeletes; + + protected $table = 'audit_test_subjects'; + + protected $guarded = []; + + public $timestamps = false; + + public function auditFields(): array + { + return ['status', 'note']; + } + + public function auditType(): string + { + return 'custom-subject'; + } +} diff --git a/php/laravel/tests/Unit/AuditTest.php b/php/laravel/tests/Unit/AuditTest.php index a6998d9..d6debe0 100644 --- a/php/laravel/tests/Unit/AuditTest.php +++ b/php/laravel/tests/Unit/AuditTest.php @@ -2,15 +2,8 @@ namespace Smartness\TraceFlow\Tests\Unit; -use GuzzleHttp\Client; -use GuzzleHttp\Handler\MockHandler; -use GuzzleHttp\HandlerStack; -use GuzzleHttp\Middleware; -use GuzzleHttp\Psr7\Response; use PHPUnit\Framework\TestCase; -use Smartness\TraceFlow\Audit\AuditClient; use Smartness\TraceFlow\Audit\AuditContext; -use Smartness\TraceFlow\Audit\AuditDeliveryException; use Smartness\TraceFlow\Audit\AuditEvent; class AuditTest extends TestCase @@ -27,30 +20,6 @@ public function test_diff_preserves_null_absence_and_json_types(): void $this->assertSame([], (array) AuditEvent::diff(['value' => 1], ['value' => 1.0])['after']); } - public function test_delivery_retries_identical_payload_and_returns_receipt(): void - { - $event = AuditEvent::create('test', ['action' => 'viewed', 'metadata' => ['count' => 0]]); - $history = []; - $handler = HandlerStack::create(new MockHandler([new Response(503), new Response(201, [], json_encode(['status' => 'persisted', 'event_id' => $event->id()]))])); - $handler->push(Middleware::history($history)); - $client = new AuditClient(['source' => 'test', 'endpoint' => 'http://audit.test', 'api_key' => 'key', 'retry_delay' => 0], new Client(['handler' => $handler])); - $this->assertSame('persisted', $client->record($event)['status']); - $this->assertCount(2, $history); - $this->assertSame((string) $history[0]['request']->getBody(), (string) $history[1]['request']->getBody()); - } - - public function test_conflict_is_not_retried(): void - { - $client = new AuditClient(['source' => 'test', 'endpoint' => 'http://audit.test', 'api_key' => 'key'], new Client(['handler' => HandlerStack::create(new MockHandler([new Response(409)]))])); - try { - $client->recordAudit(['action' => 'viewed']); - $this->fail('Expected conflict'); - } catch (AuditDeliveryException $error) { - $this->assertFalse($error->retryable); - $this->assertSame(409, $error->statusCode); - } - } - public function test_context_is_restored_even_after_an_exception(): void { $context = new AuditContext; @@ -64,42 +33,6 @@ public function test_context_is_restored_even_after_an_exception(): void $this->assertSame([], $context->values()); } - public function test_publisher_deadline_caps_http_timeout_and_does_not_sleep_past_budget(): void - { - $event = AuditEvent::create('test', ['action' => 'viewed']); - $history = []; - $handler = HandlerStack::create(new MockHandler([new Response(503)])); - $handler->push(Middleware::history($history)); - $client = new AuditClient(['source' => 'test', 'endpoint' => 'http://audit.test', 'api_key' => 'key', 'retry_delay' => 30000], new Client(['handler' => $handler])); - $start = microtime(true); - try { - $client->record($event, 0.1); - $this->fail('Expected retryable delivery failure'); - } catch (AuditDeliveryException $error) { - $this->assertTrue($error->retryable); - } - $this->assertLessThan(1, microtime(true) - $start); - $this->assertCount(1, $history); - $this->assertLessThanOrEqual(0.1, $history[0]['options']['timeout']); - $this->assertFalse($history[0]['options']['allow_redirects']); - } - - public function test_authentication_and_missing_route_failures_remain_recoverable_without_immediate_retries(): void - { - foreach ([401, 403, 404] as $status) { - $mock = new MockHandler([new Response($status)]); - $client = new AuditClient(['source' => 'test', 'endpoint' => 'http://audit.test', 'api_key' => 'key'], new Client(['handler' => HandlerStack::create($mock)])); - try { - $client->recordAudit(['action' => 'viewed']); - $this->fail('Expected delivery failure'); - } catch (AuditDeliveryException $error) { - $this->assertTrue($error->retryable); - $this->assertSame($status, $error->statusCode); - $this->assertSame(0, $mock->count()); - } - } - } - public function test_envelope_free_text_and_nested_fields_follow_the_backend_contract(): void { $valid = ['action' => 'viewed', 'description' => str_repeat('é', 2000), 'actor' => ['type' => 'user', 'id' => '1', 'label' => str_repeat('😀', 150)], 'redacted_fields' => ['guest.phone'], 'origin' => ['channel' => 'api', 'client_source' => 'cb3', 'release' => '1.0']]; diff --git a/php/laravel/tests/audit-installed-framework-smoke.php b/php/laravel/tests/audit-installed-framework-smoke.php index 1c79f09..13c561a 100644 --- a/php/laravel/tests/audit-installed-framework-smoke.php +++ b/php/laravel/tests/audit-installed-framework-smoke.php @@ -5,12 +5,15 @@ use Illuminate\Container\Container; use Illuminate\Contracts\Auth\Factory; use Illuminate\Database\Capsule\Manager; +use Illuminate\Database\DatabaseTransactionsManager; use Illuminate\Database\Eloquent\Model; +use Illuminate\Database\Eloquent\SoftDeletes; use Illuminate\Events\Dispatcher; use Illuminate\Foundation\Application; use Illuminate\Support\Facades\Facade; use Smartness\TraceFlow\Audit\AuditClient; use Smartness\TraceFlow\Audit\AuditContext; +use Smartness\TraceFlow\Audit\AuditPublisher; use Smartness\TraceFlow\Audit\LogsAudit; // Compatibility test using already installed framework classes, without booting @@ -44,27 +47,62 @@ public function user() Facade::setFacadeApplication($app); $app->scoped(AuditContext::class, fn () => new AuditContext); $app->instance(AuditClient::class, new AuditClient(['source' => 'compatibility-test'])); +// Captures pushes instead of contacting a broker. +$pushes = []; +$queen = new class($pushes) +{ + public function __construct(private array &$pushes) {} + + public function queue(string $name): object + { + $pushes = &$this->pushes; + + return new class($pushes) + { + public function __construct(private array &$pushes) {} + + public function partition(string $partition): static + { + return $this; + } + + public function push(array $items): object + { + $pushes = &$this->pushes; + + return new class($pushes, $items) + { + public function __construct(private array &$pushes, private array $items) {} + + public function execute(): array + { + foreach ($this->items as $item) { + $this->pushes[] = json_decode(json_encode($item['data'])); + } + + return [['status' => 'queued']]; + } + }; + } + }; + } +}; +$app->instance(AuditPublisher::class, new AuditPublisher(['enabled' => true], fn () => $queen)); $capsule = new Manager($app); $capsule->addConnection(['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '']); $capsule->setAsGlobal(); $capsule->bootEloquent(); $db = $capsule->getConnection(); +$db->setTransactionManager(new DatabaseTransactionsManager); $db->getSchemaBuilder()->create('audit_subjects', function ($table) { $table->id(); $table->string('status'); $table->softDeletes(); }); -$db->getSchemaBuilder()->create('traceflow_audit_outbox', function ($table) { - $table->uuid('event_id')->primary(); - $table->text('payload'); - $table->timestamp('created_at'); - $table->timestamp('available_at'); - $table->integer('attempts'); -}); class InstalledFrameworkAuditSubject extends Model { - use Illuminate\Database\Eloquent\SoftDeletes, LogsAudit; + use LogsAudit, SoftDeletes; protected $table = 'audit_subjects'; @@ -92,7 +130,7 @@ public function auditFields(): array throw $error; } } -$events = $db->table('traceflow_audit_outbox')->get()->map(fn ($row) => json_decode($row->payload)); +$events = collect($pushes); if ($events->count() !== 4 || InstalledFrameworkAuditSubject::count() !== 1) { throw new RuntimeException('Transactional audit compatibility failed'); } @@ -100,4 +138,4 @@ public function auditFields(): array if ($updated->changes->before->status !== 'draft' || $updated->changes->after->status !== 'sent') { throw new RuntimeException('Audit image compatibility failed'); } -echo 'Laravel '.Application::VERSION.": create/update/delete/restore, frozen images and rollback passed on isolated SQLite\n"; +echo 'Laravel '.Application::VERSION.": create/update/delete/restore, frozen images, publish after commit and rollback passed on isolated SQLite\n";