fix: correct telemetry span nesting and lifecycle (#2561)

Author: norberttech

documentation/components/core/transformations.md

@@ -232,6 +232,14 @@ This pattern is particularly useful when you need to:
 - Create transformation pipelines that can be reused
 - Separate transformation logic from extraction and loading
 
+The `Transformation` is expanded **once per loader instance**, on the first batch, and every subsequent batch is
+streamed through that same pipeline. Stateful transformations - `limit()`, `add_row_index()` - therefore apply across
+the whole stream, not per batch.
+
+Batching transformations (`batch_size()`, `batch_by()`) expand to a `Processor`, which re-batches only within the
+batches the outer pipeline hands the loader - a `Loader` never sees the stream. To re-batch the pipeline itself, use
+`$df->batchSize(...)` instead of `to_transformation(batch_size(...), ...)`.
+
 ## Creating Custom Transformations
 
 You can create custom transformations by implementing the `Transformation` interface:

documentation/components/libs/telemetry.md

@@ -286,6 +286,36 @@ $span->setAttribute('order.total', 99.99);
 $tracer->complete($span);
 ```
 
+**Nesting spans:**
+
+Creating a span does not make it the current one — per the OpenTelemetry specification, span creation must not
+change the active context. A span created inside another becomes its child only if the outer span was activated:
+
+```php
+<?php
+
+$order = $tracer->span('process-order');
+$scope = $tracer->activate($order);
+
+try {
+    // child of 'process-order'
+    $tracer->complete($tracer->span('charge-card'));
+} finally {
+    $scope->detach();
+    $tracer->complete($order);
+}
+```
+
+Without `activate()`, `charge-card` is a sibling of `process-order`, not a child.
+
+Detach scopes in reverse order of activation, and before completing the span. `Scope::detach()` returns
+`Scope::DETACHED` on success, `Scope::INACTIVE` if the scope was already detached, and `Scope::MISMATCH` if it was
+not the innermost active scope.
+
+Do **not** activate a span whose lifetime is an object rather than a block — a file stream, a database cursor, a
+long-running transaction handle. Several of those are open at once, so activating them makes each the parent of
+whichever is opened next, producing a staircase instead of siblings. Start such spans and leave them inactive.
+
 For automatic exception handling and span completion, use the `trace()` method:
 
 ```php

documentation/components/libs/telemetry/contracts.md

@@ -67,8 +67,9 @@ Three small immutable value objects in `Flow\Telemetry\Batch` carry the payload
 
 **Interface:** `Flow\Telemetry\Context\ContextStorage`
 
-Stores and retrieves telemetry context (active span, baggage) within a request lifecycle. Enables automatic context
-propagation to child spans.
+Stores and retrieves telemetry context (active span, baggage) within a request lifecycle. `attach()` returns a
+`Scope` that must be detached in reverse order; a span becomes the parent of later spans only while its scope is
+attached — see `Tracer::activate()`.
 
 | Implementation         | Package              | Description                                |
 |------------------------|----------------------|--------------------------------------------|
@@ -102,12 +103,12 @@ sits between the Tracer and the unified `Exporter`. The `exporter()` accessor re
 Makes sampling decisions for traces. Determines whether a span should be recorded and exported based on configurable
 rules.
 
-| Implementation             | Package              | Description                                         |
-|----------------------------|----------------------|-----------------------------------------------------|
-| `AlwaysOnSampler`          | `flow-php/telemetry` | Records all traces                                  |
-| `AlwaysOffSampler`         | `flow-php/telemetry` | Records no traces                                   |
-| `TraceIdRatioBasedSampler` | `flow-php/telemetry` | Records a configurable percentage of traces         |
-| `ParentBasedSampler`       | `flow-php/telemetry` | Inherits sampling decision from parent span context |
+| Implementation             | Package              | Description                                                                                              |
+|----------------------------|----------------------|----------------------------------------------------------------------------------------------------------|
+| `AlwaysOnSampler`          | `flow-php/telemetry` | Records all traces                                                                                       |
+| `AlwaysOffSampler`         | `flow-php/telemetry` | Records no traces                                                                                        |
+| `TraceIdRatioBasedSampler` | `flow-php/telemetry` | Records a configurable percentage of traces                                                              |
+| `ParentBasedSampler`       | `flow-php/telemetry` | Inherits sampling decision from parent span context                                                      |
 | `AttributeMatchingSampler` | `flow-php/telemetry` | Drops spans matching an `AttributeFilter` (start-time attributes), defers the rest to a delegate sampler |
 
 ---
@@ -168,13 +169,13 @@ Processes log records from loggers. Determines how logs are buffered and when th
 (marker interface extending `LogProcessor`) identifies the terminal/leaf processors; `PipelineLogProcessor` runs an
 ordered chain of `LogMiddleware` and forwards survivors to one `LogSink`.
 
-| Implementation            | Package              | Sink? | Description                                          |
-|---------------------------|----------------------|-------|------------------------------------------------------|
-| `PassThroughLogProcessor` | `flow-php/telemetry` | yes   | Exports each log immediately when recorded           |
-| `BatchingLogProcessor`    | `flow-php/telemetry` | yes   | Buffers logs and exports in configurable batches     |
-| `CompositeLogProcessor`   | `flow-php/telemetry` | yes   | Delegates to multiple processors                     |
-| `MemoryLogProcessor`      | `flow-php/telemetry` | yes   | Stores logs in memory for testing                    |
-| `VoidLogProcessor`        | `flow-php/telemetry` | yes   | No-op processor that discards all logs               |
+| Implementation            | Package              | Sink? | Description                                            |
+|---------------------------|----------------------|-------|--------------------------------------------------------|
+| `PassThroughLogProcessor` | `flow-php/telemetry` | yes   | Exports each log immediately when recorded             |
+| `BatchingLogProcessor`    | `flow-php/telemetry` | yes   | Buffers logs and exports in configurable batches       |
+| `CompositeLogProcessor`   | `flow-php/telemetry` | yes   | Delegates to multiple processors                       |
+| `MemoryLogProcessor`      | `flow-php/telemetry` | yes   | Stores logs in memory for testing                      |
+| `VoidLogProcessor`        | `flow-php/telemetry` | yes   | No-op processor that discards all logs                 |
 | `PipelineLogProcessor`    | `flow-php/telemetry` | no    | Runs middleware in order, then forwards to a `LogSink` |
 
 ### LogMiddleware
@@ -184,11 +185,11 @@ ordered chain of `LogMiddleware` and forwards survivors to one `LogSink`.
 A chainable step inside a `PipelineLogProcessor`. `process(LogEntry): ?LogEntry` returns the entry to pass on (possibly
 enriched) or `null` to drop it. Stateless - flush/shutdown belong to the pipeline's `LogSink`.
 
-| Implementation                     | Package              | Description                                          |
-|------------------------------------|----------------------|------------------------------------------------------|
-| `EnrichingLogMiddleware`           | `flow-php/telemetry` | Merges default attributes (call-site values win)     |
-| `SeverityFilteringLogMiddleware`   | `flow-php/telemetry` | Drops log entries below a minimum severity threshold |
-| `AttributeFilteringLogMiddleware`  | `flow-php/telemetry` | Drops/keeps log entries by an `AttributeFilter`      |
+| Implementation                    | Package              | Description                                          |
+|-----------------------------------|----------------------|------------------------------------------------------|
+| `EnrichingLogMiddleware`          | `flow-php/telemetry` | Merges default attributes (call-site values win)     |
+| `SeverityFilteringLogMiddleware`  | `flow-php/telemetry` | Drops log entries below a minimum severity threshold |
+| `AttributeFilteringLogMiddleware` | `flow-php/telemetry` | Drops/keeps log entries by an `AttributeFilter`      |
 
 ---
 

documentation/upgrading.md

@@ -7,6 +7,59 @@ specific version to ensure a smooth upgrade process.
 
 ---
 
+## Upgrading from 0.42.x to 0.43.x
+
+### 1) `flow-php/etl` - `to_transformation()` expands a `Transformation` once per loader, not once per batch
+
+| Before                                                                            | After                   |
+|-----------------------------------------------------------------------------------|-------------------------|
+| `to_transformation(limit(3), $loader)`, 6 batches → 6 rows loaded                 | 3 rows loaded           |
+| `to_transformation(add_row_index('n'), $loader)`, 6 batches → `n = [1,1,1,1,1,1]` | `n = [1,2,3,4,5,6]`     |
+| nested `DataFrame` span per batch                                                 | one nested span per run |
+
+Unchanged: any pipeline using `->collect()`, `drop()`, `select()`, `mask_columns()`, `batch_size()`, `batch_by()`.
+
+### 2) `flow-php/telemetry` - `Tracer::span()` no longer activates the span
+
+| Before                                             | After                                                             |
+|----------------------------------------------------|-------------------------------------------------------------------|
+| `$tracer->span('x')` makes the span current        | does not; `$tracer->activate($span): Scope` does                  |
+| `$tracer->complete($span)` also detaches the scope | ends the span only                                                |
+| `span(…, SpanContext $parentContext)`              | `span(…, Context $parent)`                                        |
+| `trace(…, SpanContext $parentContext)`             | `trace(…, Context $parent)`                                       |
+| `Scope::detach()` always returns `0`               | returns `Scope::DETACHED`, `Scope::INACTIVE` or `Scope::MISMATCH` |
+
+Call sites relying on implicit nesting still compile and silently produce siblings. Rewrite each one that needs
+children:
+
+Before:
+
+```php
+$span = $tracer->span('parent');
+
+try {
+    // ...
+} finally {
+    $tracer->complete($span);
+}
+```
+
+After:
+
+```php
+$span = $tracer->span('parent');
+$scope = $tracer->activate($span);
+
+try {
+    // ...
+} finally {
+    $scope->detach();
+    $tracer->complete($span);
+}
+```
+
+---
+
 ## Upgrading from 0.41.x to 0.42.x
 
 ### 1) `flow-php/symfony-telemetry-bundle` - messenger tracing simplified
@@ -479,6 +532,7 @@ Custom aggregators must implement `references()` - return the references the agg
 | `DataFrame::pivot()`                              | removed; `GroupedDataFrame::pivot()` only         |
 
 ### 32) `flow-php/etl-adapter-csv`, `-excel`, `-json`,
+
 `-xml` - explicit schema no longer projects partition columns away
 
 | Before                                                           | After                                                |
@@ -2255,7 +2309,7 @@ After:
     ->run();
 ```
 
-### 4) ConfigBuilder::putInputIntoRows () output is now prefixed with _   (underscore)
+### 4) ConfigBuilder::putInputIntoRows () output is now prefixed with _    (underscore)
 
 In order to avoid collisions with datasets columns, additional columns created after using putInputIntoRows ()
 would now be prefixed with `_` (underscore) symbol.

src/bridge/monolog/telemetry/tests/Flow/Bridge/Monolog/Telemetry/Tests/Integration/TelemetryHandlerIntegrationTest.php

@@ -233,7 +233,9 @@ public function test_logs_after_span_completion_have_no_trace_context(): void
         $monolog->pushHandler(telemetry_handler($context->logger));
 
         $span = $tracer->span('short-operation');
+        $scope = $tracer->activate($span);
         $monolog->info('Inside span');
+        $scope->detach();
         $tracer->complete($span);
 
         $monolog->info('After span completed');
@@ -255,10 +257,12 @@ public function test_logs_within_active_span_include_trace_and_span_ids(): void
         $monolog->pushHandler(telemetry_handler($context->logger));
 
         $span = $tracer->span('process-order');
+        $scope = $tracer->activate($span);
 
         $monolog->info('Processing order', ['order_id' => 123]);
         $monolog->warning('Low inventory', ['product_id' => 456]);
 
+        $scope->detach();
         $tracer->complete($span);
 
         $entries = $context->processor->entries();
@@ -436,14 +440,18 @@ public function test_nested_spans_propagate_child_span_id_to_logs(): void
         $monolog->pushHandler(telemetry_handler($context->logger));
 
         $parentSpan = $tracer->span('parent-operation');
+        $parentScope = $tracer->activate($parentSpan);
         $monolog->info('In parent span');
 
         $childSpan = $tracer->span('child-operation');
+        $childScope = $tracer->activate($childSpan);
         $monolog->info('In child span');
 
+        $childScope->detach();
         $tracer->complete($childSpan);
         $monolog->info('Back in parent span');
 
+        $parentScope->detach();
         $tracer->complete($parentSpan);
 
         $entries = $context->processor->entries();

src/bridge/phpunit/telemetry/src/Flow/Bridge/PHPUnit/Telemetry/SpanStack.php

@@ -4,13 +4,14 @@
 
 namespace Flow\Bridge\PHPUnit\Telemetry;
 
+use Flow\Telemetry\Context\Scope;
 use Flow\Telemetry\Tracer\Span;
 use SplStack;
 
 final class SpanStack
 {
     /**
-     * @var \SplStack<Span>
+     * @var \SplStack<array{span: Span, scope: Scope}>
      */
     private SplStack $stack;
 
@@ -21,14 +22,14 @@ final class SpanStack
 
     public function __construct()
     {
-        /** @var \SplStack<Span> $stack */
+        /** @var \SplStack<array{span: Span, scope: Scope}> $stack */
         $stack = new SplStack();
         $this->stack = $stack;
     }
 
     public function clear(): void
     {
-        /** @var \SplStack<Span> $stack */
+        /** @var \SplStack<array{span: Span, scope: Scope}> $stack */
         $stack = new SplStack();
         $this->stack = $stack;
         $this->suiteSpans = [];
@@ -40,7 +41,7 @@ public function current(): ?Span
             return null;
         }
 
-        return $this->stack->top();
+        return $this->stack->top()['span'];
     }
 
     public function getSuiteSpan(string $suiteName): ?Span
@@ -53,18 +54,25 @@ public function isEmpty(): bool
         return $this->stack->isEmpty();
     }
 
+    /**
+     * Detaches the entry's context scope before returning it, so the stack owns the LIFO ordering that
+     * Scope::detach() requires.
+     */
     public function pop(): ?Span
     {
         if ($this->stack->isEmpty()) {
             return null;
         }
 
-        return $this->stack->pop();
+        $entry = $this->stack->pop();
+        $entry['scope']->detach();
+
+        return $entry['span'];
     }
 
-    public function push(Span $span): void
+    public function push(Span $span, Scope $scope): void
     {
-        $this->stack->push($span);
+        $this->stack->push(['span' => $span, 'scope' => $scope]);
     }
 
     public function removeSuiteSpan(string $suiteName): void

src/bridge/phpunit/telemetry/src/Flow/Bridge/PHPUnit/Telemetry/Subscriber/TestPreparationStartedSubscriber.php

@@ -58,7 +58,8 @@ public function notify(PreparationStarted $event): void
                 PHPUnitTelemetryAttributes::ATTR_TEST_METHOD => $methodName,
             ]);
 
-            $this->spanStack->push($span);
+            // activated: spans the test itself emits must nest under the test span
+            $this->spanStack->push($span, $tracer->activate($span));
         } catch (Throwable) {
             // Silent failure - telemetry must never break tests
         }

src/bridge/phpunit/telemetry/src/Flow/Bridge/PHPUnit/Telemetry/Subscriber/TestSuiteFinishedSubscriber.php

@@ -84,9 +84,10 @@ public function notify(Finished $event): void
                     ]);
                 }
 
+                // pop first: it detaches the context scope, which must happen before the span completes
+                $this->spanStack->pop();
                 $tracer->complete($span);
 
-                $this->spanStack->pop();
                 $this->spanStack->removeSuiteSpan($suiteName);
             }
 

src/bridge/phpunit/telemetry/src/Flow/Bridge/PHPUnit/Telemetry/Subscriber/TestSuiteStartedSubscriber.php

@@ -53,8 +53,9 @@ public function notify(Started $event): void
                 PHPUnitTelemetryAttributes::ATTR_SUITE_IS_ROOT => $isRoot,
             ]);
 
+            // activated: suites nest, and test spans must nest under their suite
             $this->spanStack->setSuiteSpan($suite->name(), $span);
-            $this->spanStack->push($span);
+            $this->spanStack->push($span, $tracer->activate($span));
             $this->suiteOutcomes->push();
         } catch (Throwable) {
             // Silent failure - telemetry must never break tests

src/bridge/phpunit/telemetry/tests/Flow/Bridge/PHPUnit/Telemetry/Tests/Unit/SpanStackTest.php

@@ -5,6 +5,7 @@
 namespace Flow\Bridge\PHPUnit\Telemetry\Tests\Unit;
 
 use Flow\Bridge\PHPUnit\Telemetry\SpanStack;
+use Flow\Telemetry\Tests\Mother\ScopeMother;
 use Flow\Telemetry\Tests\Mother\SpanMother;
 use PHPUnit\Framework\TestCase;
 
@@ -17,8 +18,8 @@ public function test_clear_removes_all_spans(): void
         $span2 = SpanMother::create('span-2');
         $suiteSpan = SpanMother::create('suite-span');
 
-        $stack->push($span1);
-        $stack->push($span2);
+        $stack->push($span1, ScopeMother::attached());
+        $stack->push($span2, ScopeMother::attached());
         $stack->setSuiteSpan('TestSuite', $suiteSpan);
 
         $stack->clear();
@@ -32,7 +33,7 @@ public function test_current_returns_top_without_removing(): void
         $stack = new SpanStack();
         $span = SpanMother::create('test-span');
 
-        $stack->push($span);
+        $stack->push($span, ScopeMother::attached());
 
         static::assertSame($span, $stack->current());
         static::assertSame($span, $stack->current());
@@ -46,9 +47,9 @@ public function test_lifo_order(): void
         $span2 = SpanMother::create('span-2');
         $span3 = SpanMother::create('span-3');
 
-        $stack->push($span1);
-        $stack->push($span2);
-        $stack->push($span3);
+        $stack->push($span1, ScopeMother::attached());
+        $stack->push($span2, ScopeMother::attached());
+        $stack->push($span3, ScopeMother::attached());
 
         static::assertSame($span3, $stack->pop());
         static::assertSame($span2, $stack->pop());
@@ -61,7 +62,7 @@ public function test_push_and_pop(): void
         $stack = new SpanStack();
         $span = SpanMother::create('test-span');
 
-        $stack->push($span);
+        $stack->push($span, ScopeMother::attached());
 
         static::assertFalse($stack->isEmpty());
         static::assertSame($span, $stack->current());