BatchingLogRecordProcessor does not flush on shutdown

The source states that LogRecordProcessor.shutdown includes the effect of forceFlush. On the other hand in BatchLogRecordProcessor the flushing thread skips the last flush since the task.cancel in shutdown causes every cancelable operation from that task to fail with error.Canceled:

pub const BatchingLogRecordProcessor = struct {

    // ...

    fn shutdown(ctx: *anyopaque) anyerror!void {
        // ...

        // Cancel the background task (unblocks its wait and waits for it to finish)
        if (self.export_task) |*task| {
            task.cancel(self.io); // <---- exportLoop is canceled
            self.export_task = null;
        }
    }

    // ...

    fn exportLoop(self: *Self) void {
        while (true) {
            self.mutex.lockUncancelable(self.io);
            if (self.should_shutdown.load(.acquire)) {
                while (self.queue.len > 0) {
                    if (!self.exportBatch()) break; // <--- exportBatch may be canceled
                }
                self.mutex.unlock(self.io);
                break;
            }

            // ...
        }
    }

    /// Must be called while holding the mutex
    fn exportBatch(self: *Self) bool {

        // ...

        // Export the batch (unlock mutex during export to allow concurrent onEmit calls)
        self.mutex.unlock(self.io);
        self.exporter.exportLogs(logs_to_export) catch |err| { // <--- the last exportLogs may be canceled, causing the last flush to be lost
            // ...
        };
        self.mutex.lockUncancelable(self.io);

        // ...
    }

    // ...
};

You can reproduce this problem with the following script:

const std = @import("std");
const sdk = @import("opentelemetry-sdk");

const N = 6;

const Exporter = struct {
    io: std.Io,
    got: usize = 0,

    fn iface(self: *Exporter) sdk.logs.LogRecordExporter {
        return .{ .ptr = self, .vtable = &.{ .exportLogsFn = exp, .shutdownFn = nop } };
    }

    // Count exported records
    fn exp(ctx: *anyopaque, recs: []sdk.logs.ReadbleLogRecord) anyerror!void {
        const self: *Exporter = @ptrCast(@alignCast(ctx));
        var never_set: std.Io.Event = .unset;
        const in_1ms: std.Io.Timeout = .{ .duration = .{ .raw = .{ .nanoseconds = std.time.ns_per_ms }, .clock = .awake } };
        never_set.waitTimeout(self.io, in_1ms) catch |e| switch (e) {
            error.Timeout => {},
            else => return e,
        };
        self.got += recs.len;
    }

    fn nop(_: *anyopaque) anyerror!void {}
};

fn run(gpa: std.mem.Allocator, io: std.Io, flush: bool) !usize {
    var ex = Exporter{ .io = io };
    var proc = try sdk.logs.BatchingLogRecordProcessor.init(gpa, io, ex.iface(), .{
        // the background loop can never fire, so forceFlush/shutdown is the only export
        .max_export_batch_size = N + 1,
        .scheduled_delay_millis = 60_000,
    });

    var provider = try sdk.logs.LoggerProvider.init(gpa, io, null);
    defer provider.deinit();
    try provider.addLogRecordProcessor(proc.asLogRecordProcessor());

    const logger = try provider.getLogger(.{ .name = "repro" });
    for (0..N) |_| {
        logger.emit(.info, "record", .{});
    }

    if (flush) {
        try provider.forceFlush();
    }

    try provider.shutdown();
    proc.deinit();
    return ex.got;
}

pub fn main(init: std.process.Init) !void {
    const without = try run(init.gpa, init.io, false);
    const with = try run(init.gpa, init.io, true);

    std.debug.print("no explicit flush:   {d}/{d}\n", .{ without, N });
    std.debug.print("with explicit flush: {d}/{d}\n", .{ with, N });

    if (without < N) std.process.exit(1);
}

Which prints:

error: BatchingLogRecordProcessor failed to export log batch: error.Canceled
no explicit flush:   0/6
with explicit flush: 6/6

Traces have the same problem, where shutdown cancels the worker causing exportSpans to fail

Start with opentelemetry-sdk/src/sdk/logs/log_record_processor.zig, especially BatchingLogRecordProcessor.shutdown, exportLoop, and exportBatch, and run the supplied reproduction to observe the canceled final export. Then inspect opentelemetry-sdk/src/sdk/trace/span_processor.zig, including shutdown and exportSpans. Done means shutdown performs the required final flush for both logs and traces without requiring an explicit forceFlush.