Add caller-side DuckDB CPU attribution metric

src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs

@@ -0,0 +1,62 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+
+namespace KurrentDB.Core.DuckDB;
+
+// Attributes CPU consumed by DuckDB operations executed on KurrentDB threads.
+// DuckDB also schedules work onto its own native worker threads; that share is not visible
+// to caller-side measurement and is reported separately once KurrentDB owns those workers.
+public class DuckDBCpuMetrics {
+	public const string MeterName = "KurrentDB.DuckDB";
+
+	public static class Activities {
+		public const string Query = "query";
+		public const string Read = "read";
+		public const string Commit = "commit";
+		public const string Checkpoint = "checkpoint";
+	}
+
+	private static readonly KeyValuePair<string, object> SourceTag = new("source", "caller");
+
+	private readonly Counter<double> _cpuSeconds;
+
+	public DuckDBCpuMetrics(Meter meter, string serviceName) {
+		_cpuSeconds = meter.CreateCounter<double>(
+			$"{serviceName}.duckdb.cpu.seconds",
+			description: "CPU time consumed by DuckDB operations on KurrentDB threads, in seconds");
+	}
+
+	public CpuScope Measure(string activity) => new(this, activity);
+
+	// A ref struct so it cannot live across an await: the CPU delta is only valid when start
+	// and stop are read on the same thread.
+	public readonly ref struct CpuScope {
+		private readonly DuckDBCpuMetrics _metrics;
+		private readonly string _activity;
+		private readonly long _startNanoseconds;
+
+		internal CpuScope(DuckDBCpuMetrics metrics, string activity) {
+			if (!ThreadCpuTime.IsSupported)
+				return;
+
+			_metrics = metrics;
+			_activity = activity;
+			_startNanoseconds = ThreadCpuTime.CurrentNanoseconds;
+		}
+
+		public void Dispose() {
+			if (_metrics is null)
+				return;
+
+			var elapsedNanoseconds = ThreadCpuTime.CurrentNanoseconds - _startNanoseconds;
+			if (elapsedNanoseconds > 0)
+				_metrics._cpuSeconds.Add(
+					elapsedNanoseconds / 1e9,
+					new KeyValuePair<string, object>("activity", _activity),
+					SourceTag);
+		}
+	}
+}

src/KurrentDB.Core/DuckDB/InjectionExtensions.cs

@@ -2,8 +2,10 @@
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
 using System;
+using System.Diagnostics.Metrics;
 using System.Threading.Tasks;
 using Kurrent.Quack.ConnectionPool;
+using KurrentDB.Common.Configuration;
 using KurrentDB.DuckDB;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Connections;
@@ -21,6 +23,10 @@ public static IServiceCollection AddDuckDb(this IServiceCollection services) {
 		services.AddSingleton<DuckDBConnectionPool>(sp => sp.GetRequiredService<DuckDBConnectionPoolLifetime>().Shared);
 		services.AddSingleton<DuckDbConnectionPoolMiddleware>();
 		services.AddSingleton<ConnectionInterceptor>(CreatePoolPerConnectionInterceptor);
+		services.AddSingleton(static sp => {
+			var serviceName = sp.GetService<MetricsConfiguration>()?.ServiceName ?? "kurrentdb";
+			return new DuckDBCpuMetrics(new Meter(DuckDBCpuMetrics.MeterName, "1.0.0"), serviceName);
+		});
 		return services;
 	}
 

src/KurrentDB.Core/DuckDB/ThreadCpuTime.cs

@@ -0,0 +1,63 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System;
+using System.Runtime.InteropServices;
+
+namespace KurrentDB.Core.DuckDB;
+
+// Reads the cumulative CPU time consumed by the calling thread.
+// A delta between two readings is only meaningful when both are taken on the same thread.
+internal static class ThreadCpuTime {
+	// CLOCK_THREAD_CPUTIME_ID differs per libc: bits/time.h on Linux, sys/_types/_clockid_t on macOS
+	private const int ClockThreadCpuTimeIdLinux = 3;
+	private const int ClockThreadCpuTimeIdMacOS = 16;
+
+	public static readonly bool IsSupported = Detect();
+
+	public static long CurrentNanoseconds => OperatingSystem.IsWindows()
+		? GetWindowsThreadCpuNanoseconds()
+		: GetPosixThreadCpuNanoseconds();
+
+	private static long GetPosixThreadCpuNanoseconds() {
+		var clockId = OperatingSystem.IsLinux() ? ClockThreadCpuTimeIdLinux : ClockThreadCpuTimeIdMacOS;
+		return clock_gettime(clockId, out var ts) == 0
+			? ts.Seconds * 1_000_000_000L + ts.Nanoseconds
+			: 0;
+	}
+
+	private static long GetWindowsThreadCpuNanoseconds() {
+		// GetThreadTimes reports in 100ns units, accrued at scheduler-quantum granularity.
+		// Cumulative totals are statistically accurate; very short individual deltas are not.
+		return GetThreadTimes(GetCurrentThread(), out _, out _, out var kernelTime, out var userTime)
+			? (kernelTime + userTime) * 100
+			: 0;
+	}
+
+	private static bool Detect() {
+		if (!OperatingSystem.IsWindows() && !OperatingSystem.IsLinux() && !OperatingSystem.IsMacOS())
+			return false;
+
+		try {
+			_ = CurrentNanoseconds;
+			return true;
+		} catch (Exception e) when (e is DllNotFoundException or EntryPointNotFoundException) {
+			return false;
+		}
+	}
+
+	[StructLayout(LayoutKind.Sequential)]
+	private struct Timespec {
+		public nint Seconds;
+		public nint Nanoseconds;
+	}
+
+	[DllImport("libc")]
+	private static extern int clock_gettime(int clockId, out Timespec ts);
+
+	[DllImport("kernel32.dll", ExactSpelling = true)]
+	private static extern nint GetCurrentThread();
+
+	[DllImport("kernel32.dll", ExactSpelling = true)]
+	private static extern bool GetThreadTimes(nint thread, out long creationTime, out long exitTime, out long kernelTime, out long userTime);
+}

src/KurrentDB.SecondaryIndexing.Tests/Diagnostics/DuckDBCpuMetricsTests.cs

@@ -0,0 +1,58 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using KurrentDB.Core.DuckDB;
+
+namespace KurrentDB.SecondaryIndexing.Tests.Diagnostics;
+
+public class DuckDBCpuMetricsTests {
+	[Fact]
+	public void busy_scope_records_cpu_seconds_with_activity_and_source_tags() {
+		using var meter = new Meter("test");
+		var metrics = new DuckDBCpuMetrics(meter, "kurrentdb");
+
+		List<(double Value, KeyValuePair<string, object?>[] Tags)> measurements = [];
+		using var listener = Listen(meter, measurements);
+
+		using (metrics.Measure(DuckDBCpuMetrics.Activities.Commit)) {
+			// busy spin long enough to accrue CPU past Windows' quantum-granular thread accounting
+			var stopwatch = Stopwatch.StartNew();
+			while (stopwatch.ElapsedMilliseconds < 100) {
+			}
+		}
+
+		var measurement = Assert.Single(measurements);
+		Assert.InRange(measurement.Value, 0.001, 30);
+		Assert.Contains(measurement.Tags, t => t is { Key: "activity", Value: "commit" });
+		Assert.Contains(measurement.Tags, t => t is { Key: "source", Value: "caller" });
+	}
+
+	[Fact]
+	public void idle_scope_records_at_most_negligible_cpu() {
+		using var meter = new Meter("test");
+		var metrics = new DuckDBCpuMetrics(meter, "kurrentdb");
+
+		List<(double Value, KeyValuePair<string, object?>[] Tags)> measurements = [];
+		using var listener = Listen(meter, measurements);
+
+		using (metrics.Measure(DuckDBCpuMetrics.Activities.Read)) {
+			Thread.Sleep(100);
+		}
+
+		Assert.True(measurements.Sum(m => m.Value) < 0.05, $"expected near-zero CPU, got {measurements.Sum(m => m.Value)}s");
+	}
+
+	private static MeterListener Listen(Meter meter, List<(double, KeyValuePair<string, object?>[])> measurements) {
+		var listener = new MeterListener();
+		listener.InstrumentPublished = (instrument, l) => {
+			if (instrument.Meter == meter && instrument.Name == "kurrentdb.duckdb.cpu.seconds")
+				l.EnableMeasurementEvents(instrument);
+		};
+		listener.SetMeasurementEventCallback<double>(
+			(_, value, tags, _) => measurements.Add((value, tags.ToArray())));
+		listener.Start();
+		return listener;
+	}
+}

src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexProcessorTests.cs

@@ -4,6 +4,7 @@
 using DotNext;
 using Kurrent.Quack;
 using KurrentDB.Core.Data;
+using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Index.Hashes;
 using KurrentDB.Core.Tests.Fakes;
 using KurrentDB.SecondaryIndexing.Indexes.Default;
@@ -211,7 +212,8 @@ public DefaultIndexProcessorTests() {
 
 		var publisher = new FakePublisher();
 
-		_processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance);
+		_processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance,
+			new DuckDBCpuMetrics(new("test"), "kurrentdb"));
 	}
 
 	public override ValueTask DisposeAsync() {

src/KurrentDB.SecondaryIndexing.Tests/Indexes/DefaultIndexReaderTests/IndexTestBase.cs

@@ -2,6 +2,7 @@
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
 using KurrentDB.Core.Data;
+using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Index.Hashes;
 using KurrentDB.Core.Tests.Fakes;
 using KurrentDB.SecondaryIndexing.Indexes.Default;
@@ -22,9 +23,10 @@ protected IndexTestBase() {
 		var hasher = new CompositeHasher<string>(new XXHashUnsafe(), new Murmur3AUnsafe());
 		var publisher = new FakePublisher();
 
-		_processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance);
+		var cpuMetrics = new DuckDBCpuMetrics(new("test"), "kurrentdb");
+		_processor = new(DuckDb, publisher, hasher, new("test"), NullLoggerFactory.Instance, cpuMetrics);
 
-		Sut = new(DuckDb, _processor, _readIndexStub.ReadIndex);
+		Sut = new(DuckDb, _processor, _readIndexStub.ReadIndex, cpuMetrics);
 	}
 
 	protected void IndexEvents(ResolvedEvent[] events, bool shouldCommit) {

src/KurrentDB.SecondaryIndexing.Tests/IntegrationTests/ReadTests.cs

@@ -1,8 +1,10 @@
 // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
+using System.Diagnostics.Metrics;
 using KurrentDB.Core.ClientPublisher;
 using KurrentDB.Core.Data;
+using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Services;
 using KurrentDB.Core.Services.Storage.ReaderIndex;
 using KurrentDB.Core.Services.Transport.Common;
@@ -44,6 +46,50 @@ await engine.ExecuteAsync(
 		Assert.True(consumer.RowCount > 0);
 	}
 
+	// Guards that the streaming query path (QueryResultReader.TryRead/FinalizeEnumeration, where DuckDB
+	// produces result chunks) emits caller-attributed CPU under the query activity. We assert tagging and
+	// wiring rather than a per-fetch measurement count: a single sub-vector-size chunk fetch can complete
+	// below the Windows GetThreadTimes quantum and record zero, so a count assertion would be flaky there.
+	[Fact]
+	public async Task QueryEngineStreamingReadAttributesQueryCpuToCaller() {
+		var engine = Fixture.NodeServices.GetRequiredService<IQueryEngine>();
+		using var preparedSql = engine.PrepareQuery(
+			"SELECT metadata FROM kdb.records"u8,
+			new() { UseDigitalSignature = true });
+
+		object gate = new();
+		List<KeyValuePair<string, object?>[]> queryMeasurements = [];
+		using var listener = new MeterListener();
+		listener.InstrumentPublished = (instrument, l) => {
+			if (instrument.Meter.Name == DuckDBCpuMetrics.MeterName && instrument.Name == "kurrentdb.duckdb.cpu.seconds")
+				l.EnableMeasurementEvents(instrument);
+		};
+		listener.SetMeasurementEventCallback<double>((_, _, tags, _) => {
+			var copy = tags.ToArray();
+			// Ignore background index commit/checkpoint activity on the shared meter.
+			if (copy.Any(t => t is { Key: "activity", Value: "query" })) {
+				lock (gate)
+					queryMeasurements.Add(copy);
+			}
+		});
+		listener.Start();
+
+		var consumer = new RowCountReader();
+		await engine.ExecuteAsync(
+			preparedSql.Memory,
+			consumer,
+			new() { CheckIntegrity = true },
+			TestContext.Current.CancellationToken);
+
+		listener.Dispose(); // stop callbacks before asserting
+
+		Assert.True(consumer.RowCount > 0); // the streaming fetch loop actually ran
+		lock (gate) {
+			Assert.NotEmpty(queryMeasurements);
+			Assert.All(queryMeasurements, tags => Assert.Contains(tags, t => t is { Key: "source", Value: "caller" }));
+		}
+	}
+
 	[Theory]
 	[InlineData(true)]
 	[InlineData(false)]

src/KurrentDB.SecondaryIndexing/Indexes/Category/CategoryIndexReader.cs

@@ -4,6 +4,7 @@
 using Kurrent.Quack;
 using Kurrent.Quack.ConnectionPool;
 using KurrentDB.Core.Data;
+using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Services.Storage.ReaderIndex;
 using KurrentDB.SecondaryIndexing.Indexes.Default;
 using KurrentDB.SecondaryIndexing.Storage;
@@ -16,8 +17,9 @@ namespace KurrentDB.SecondaryIndexing.Indexes.Category;
 internal class CategoryIndexReader(
 	DuckDBConnectionPool sharedPool,
 	DefaultIndexProcessor processor,
-	IReadIndex<string> index)
-	: SecondaryIndexReaderBase(sharedPool, index) {
+	IReadIndex<string> index,
+	DuckDBCpuMetrics cpuMetrics)
+	: SecondaryIndexReaderBase(sharedPool, index, cpuMetrics) {
 	protected override string GetId(string indexName) =>
 		CategoryIndex.TryParseCategoryName(indexName, out var categoryName)
 			? categoryName

src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexProcessor.cs

@@ -14,6 +14,7 @@
 using KurrentDB.Common.Configuration;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Data;
+using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Index.Hashes;
 using KurrentDB.Core.Messages;
 using KurrentDB.Core.Services;
@@ -33,6 +34,7 @@ internal class DefaultIndexProcessor : Disposable, ISecondaryIndexProcessor {
 	private readonly ILongHasher<string> _hasher;
 	private readonly ILogger<DefaultIndexProcessor> _log;
 	private readonly BufferedView _appender;
+	private readonly DuckDBCpuMetrics _cpuMetrics;
 	private Atomic<TFPos> _lastPosition;
 
 	public TFPos LastIndexedPosition {
@@ -47,10 +49,12 @@ public DefaultIndexProcessor(
 		[FromKeyedServices(SecondaryIndexingConstants.InjectionKey)]
 		Meter meter,
 		ILoggerFactory loggerFactory,
+		DuckDBCpuMetrics cpuMetrics,
 		MetricsConfiguration? metricsConfiguration = null,
 		TimeProvider? clock = null
 	) {
 		_connection = db.Open();
+		_cpuMetrics = cpuMetrics;
 		_log = loggerFactory.CreateLogger<DefaultIndexProcessor>();
 		_appender = new(_connection, "idx_all", "log_position", DefaultIndexViewName);
 		var serviceName = metricsConfiguration?.ServiceName ?? "kurrentdb";
@@ -153,6 +157,7 @@ public void Commit() {
 
 		try {
 			using var duration = Tracker.StartCommitDuration();
+			using var cpu = _cpuMetrics.Measure(DuckDBCpuMetrics.Activities.Commit);
 			_appender.Flush();
 		} catch (Exception e) {
 			_log.LogError(e, "Failed to commit records to index at log position {LogPosition}", LastIndexedPosition);

src/KurrentDB.SecondaryIndexing/Indexes/Default/DefaultIndexReader.cs

@@ -4,6 +4,7 @@
 using Kurrent.Quack;
 using Kurrent.Quack.ConnectionPool;
 using KurrentDB.Core.Data;
+using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Services;
 using KurrentDB.Core.Services.Storage.ReaderIndex;
 using KurrentDB.SecondaryIndexing.Storage;
@@ -14,8 +15,9 @@ namespace KurrentDB.SecondaryIndexing.Indexes.Default;
 internal class DefaultIndexReader(
 	DuckDBConnectionPool sharedPool,
 	DefaultIndexProcessor processor,
-	IReadIndex<string> index
-) : SecondaryIndexReaderBase(sharedPool, index) {
+	IReadIndex<string> index,
+	DuckDBCpuMetrics cpuMetrics
+) : SecondaryIndexReaderBase(sharedPool, index, cpuMetrics) {
 	protected override string GetId(string indexName) => string.Empty;
 
 	protected override List<IndexQueryRecord> GetDbRecordsForwards(DuckDBConnectionPool db,