fix(portal-source): end batch spans on empty batches and stream end

packages/subsquid-pipes/src/core/portal-source.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, describe, expect, expectTypeOf, it } from 'vitest'
 
+import { SpanHooks } from '~/core/profiling.js'
 import { createTarget } from '~/core/target.js'
 import { Target } from '~/core/target.js'
 import { TransformerArgs } from '~/core/transformer.js'
@@ -388,6 +389,79 @@ describe('Portal abstract stream', () => {
   })
 })
 
+describe('profiler span lifecycle', () => {
+  let mockPortal: MockPortal
+
+  afterEach(async () => {
+    await mockPortal?.close()
+  })
+
+  it('ends every batch span — including empty batches and final iteration', async () => {
+    mockPortal = await createMockPortal([
+      // Empty batch (204) — previously caused batchSpan to leak because the
+      // loop reassigns batchSpan without ending the old one when blocks.length === 0.
+      { statusCode: 204 },
+      // Non-empty batch — batchSpan ends via batchEnd(ctx).
+      {
+        statusCode: 200,
+        data: [{ header: { number: 1, hash: '0x123', timestamp: 1000 } }],
+      },
+      // Another empty batch before we finish the range.
+      { statusCode: 204 },
+      // Final non-empty batch hits toBlock; the last batchSpan created by
+      // the bottom-of-loop reassignment must also be ended on normal exit.
+      {
+        statusCode: 200,
+        data: [{ header: { number: 2, hash: '0x456', timestamp: 2000 } }],
+      },
+    ])
+
+    const starts: string[] = []
+    const ends: string[] = []
+
+    const makeHooks = (name: string): SpanHooks => ({
+      onStart(childName) {
+        starts.push(childName)
+        return makeHooks(childName)
+      },
+      onEnd() {
+        ends.push(name)
+      },
+    })
+
+    const rootHooks: SpanHooks = {
+      onStart(name) {
+        starts.push(name)
+        return makeHooks(name)
+      },
+      onEnd() {
+        // root span on the source options isn't ended itself; only children are.
+      },
+    }
+
+    const stream = evmPortalStream({
+      id: 'test',
+      portal: mockPortal.url,
+      outputs: blockDecoder({ from: 0, to: 2 }),
+      profiler: rootHooks,
+    })
+
+    await readAll(stream)
+
+    const batchStarts = starts.filter((n) => n === 'batch').length
+    const batchEnds = ends.filter((n) => n === 'batch').length
+
+    expect(batchStarts).toBeGreaterThan(0)
+    expect(batchEnds).toBe(batchStarts)
+
+    // 'fetch data' spans are started once per loop iteration and must all end too.
+    const fetchStarts = starts.filter((n) => n === 'fetch data').length
+    const fetchEnds = ends.filter((n) => n === 'fetch data').length
+    expect(fetchStarts).toBeGreaterThan(0)
+    expect(fetchEnds).toBe(fetchStarts)
+  })
+})
+
 describe('pipe/pipeTo type guards', () => {
   it('pipe() should not accept objects with a write() method (Target)', () => {
     type SinkLike = { write: () => void }

packages/subsquid-pipes/src/core/portal-source.ts

@@ -255,11 +255,21 @@ export class PortalSource<Q extends QueryBuilder<any>, T = any> {
           const data = await this.applyTransformers(ctx, batch.blocks as T)
 
           yield { data, ctx }
+          // batchSpan was ended by the consumer via batchEnd(ctx) -> ctx.profiler.end()
+        } else {
+          // Nothing to yield, so batchEnd() is never called. Close the span here
+          // to keep onStart/onEnd balanced and avoid leaking spans into sibling parents.
+          batchSpan.end()
         }
 
         batchSpan = Span.root('batch', this.#options.profiler).addLabels('core')
         readSpan = batchSpan.start('fetch data').addLabels('core')
       }
+
+      // The last batchSpan/readSpan created by the bottom-of-loop reassignment
+      // are dangling after the for-await exits normally. End them so onEnd fires.
+      readSpan.end()
+      batchSpan.end()
     }
 
     await this.stop()