fix(api): stream audio store path to eliminate whole-file buffering (OOM)
Processors now emit a ProcessedAudio plan with a streamed writer instead of a whole-file AudioBinary; vault writes stream via RegisterResourceStreamingAsync. Header parsing is bounded. Wave 2 (waveform/Opus) still re-reads the full file by design.
This commit is contained in:
daniel-c-harvey
@@ -0,0 +1,389 @@
|
||||
using System.Text;
|
||||
using DeepDrftContent;
|
||||
using DeepDrftContent.Constants;
|
||||
using DeepDrftContent.FileDatabase.Models;
|
||||
using DeepDrftContent.Processors;
|
||||
using FileDb = DeepDrftContent.FileDatabase.Services.FileDatabase;
|
||||
|
||||
namespace DeepDrftTests;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for the streamed audio store path (Wave 1 OOM fix): processors emit a
|
||||
/// <see cref="ProcessedAudio"/> plan whose body is written to the vault without ever materializing
|
||||
/// the whole file in a managed <c>byte[]</c>. Covers passthrough byte-identity (standard-PCM WAV,
|
||||
/// MP3, FLAC), streamed WAV normalization (EXTENSIBLE), the new streaming vault register round-trip,
|
||||
/// and the memory-bounding contract of the store primitive (bounded-buffer, sequential, forward-only
|
||||
/// writes — never a single whole-file write).
|
||||
/// </summary>
|
||||
[TestFixture]
|
||||
public class AudioStoreStreamingTests
|
||||
{
|
||||
private const ushort WaveFormatPcm = 0x0001;
|
||||
private const ushort WaveFormatExtensible = 0xFFFE;
|
||||
|
||||
private string _testDir = string.Empty;
|
||||
|
||||
[SetUp]
|
||||
public void SetUp()
|
||||
{
|
||||
_testDir = Path.Combine(Path.GetTempPath(), "AudioStoreStreamingTests", Guid.NewGuid().ToString());
|
||||
Directory.CreateDirectory(_testDir);
|
||||
}
|
||||
|
||||
[TearDown]
|
||||
public void TearDown()
|
||||
{
|
||||
try { Directory.Delete(_testDir, recursive: true); }
|
||||
catch { /* Best-effort cleanup — ignore failures */ }
|
||||
}
|
||||
|
||||
private static AudioProcessorRouter Router() =>
|
||||
new(new AudioProcessor(), new Mp3AudioProcessor(), new FlacAudioProcessor());
|
||||
|
||||
private static TrackContentService Content(FileDb db) => new(db, Router());
|
||||
|
||||
// -- End-to-end store byte-identity (passthrough) -----------------------------------------
|
||||
|
||||
[Test]
|
||||
public async Task StandardPcmWav_StoredByteIdenticalToSource()
|
||||
{
|
||||
var source = BuildPcmWav(channels: 2, sampleRate: 44100, bitsPerSample: 16, dataBytes: 200_000);
|
||||
var path = await WriteAsync(source, ".wav");
|
||||
var db = await FileDb.FromAsync(_testDir);
|
||||
|
||||
var content = Content(db!);
|
||||
var entity = await content.AddTrackAsync(path, "Track", "Artist");
|
||||
Assert.That(entity, Is.Not.Null);
|
||||
|
||||
var stored = await content.GetAudioBinaryAsync(entity!.EntryKey);
|
||||
Assert.That(stored, Is.Not.Null);
|
||||
Assert.That(stored!.Buffer, Is.EqualTo(source), "Standard PCM must be stored byte-identical");
|
||||
Assert.That(stored.Duration, Is.GreaterThan(0.0));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task Mp3_StoredByteIdenticalToSource_MetadataCorrect()
|
||||
{
|
||||
var source = BuildMinimalMp3();
|
||||
var path = await WriteAsync(source, ".mp3");
|
||||
var db = await FileDb.FromAsync(_testDir);
|
||||
|
||||
var content = Content(db!);
|
||||
var entity = await content.AddTrackAsync(path, "Track", "Artist");
|
||||
Assert.That(entity, Is.Not.Null);
|
||||
|
||||
var stored = await content.GetAudioBinaryAsync(entity!.EntryKey);
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(stored, Is.Not.Null);
|
||||
Assert.That(stored!.Extension, Is.EqualTo(".mp3"));
|
||||
Assert.That(stored.Buffer, Is.EqualTo(source), "MP3 must be stored byte-identical (no transcode)");
|
||||
Assert.That(stored.Bitrate, Is.EqualTo(128));
|
||||
});
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task Flac_StoredByteIdenticalToSource_MetadataCorrect()
|
||||
{
|
||||
var source = BuildMinimalFlac();
|
||||
var path = await WriteAsync(source, ".flac");
|
||||
var db = await FileDb.FromAsync(_testDir);
|
||||
|
||||
var content = Content(db!);
|
||||
var entity = await content.AddTrackAsync(path, "Track", "Artist");
|
||||
Assert.That(entity, Is.Not.Null);
|
||||
|
||||
var stored = await content.GetAudioBinaryAsync(entity!.EntryKey);
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(stored, Is.Not.Null);
|
||||
Assert.That(stored!.Extension, Is.EqualTo(".flac"));
|
||||
Assert.That(stored.Buffer, Is.EqualTo(source), "FLAC must be stored byte-identical (no transcode)");
|
||||
Assert.That(stored.Duration, Is.GreaterThan(0.0));
|
||||
});
|
||||
}
|
||||
|
||||
// -- Streamed normalization ---------------------------------------------------------------
|
||||
|
||||
[Test]
|
||||
public async Task ExtensibleFloatWav_StoredAsNormalizedStandardPcm()
|
||||
{
|
||||
// A >80 KB float data region forces the streamed transform across multiple bounded chunks.
|
||||
var floatData = BuildFloatRamp(sampleCount: 40_000); // 160 000 bytes in, 120 000 bytes out (24-bit)
|
||||
var source = BuildExtensibleWav(channels: 2, sampleRate: 44100, containerBits: 32, validBits: 32,
|
||||
subFormatTag: 0x0003, sampleData: floatData);
|
||||
var path = await WriteAsync(source, ".wav");
|
||||
var db = await FileDb.FromAsync(_testDir);
|
||||
|
||||
var content = Content(db!);
|
||||
var entity = await content.AddTrackAsync(path, "Track", "Artist");
|
||||
Assert.That(entity, Is.Not.Null);
|
||||
|
||||
var stored = await content.GetAudioBinaryAsync(entity!.EntryKey);
|
||||
Assert.That(stored, Is.Not.Null);
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(BitConverter.ToUInt16(stored!.Buffer, 20), Is.EqualTo(WaveFormatPcm),
|
||||
"EXTENSIBLE must be normalized to standard PCM (audioFormat = 1)");
|
||||
Assert.That(BitConverter.ToUInt16(stored.Buffer, 34), Is.EqualTo(24), "Float must normalize to 24-bit");
|
||||
Assert.That(stored.Buffer.Length, Is.EqualTo(44 + (floatData.Length / 4) * 3),
|
||||
"Output size = 44-byte header + 3 bytes per float sample");
|
||||
});
|
||||
}
|
||||
|
||||
// -- Streaming vault register round-trip ---------------------------------------------------
|
||||
|
||||
[Test]
|
||||
public async Task RegisterResourceStreamingAsync_RoundTrips()
|
||||
{
|
||||
var db = await FileDb.FromAsync(_testDir);
|
||||
await db!.CreateVaultAsync(VaultConstants.Tracks, MediaVaultType.Audio);
|
||||
|
||||
var payload = Enumerable.Range(0, 50_000).Select(i => (byte)(i % 256)).ToArray();
|
||||
var meta = MetaDataFactory.CreateAudioMetaData("entry-1", ".wav", 12.5, 1411);
|
||||
|
||||
var ok = await db.RegisterResourceStreamingAsync(
|
||||
VaultConstants.Tracks, "entry-1", meta,
|
||||
(dest, ct) => dest.WriteAsync(payload, ct).AsTask());
|
||||
Assert.That(ok, Is.True);
|
||||
|
||||
var loaded = await db.LoadResourceAsync<AudioBinary>(VaultConstants.Tracks, "entry-1");
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(loaded, Is.Not.Null);
|
||||
Assert.That(loaded!.Buffer, Is.EqualTo(payload), "Streamed bytes must round-trip exactly");
|
||||
Assert.That(loaded.Duration, Is.EqualTo(12.5));
|
||||
Assert.That(loaded.Bitrate, Is.EqualTo(1411));
|
||||
});
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task RegisterResourceStreamingAsync_UnknownVault_ReturnsFalse()
|
||||
{
|
||||
var db = await FileDb.FromAsync(_testDir);
|
||||
var meta = MetaDataFactory.CreateAudioMetaData("e", ".wav", 1.0, 1411);
|
||||
|
||||
var ok = await db!.RegisterResourceStreamingAsync(
|
||||
"does-not-exist", "e", meta, (dest, ct) => Task.CompletedTask);
|
||||
|
||||
Assert.That(ok, Is.False, "Register into a missing vault must swallow and return false");
|
||||
}
|
||||
|
||||
// -- Memory-bounding of the store primitive -----------------------------------------------
|
||||
|
||||
[Test]
|
||||
public async Task WriteToAsync_StreamsInBoundedSequentialChunks_NotOneWholeFileWrite()
|
||||
{
|
||||
// A multi-hundred-KB passthrough file: a buffered implementation would issue one giant write of
|
||||
// the whole body; the streamed primitive must write in bounded, forward-only chunks.
|
||||
var source = BuildPcmWav(channels: 2, sampleRate: 44100, bitsPerSample: 16, dataBytes: 600_000);
|
||||
var path = await WriteAsync(source, ".wav");
|
||||
|
||||
var processed = await new AudioProcessor().ProcessWavFileAsync(path);
|
||||
Assert.That(processed, Is.Not.Null);
|
||||
|
||||
var probe = new BoundedWriteProbeStream();
|
||||
await processed!.WriteToAsync(probe);
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(probe.TotalBytes, Is.EqualTo(source.Length), "All bytes must be written");
|
||||
Assert.That(probe.WriteCount, Is.GreaterThan(1), "Body must be streamed in multiple chunks");
|
||||
Assert.That(probe.MaxWriteSize, Is.LessThanOrEqualTo(81920),
|
||||
"No single write may exceed the bounded buffer — i.e. the whole file is never buffered");
|
||||
});
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task WriteToAsync_NormalizedFloat_StreamsToForwardOnlyStream()
|
||||
{
|
||||
// The normalized path seeks the *source* but must only write the destination sequentially.
|
||||
var floatData = BuildFloatRamp(sampleCount: 30_000);
|
||||
var source = BuildExtensibleWav(channels: 2, sampleRate: 44100, containerBits: 32, validBits: 32,
|
||||
subFormatTag: 0x0003, sampleData: floatData);
|
||||
var path = await WriteAsync(source, ".wav");
|
||||
|
||||
var processed = await new AudioProcessor().ProcessWavFileAsync(path);
|
||||
Assert.That(processed, Is.Not.Null);
|
||||
|
||||
var probe = new BoundedWriteProbeStream();
|
||||
await processed!.WriteToAsync(probe);
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(probe.TotalBytes, Is.EqualTo(44 + (floatData.Length / 4) * 3));
|
||||
Assert.That(probe.WriteCount, Is.GreaterThan(1), "Header + multiple transformed body chunks");
|
||||
Assert.That(probe.MaxWriteSize, Is.LessThanOrEqualTo(81920));
|
||||
});
|
||||
}
|
||||
|
||||
// -- builders -----------------------------------------------------------------------------
|
||||
|
||||
private async Task<string> WriteAsync(byte[] bytes, string extension)
|
||||
{
|
||||
var path = Path.Combine(_testDir, Guid.NewGuid().ToString("N") + extension);
|
||||
await File.WriteAllBytesAsync(path, bytes);
|
||||
return path;
|
||||
}
|
||||
|
||||
private static byte[] BuildPcmWav(int channels, int sampleRate, int bitsPerSample, int dataBytes)
|
||||
{
|
||||
var blockAlign = (ushort)(channels * (bitsPerSample / 8));
|
||||
var byteRate = (uint)(sampleRate * blockAlign);
|
||||
var data = new byte[dataBytes];
|
||||
for (var i = 0; i < data.Length; i++) data[i] = (byte)(i % 251);
|
||||
|
||||
using var ms = new MemoryStream();
|
||||
using var w = new BinaryWriter(ms, Encoding.ASCII, leaveOpen: true);
|
||||
w.Write(Encoding.ASCII.GetBytes("RIFF"));
|
||||
w.Write((uint)(36 + data.Length));
|
||||
w.Write(Encoding.ASCII.GetBytes("WAVE"));
|
||||
w.Write(Encoding.ASCII.GetBytes("fmt "));
|
||||
w.Write(16u);
|
||||
w.Write(WaveFormatPcm);
|
||||
w.Write((ushort)channels);
|
||||
w.Write((uint)sampleRate);
|
||||
w.Write(byteRate);
|
||||
w.Write(blockAlign);
|
||||
w.Write((ushort)bitsPerSample);
|
||||
w.Write(Encoding.ASCII.GetBytes("data"));
|
||||
w.Write((uint)data.Length);
|
||||
w.Write(data);
|
||||
w.Flush();
|
||||
return ms.ToArray();
|
||||
}
|
||||
|
||||
private static byte[] BuildExtensibleWav(
|
||||
int channels, int sampleRate, int containerBits, int validBits, ushort subFormatTag, byte[] sampleData)
|
||||
{
|
||||
var blockAlign = (ushort)(channels * (containerBits / 8));
|
||||
var byteRate = (uint)(sampleRate * blockAlign);
|
||||
const uint fmtChunkSize = 40;
|
||||
|
||||
using var ms = new MemoryStream();
|
||||
using var w = new BinaryWriter(ms, Encoding.ASCII, leaveOpen: true);
|
||||
w.Write(Encoding.ASCII.GetBytes("RIFF"));
|
||||
w.Write((uint)(36 + (fmtChunkSize - 16) + sampleData.Length));
|
||||
w.Write(Encoding.ASCII.GetBytes("WAVE"));
|
||||
w.Write(Encoding.ASCII.GetBytes("fmt "));
|
||||
w.Write(fmtChunkSize);
|
||||
w.Write(WaveFormatExtensible);
|
||||
w.Write((ushort)channels);
|
||||
w.Write((uint)sampleRate);
|
||||
w.Write(byteRate);
|
||||
w.Write(blockAlign);
|
||||
w.Write((ushort)containerBits);
|
||||
w.Write((ushort)22); // cbSize
|
||||
w.Write((ushort)validBits); // wValidBitsPerSample
|
||||
w.Write((uint)0); // channel mask
|
||||
var guid = new byte[16];
|
||||
guid[0] = (byte)(subFormatTag & 0xFF);
|
||||
guid[1] = (byte)((subFormatTag >> 8) & 0xFF);
|
||||
w.Write(guid);
|
||||
w.Write(Encoding.ASCII.GetBytes("data"));
|
||||
w.Write((uint)sampleData.Length);
|
||||
w.Write(sampleData);
|
||||
w.Flush();
|
||||
return ms.ToArray();
|
||||
}
|
||||
|
||||
private static byte[] BuildFloatRamp(int sampleCount)
|
||||
{
|
||||
var bytes = new byte[sampleCount * 4];
|
||||
for (var i = 0; i < sampleCount; i++)
|
||||
{
|
||||
var sample = (float)((i % 200) / 200.0 - 0.5); // a deterministic ramp in [-0.5, 0.5)
|
||||
BitConverter.GetBytes(sample).CopyTo(bytes, i * 4);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
private static byte[] BuildMinimalMp3()
|
||||
{
|
||||
// One MPEG1 Layer III CBR frame: 128 kbps, 44.1 kHz, stereo. Body zero-filled (silence).
|
||||
const int frameSize = 417; // floor(144 * 128000 / 44100)
|
||||
var buffer = new byte[frameSize];
|
||||
buffer[0] = 0xFF;
|
||||
buffer[1] = 0xFB; // sync + MPEG1 + Layer III + no CRC
|
||||
buffer[2] = (byte)((9 << 4) | (0 << 2)); // bitrate index 9 (128 kbps), sample-rate index 0 (44.1 kHz)
|
||||
buffer[3] = 0x00; // stereo
|
||||
return buffer;
|
||||
}
|
||||
|
||||
private static byte[] BuildMinimalFlac()
|
||||
{
|
||||
using var ms = new MemoryStream();
|
||||
ms.Write(Encoding.ASCII.GetBytes("fLaC"));
|
||||
ms.WriteByte(0x80); // last block, STREAMINFO
|
||||
ms.WriteByte(0x00);
|
||||
ms.WriteByte(0x00);
|
||||
ms.WriteByte(34);
|
||||
|
||||
var s = new byte[34];
|
||||
const int sampleRate = 44100;
|
||||
const int channels = 2;
|
||||
const int bitsPerSample = 16;
|
||||
const long totalSamples = 44100L * 5;
|
||||
s[10] = (byte)((sampleRate >> 12) & 0xFF);
|
||||
s[11] = (byte)((sampleRate >> 4) & 0xFF);
|
||||
var bps = bitsPerSample - 1;
|
||||
s[12] = (byte)(((sampleRate & 0x0F) << 4) | (((channels - 1) & 0x07) << 1) | ((bps >> 4) & 0x01));
|
||||
s[13] = (byte)(((bps & 0x0F) << 4) | (int)((totalSamples >> 32) & 0x0F));
|
||||
s[14] = (byte)((totalSamples >> 24) & 0xFF);
|
||||
s[15] = (byte)((totalSamples >> 16) & 0xFF);
|
||||
s[16] = (byte)((totalSamples >> 8) & 0xFF);
|
||||
s[17] = (byte)(totalSamples & 0xFF);
|
||||
ms.Write(s);
|
||||
|
||||
// Trailing zero bytes standing in for encoded frames (affect only the average-bitrate compute).
|
||||
ms.Write(new byte[100_000]);
|
||||
return ms.ToArray();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A write-only, forward-only (non-seekable) stream that records how the store primitive writes:
|
||||
/// the number of writes, the largest single write, and the total. Proves the body is streamed in
|
||||
/// bounded, sequential chunks rather than buffered into one whole-file write.
|
||||
/// </summary>
|
||||
private sealed class BoundedWriteProbeStream : Stream
|
||||
{
|
||||
public int WriteCount { get; private set; }
|
||||
public int MaxWriteSize { get; private set; }
|
||||
public long TotalBytes { get; private set; }
|
||||
|
||||
public override bool CanRead => false;
|
||||
public override bool CanSeek => false;
|
||||
public override bool CanWrite => true;
|
||||
public override long Length => TotalBytes;
|
||||
public override long Position { get => TotalBytes; set => throw new NotSupportedException(); }
|
||||
|
||||
private void Record(int count)
|
||||
{
|
||||
WriteCount++;
|
||||
if (count > MaxWriteSize) MaxWriteSize = count;
|
||||
TotalBytes += count;
|
||||
}
|
||||
|
||||
public override void Write(byte[] buffer, int offset, int count) => Record(count);
|
||||
|
||||
public override void Write(ReadOnlySpan<byte> buffer) => Record(buffer.Length);
|
||||
|
||||
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Record(buffer.Length);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||
{
|
||||
Record(count);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public override void Flush() { }
|
||||
public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
||||
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
|
||||
public override void SetLength(long value) => throw new NotSupportedException();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user