From 04047501345ed4d99b31b8d031663910edd6100f Mon Sep 17 00:00:00 2001 From: Ahmed Yasin Koculu Date: Thu, 15 Oct 2026 22:26:20 +0200 Subject: [PATCH 1/2] Write 12 edge case regression tests and fix the bugs. --- .../EngineTests.cs | 4 +- .../PublicStorageCodecTests.cs | 19 +- .../StorageFormatRegressionTests.cs | 281 ++++++++++++++++++ .../Ingestion/IngestionWorker.cs | 2 +- src/ZoneTree.Analytics/Internal/Text.cs | 46 ++- .../Privacy/VisitorHasher.cs | 90 ++++-- .../Storage/DimensionIndexWriter.cs | 2 +- .../Storage/JointBackfillCheckpoint.cs | 264 +++++++++++++++- .../Storage/PartitionedStore.cs | 157 ++++++++-- .../Storage/RawEventCodec.cs | 5 + .../Storage/RollupIndexStore.cs | 2 +- src/ZoneTree.Analytics/Storage/RollupStore.cs | 2 +- .../Storage/SessionState.cs | 5 +- 13 files changed, 799 insertions(+), 80 deletions(-) create mode 100644 src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs diff --git a/src/ZoneTree.Analytics.UnitTests/EngineTests.cs b/src/ZoneTree.Analytics.UnitTests/EngineTests.cs index aa10f68..32aeca4 100644 --- a/src/ZoneTree.Analytics.UnitTests/EngineTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/EngineTests.cs @@ -659,8 +659,8 @@ public void DataSurvivesRestartAndSessionsContinue() Assert.That(totals[Metric.Sessions], Is.EqualTo(5)); }); - // Alice continues the same session after the restart. Anonymous visitors of past days - // cannot continue: their daily salt is destroyed by design. + // Alice continues the same session after the restart. Anonymous visitors continue only + // while the salt of their day is kept: it is destroyed by design. engine.Track(new AnalyticsEvent(StandardEvents.PageView) { Url = "/b", VisitorId = "alice", Timestamp = Utc(T0.AddMinutes(5)) }); engine.Flush(); var alice = Query(q => diff --git a/src/ZoneTree.Analytics.UnitTests/PublicStorageCodecTests.cs b/src/ZoneTree.Analytics.UnitTests/PublicStorageCodecTests.cs index d3684b2..91b496d 100644 --- a/src/ZoneTree.Analytics.UnitTests/PublicStorageCodecTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/PublicStorageCodecTests.cs @@ -1,4 +1,5 @@ using System.Text; +using System.Text.RegularExpressions; using ZoneTree.Analytics.Storage; namespace ZoneTree.Analytics.UnitTests; @@ -189,25 +190,27 @@ public void RawValueDecoderRejectsInvalidUtf8OverflowingIntegersAndInvalidSessio public void RollupKeyDecoderRejectsTruncatedHeaders(string header) => Assert.Throws(() => StorageKeyCodec.DecodeRollup(Encoding.UTF8.GetBytes(header))); - [TestCase("page", "a\u001Fb")] + // Control characters are escaped in the attributes: the test adapter does not run cases whose + // arguments contain them. + [TestCase("page", @"a\u001Fb")] [TestCase("", "unexpected")] [TestCase("a,b", "one")] - [TestCase("b,a", "one\u001Ftwo")] - [TestCase("a,a", "one\u001Ftwo")] - [TestCase("a,", "one\u001Ftwo")] + [TestCase("b,a", @"one\u001Ftwo")] + [TestCase("a,a", @"one\u001Ftwo")] + [TestCase("a,", @"one\u001Ftwo")] public void RollupKeyDecoderRejectsAmbiguousDimensionLayouts(string names, string values) { - var bytes = KeyCodec.RollupKey("s", "e", names, 100, values); + var bytes = KeyCodec.RollupKey("s", "e", names, 100, Regex.Unescape(values)); Assert.Throws(() => StorageKeyCodec.DecodeRollup(bytes.Span)); } [TestCase("a\0b", "value")] [TestCase("a,b", "value")] - [TestCase("page", "a\u001Fb")] - [TestCase("page", "\u0001")] + [TestCase("page", @"a\u001Fb")] + [TestCase("page", @"\u0001")] public void KeyEncoderRejectsReservedCharacters(string dimension, string value) { - var key = new RollupKey { Site = "s", Event = "e", BucketTicks = 100, Dimensions = { [dimension] = value } }; + var key = new RollupKey { Site = "s", Event = "e", BucketTicks = 100, Dimensions = { [dimension] = Regex.Unescape(value) } }; Assert.Throws(() => StorageKeyCodec.EncodeRollup(key)); Assert.Throws(() => StorageKeyCodec.EncodeRawEvent(new RawEventKey("s\0e", 100, 1))); } diff --git a/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs b/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs new file mode 100644 index 0000000..d9bc3df --- /dev/null +++ b/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs @@ -0,0 +1,281 @@ +using System.Globalization; +using System.Text.RegularExpressions; +using ZoneTree.Analytics.Privacy; +using ZoneTree.Analytics.Query; +using ZoneTree.Analytics.Storage; + +namespace ZoneTree.Analytics.UnitTests; + +/// +/// Regressions of the storage format review: each test fails on the format before its fix. +/// +public sealed class StorageFormatRegressionTests : DirectoryTest +{ + static readonly DateTime First = DateTime.UtcNow.Date.AddDays(-20).AddHours(10); + + static string DayName(DateTime value) => value.ToString("yyyy-MM-dd", CultureInfo.InvariantCulture); + + string[] Partitions(params string[] path) => + Directory.GetDirectories(Path.Combine([directory, .. path])).Select(Path.GetFileName).Order().ToArray(); + + AnalyticsEngine RawPartitionsOfOneRecord() => TestEngine.Create(directory, o => + { + o.TimeZone = TimeZoneInfo.Utc; + o.Storage.Partitioning.RawEvents = 1; + }); + + static void View(AnalyticsEngine engine, DateTime time) + { + Assert.That(engine.Track(new AnalyticsEvent(StandardEvents.PageView) { Url = "/", VisitorId = "v", Timestamp = Utc(time) }), Is.True); + engine.Flush(); + } + + static double RawPageViews(AnalyticsEngine engine, DateTime from, DateTime to) => engine.Query(new AnalyticsQuery + { + Range = Range(from, to), Source = QuerySource.RawEvents, Metrics = { Metric.PageViews } + }).Totals[Metric.PageViews]; + + [Test] + public void RolloverNeverHidesRecordsWrittenAfterItsClock() + { + // A record from a clock that ran ahead: later than the housekeeping clock below. + var ahead = First.AddDays(5); + using (var engine = RawPartitionsOfOneRecord()) + { + View(engine, First); + View(engine, ahead); + } + // The high-water mark of the partition survives reopening. + using var reopened = RawPartitionsOfOneRecord(); + View(reopened, First.AddMinutes(1)); + Assert.That(reopened.RollOverPartitions(First.Ticks), Is.EqualTo(1)); + Assert.Multiple(() => + { + Assert.That(Partitions("events"), Is.EqualTo(new[] { DayName(First), DayName(ahead.Date.AddDays(1)) })); + Assert.That(RawPageViews(reopened, ahead.Date, ahead.Date.AddDays(1)), Is.EqualTo(1)); + Assert.That(RawPageViews(reopened, First.Date, ahead.Date.AddDays(1)), Is.EqualTo(3)); + }); + } + + [Test] + public void RolloverResumesAfterAClockThatRanAheadCreatedAnEmptyFuturePartition() + { + using var engine = RawPartitionsOfOneRecord(); + View(engine, First); + // Housekeeping ran while the clock read ten days later. + Assert.That(engine.RollOverPartitions(First.AddDays(10).Ticks), Is.EqualTo(1)); + var future = First.Date.AddDays(12); + // The clock is correct again: events still go to the first partition. + View(engine, First.AddMinutes(1)); + Assert.That(engine.RollOverPartitions(First.Ticks), Is.EqualTo(1), "the partition receiving events rolls over"); + var next = First.Date.AddDays(2); + View(engine, next.AddMinutes(1)); + Assert.Multiple(() => + { + Assert.That(Partitions("events"), Is.EqualTo(new[] { DayName(First), DayName(next), DayName(future) })); + Assert.That(engine.GetStatistics().Partitions.Single(x => x.Directory == "events/" + DayName(next)).ApproximateRecordCount, Is.EqualTo(1)); + Assert.That(RawPageViews(engine, First.Date, future.AddDays(1)), Is.EqualTo(3)); + }); + } + + [Test] + public void AnonymousHistoricalVisitorsKeepTheirSaltAcrossRotationAndRestart() + { + var day = DateTime.UtcNow.Date.AddDays(-10); + void Visit(AnalyticsEngine engine, int hour) + { + engine.Track(new AnalyticsEvent(StandardEvents.PageView) + { + Url = "/", IpAddress = "10.0.0.9", UserAgent = UserAgents.ChromeWindows, Timestamp = Utc(day.AddHours(hour)) + }); + engine.Flush(); + } + using (var engine = TestEngine.Create(directory)) + { + Visit(engine, 1); + // Retention rotates the salts while the backfill continues. + engine.ApplyRetention(); + Visit(engine, 2); + } + using var reopened = TestEngine.Create(directory); + Visit(reopened, 3); + foreach (var source in new[] { QuerySource.RawEvents, QuerySource.Rollups }) + { + var result = reopened.Query(new AnalyticsQuery { Range = Day(day), Source = source, Metrics = { Metric.Visitors, Metric.PageViews } }); + Assert.That(result.Totals[Metric.PageViews], Is.EqualTo(3), source.ToString()); + Assert.That(result.Totals[Metric.Visitors], Is.EqualTo(1), source.ToString()); + } + } + + [Test] + public void UnusedHistoricalSaltsAreDestroyedADayAfterTheirLastUse() + { + var meta = new MetaStore(Path.Combine(directory, "meta")); + var hasher = new VisitorHasher(VisitorIdentification.DailyRotatingHash, null, meta); + var old = DateTime.UtcNow.Date.AddDays(-10); + var saltKey = "salt:" + DayName(old); + var hash = hasher.Compute("s", null, "ip", "ua", old.AddHours(1).Ticks); + hasher.RotateSalts(DateTime.UtcNow); + Assert.Multiple(() => + { + Assert.That(meta.Get(saltKey), Is.Not.Null, "a salt in use survives rotation"); + Assert.That(new VisitorHasher(VisitorIdentification.DailyRotatingHash, null, new MetaStore(Path.Combine(directory, "meta"))) + .Compute("s", null, "ip", "ua", old.AddHours(2).Ticks), Is.EqualTo(hash), "and reopening"); + }); + hasher.RotateSalts(DateTime.UtcNow.AddDays(2)); + Assert.Multiple(() => + { + Assert.That(meta.ListByPrefix("salt").Any(x => x.Key.EndsWith(DayName(old), StringComparison.Ordinal)), Is.False); + Assert.That(hasher.Compute("s", null, "ip", "ua", old.AddHours(3).Ticks), Is.Not.EqualTo(hash), "the destroyed salt is not reused"); + }); + } + + [Test] + public void InvalidUtf16IsCountedAsTheValueItIsStoredAs() + { + var day = DateTime.UtcNow.Date.AddDays(-2); + using var engine = TestEngine.Create(directory, o => + { + o.Rollups.Definitions.Add(new RollupDefinition("plan")); + o.Rollups.DistinctDimensions.Add("plan"); + }); + // Two different lone surrogates both encode as U+FFFD in UTF-8. + foreach (var plan in new[] { "x" + (char)0xD800, "x" + (char)0xDBFF }) + { + engine.Track(new AnalyticsEvent("purchase") + { + VisitorId = "v", Timestamp = Utc(day.AddHours(1)), Dimensions = { ["plan"] = plan } + }); + } + engine.Flush(); + foreach (var source in new[] { QuerySource.Rollups, QuerySource.RawEvents }) + { + var distinct = engine.Query(new AnalyticsQuery + { + Event = "purchase", Range = Day(day), Source = source, Metrics = { Metric.Distinct("plan") } + }); + var grouped = engine.Query(new AnalyticsQuery + { + Event = "purchase", Range = Day(day), Source = source, GroupBy = { "plan" }, Metrics = { Metric.Events } + }); + Assert.Multiple(() => + { + Assert.That(distinct.Totals[Metric.Distinct("plan")], Is.EqualTo(1), source.ToString()); + Assert.That(grouped.Rows.Single()["plan"], Is.EqualTo("x�"), source.ToString()); + Assert.That(grouped.Rows.Single()[Metric.Events], Is.EqualTo(2), source.ToString()); + }); + } + } + + // Escaped: the test adapter does not run cases whose arguments contain control characters. + [TestCase(@"a\u001Fb")] + [TestCase(@"\u0001")] + [TestCase(@"tab\tvalue")] + public void RawCodecRejectsControlCharactersReservedByRollupKeys(string value) + { + value = Regex.Unescape(value); + var raw = new RawEvent { Site = "s", Ticks = 1000, Sequence = 1, Name = "e", Dimensions = { new("plan", value) } }; + Assert.Throws(() => RawEventCodec.Encode(raw)); + // Bytes written without the public encoder do not decode either. + var bytes = RawEventCodec.Encode("e", 1000, 0, 0, 0, false, [new("plan", value)], []); + Assert.Throws(() => RawEventCodec.Decode(RawEventCodec.EncodeKey(raw).Span, bytes.Span)); + } + + [Test] + public void SessionStoreAcceptsAStateWithoutDimensions() + { + using (var engine = TestEngine.Create(directory)) + engine.Storage.Sessions.Upsert(42, new SessionState { SessionId = 7, ExpiresAtUtcTicks = DateTime.UtcNow.AddHours(1).Ticks }); + using var reopened = TestEngine.Create(directory); + Assert.That(reopened.Storage.Sessions.TryGet(42, out var stored), Is.True); + Assert.Multiple(() => + { + Assert.That(stored.SessionId, Is.EqualTo(7)); + Assert.That(stored.Dimensions, Has.Length.EqualTo(StandardDimensions.SessionLevel.Count)); + Assert.That(stored.Dimensions, Is.All.Null); + }); + } + + [Test] + public void JointBackfillJournalRoundTripsEveryField() + { + var checkpoint = new JointBackfillCheckpoint(directory, "configuration", 5); + var state = checkpoint.State; + state.Copying = true; + state.Site = "shop"; + state.Partition = "2026-10-15"; + state.Key = [1, 2, 3]; + state.Events = 10; + state.Sessions = 11; + state.Updates = 12; + state.Flushes = 13; + state.Sites = 2; + state.Active[9] = new RawSession + { + Start = 100, Last = 200, PageViews = 3, Interactions = 4, Visitor = 77, Allowed = true, + Dimensions = new(StringComparer.Ordinal) { ["source"] = "GitHub", ["country"] = null }, + HasStartEvent = true, FirstIsPageView = true, FirstPage = "/a", FirstHostname = null, + LastPage = "/b", LastHostname = "example.com", LastPageTicks = 150 + }; + state.Active[10] = new RawSession { Start = 1, Last = 2 }; + state.Redo.Add(new JointBackfillReplacement { Granularity = Granularity.Day, Partition = "2026-10-14", Key = [4], Value = [5, 6] }); + checkpoint.Save(); + + var reopened = new JointBackfillCheckpoint(directory, "configuration", 99); + var loaded = reopened.State; + var session = loaded.Active[9]; + Assert.Multiple(() => + { + Assert.That(reopened.Resuming, Is.True); + Assert.That((loaded.Copying, loaded.Coverage, loaded.Site, loaded.SiteCompleted, loaded.Partition), Is.EqualTo((true, 5L, "shop", false, "2026-10-15"))); + Assert.That(loaded.Key, Is.EqualTo(new byte[] { 1, 2, 3 })); + Assert.That((loaded.Events, loaded.Sessions, loaded.Updates, loaded.Flushes, loaded.Sites), Is.EqualTo((10L, 11L, 12L, 13L, 2))); + Assert.That((session.Start, session.Last, session.PageViews, session.Interactions, session.Visitor, session.Allowed), Is.EqualTo((100L, 200L, 3, 4, 77UL, true))); + Assert.That(session.Dimensions, Is.EqualTo(state.Active[9].Dimensions)); + Assert.That((session.HasStartEvent, session.FirstIsPageView, session.FirstPage, session.FirstHostname), Is.EqualTo((true, true, "/a", (string)null))); + Assert.That((session.LastPage, session.LastHostname, session.LastPageTicks), Is.EqualTo(("/b", "example.com", 150L))); + Assert.That(loaded.Active[10].Dimensions, Is.Null); + Assert.That(loaded.Redo.Single().Granularity, Is.EqualTo(Granularity.Day)); + Assert.That(loaded.Redo.Single().Partition, Is.EqualTo("2026-10-14")); + Assert.That(loaded.Redo.Single().Key, Is.EqualTo(new byte[] { 4 })); + Assert.That(loaded.Redo.Single().Value, Is.EqualTo(new byte[] { 5, 6 })); + Assert.That(new JointBackfillCheckpoint(directory, "other", 5).Resuming, Is.False); + }); + reopened.Complete(); + Assert.That(new JointBackfillCheckpoint(directory, "configuration", 5).Resuming, Is.False); + } + + [Test] + public void JointBackfillJournalHoldsAFlushOfDenseRows() + { + // 7,500 rows with one dense 16,386-byte visitor sketch each: about 123 MB as bytes, more + // than the largest string a JSON writer accepts once encoded as base64 inside JSON. + const int rows = 7_500; + var value = new byte[16_386]; + new Random(1).NextBytes(value); + var checkpoint = new JointBackfillCheckpoint(directory, "configuration", 0); + for (var i = 0; i < rows; ++i) + checkpoint.State.Redo.Add(new JointBackfillReplacement { Granularity = Granularity.Hour, Partition = "2026-10-15", Key = BitConverter.GetBytes(i), Value = value }); + checkpoint.Save(); + var journal = Directory.GetFiles(Path.Combine(directory, "backfill", "joint-rollups")).Sum(path => new FileInfo(path).Length); + Assert.That(journal, Is.LessThan(rows * (value.Length + 64L)), "the journal stores bytes, not text"); + var redo = new JointBackfillCheckpoint(directory, "configuration", 0).State.Redo; + Assert.That(redo, Has.Count.EqualTo(rows)); + Assert.That(redo[^1].Key, Is.EqualTo(BitConverter.GetBytes(rows - 1))); + Assert.That(redo[^1].Value, Is.EqualTo(value)); + } + + [Test] + public void ATornJointBackfillJournalFallsBackToThePreviousCheckpoint() + { + var checkpoint = new JointBackfillCheckpoint(directory, "configuration", 0); + checkpoint.State.Events = 1; + checkpoint.Save(); + checkpoint.State.Events = 2; + checkpoint.Save(); + var newest = Directory.GetFiles(Path.Combine(directory, "backfill", "joint-rollups")).OrderByDescending(File.GetLastWriteTimeUtc).First(); + var bytes = File.ReadAllBytes(newest); + File.WriteAllBytes(newest, bytes[..(bytes.Length - 1)]); + Assert.That(new JointBackfillCheckpoint(directory, "configuration", 0).State.Events, Is.EqualTo(1)); + } +} diff --git a/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs b/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs index c8baeea..58e39a9 100644 --- a/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs +++ b/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs @@ -516,7 +516,7 @@ void EmitSessionDeltas(SessionState session) void WriteRawEvent(EnrichedEvent ev, SessionState session) { var store = services.RawEvents.Store; - var partition = store.Route(ev.Ticks); + var partition = store.RouteWrite(ev.Ticks); if (rawLease == null || rawLease.Name != partition) { rawLease?.Dispose(); diff --git a/src/ZoneTree.Analytics/Internal/Text.cs b/src/ZoneTree.Analytics/Internal/Text.cs index 220722e..d5b20fe 100644 --- a/src/ZoneTree.Analytics/Internal/Text.cs +++ b/src/ZoneTree.Analytics/Internal/Text.cs @@ -39,7 +39,7 @@ public static void ValidateUnicode(string value, string parameter) /// /// Sanitizes a name (event, site, dimension or value name): removes control characters, - /// trims and truncates. Returns empty for null. + /// replaces unpaired surrogates, trims and truncates. Returns empty for null. /// public static string Name(string value) { @@ -47,7 +47,43 @@ public static string Name(string value) return ""; if (ContainsControl(value)) value = string.Concat(value.Where(c => !char.IsControl(c))); - return Truncate(value.Trim(), MaxNameLength); + return Truncate(ReplaceUnpairedSurrogates(value).Trim(), MaxNameLength); + } + + /// + /// Replaces each unpaired surrogate with U+FFFD, as UTF-8 encoding does when the text is + /// stored. Sanitized text then equals the text read back, so its hashes and in-memory keys + /// agree with the stored bytes. + /// + static string ReplaceUnpairedSurrogates(string value) + { + var first = value.AsSpan().IndexOfAnyInRange('\uD800', '\uDFFF'); + if (first < 0) + return value; + var index = IndexOfUnpairedSurrogate(value, first); + if (index < 0) + return value; + var chars = value.ToCharArray(); + for (; index >= 0; index = IndexOfUnpairedSurrogate(value, index + 1)) + chars[index] = '�'; + return new string(chars); + } + + static int IndexOfUnpairedSurrogate(string value, int start) + { + for (var i = start; i < value.Length; ++i) + { + var c = value[i]; + if (!char.IsSurrogate(c)) + continue; + if (char.IsHighSurrogate(c) && i + 1 < value.Length && char.IsLowSurrogate(value[i + 1])) + { + ++i; + continue; + } + return i; + } + return -1; } /// @@ -66,8 +102,8 @@ public static ulong StableHash(string value) } /// - /// Sanitizes a dimension value: replaces control characters with spaces, - /// trims and truncates. Returns empty for null. + /// Sanitizes a dimension value: replaces control characters with spaces and unpaired + /// surrogates with U+FFFD, trims and truncates. Returns empty for null. /// public static string Value(string value, int maxLength) { @@ -81,7 +117,7 @@ public static string Value(string value, int maxLength) span[i] = char.IsControl(source[i]) ? ' ' : source[i]; }); } - return Truncate(value.Trim(), maxLength); + return Truncate(ReplaceUnpairedSurrogates(value).Trim(), maxLength); } static bool ContainsControl(string value) diff --git a/src/ZoneTree.Analytics/Privacy/VisitorHasher.cs b/src/ZoneTree.Analytics/Privacy/VisitorHasher.cs index 8eb82ce..945bb60 100644 --- a/src/ZoneTree.Analytics/Privacy/VisitorHasher.cs +++ b/src/ZoneTree.Analytics/Privacy/VisitorHasher.cs @@ -15,8 +15,10 @@ namespace ZoneTree.Analytics.Privacy; /// /// Explicit visitor ids: HMAC(secret, site | id). Stable over time. /// Anonymous visitors: HMAC(salt of the UTC day, site | ip | user agent). -/// Salts are random, kept only for the current and previous day, then destroyed, -/// so hashes cannot be recomputed or linked across days. +/// Salts are random, kept for the current and previous day, then destroyed, so hashes cannot +/// be recomputed or linked across days. The salt of an older day, created for events with +/// historical timestamps, is kept until a day after its last use, so a backfill hashes each +/// visitor of that day the same way across salt rotations and restarts. /// /// sealed class VisitorHasher @@ -25,6 +27,12 @@ sealed class VisitorHasher const string SaltPrefix = "salt:"; + /// The UTC ticks of the last use of an older day's salt, per day. + const string SaltUsePrefix = "salt-used:"; + + /// How often the last use of an older day's salt is persisted. + const long UseRefreshTicks = TimeSpan.TicksPerHour; + readonly VisitorIdentification mode; readonly byte[] secret; @@ -33,6 +41,9 @@ sealed class VisitorHasher readonly ConcurrentDictionary salts = new(); + /// The last persisted use of each older day's salt in use. + readonly ConcurrentDictionary uses = new(); + readonly object saltLock = new(); public VisitorHasher(VisitorIdentification mode, byte[] configuredSecret, MetaStore meta) @@ -81,54 +92,77 @@ public ulong Compute(string site, string visitorId, string ipAddress, string use return Hash(key, "anon", site, ipAddress ?? "", userAgent ?? ""); } + static string DayName(long utcDay) => new DateTime(utcDay).ToString("yyyy-MM-dd", CultureInfo.InvariantCulture); + + static long ParseDay(string name) => DateTime.ParseExact(name, "yyyy-MM-dd", CultureInfo.InvariantCulture).Ticks; + + /// The first day whose salt is kept by date: the previous UTC day. + static long KeptFrom(long utcNowTicks) => utcNowTicks - utcNowTicks % TimeSpan.TicksPerDay - TimeSpan.TicksPerDay; + byte[] GetSalt(long utcDay) { - if (salts.TryGetValue(utcDay, out var salt)) + var now = DateTime.UtcNow.Ticks; + var older = utcDay < KeptFrom(now); + if (salts.TryGetValue(utcDay, out var salt) && + (!older || (uses.TryGetValue(utcDay, out var used) && now - used < UseRefreshTicks))) return salt; lock (saltLock) { - if (salts.TryGetValue(utcDay, out salt)) - return salt; - var name = SaltPrefix + new DateTime(utcDay).ToString("yyyy-MM-dd", CultureInfo.InvariantCulture); - var stored = meta.Get(name); - if (stored != null) + var name = SaltPrefix + DayName(utcDay); + var created = false; + if (!salts.TryGetValue(utcDay, out salt)) { - salt = Convert.FromBase64String(stored); + var stored = meta.Get(name); + created = stored == null; + salt = created ? RandomNumberGenerator.GetBytes(32) : Convert.FromBase64String(stored); } - else + var recordUse = older && (!uses.TryGetValue(utcDay, out var lastUse) || now - lastUse >= UseRefreshTicks); + if (created || recordUse) { - salt = RandomNumberGenerator.GetBytes(32); - meta.Set(name, Convert.ToBase64String(salt)); + meta.Update(entries => + { + if (created) + entries[name] = Convert.ToBase64String(salt); + if (recordUse) + entries[SaltUsePrefix + DayName(utcDay)] = now.ToString(CultureInfo.InvariantCulture); + }); } + if (recordUse) + uses[utcDay] = now; salts[utcDay] = salt; return salt; } } /// - /// Destroys salts older than the previous UTC day. + /// Destroys the salts older than the previous UTC day that were not used during the last day. /// public void RotateSalts(DateTime utcNow) { - var keepFrom = utcNow.Date.AddDays(-1).Ticks; + var keepFrom = KeptFrom(utcNow.Ticks); + var idleFrom = utcNow.Ticks - TimeSpan.TicksPerDay; lock (saltLock) { - foreach (var day in salts.Keys) - { - if (day < keepFrom) - salts.TryRemove(day, out _); - } - var expired = meta.ListByPrefix(SaltPrefix) - .Select(x => x.Key) - .Where(name => DateTime.ParseExact(name[SaltPrefix.Length..], "yyyy-MM-dd", CultureInfo.InvariantCulture).Ticks < keepFrom) - .ToList(); - if (expired.Count > 0) + var lastUses = meta.ListByPrefix(SaltUsePrefix).ToDictionary( + x => ParseDay(x.Key[SaltUsePrefix.Length..]), x => long.Parse(x.Value, CultureInfo.InvariantCulture)); + var days = meta.ListByPrefix(SaltPrefix).Select(x => ParseDay(x.Key[SaltPrefix.Length..])) + .Concat(salts.Keys).Concat(lastUses.Keys).ToHashSet(); + var expired = days.Where(day => day < keepFrom && + Math.Max(lastUses.GetValueOrDefault(day, long.MinValue), uses.GetValueOrDefault(day, long.MinValue)) < idleFrom).ToList(); + if (expired.Count == 0) + return; + meta.Update(entries => { - meta.Update(entries => + foreach (var day in expired) { - foreach (var name in expired) - entries.Remove(name); - }); + entries.Remove(SaltPrefix + DayName(day)); + entries.Remove(SaltUsePrefix + DayName(day)); + } + }); + foreach (var day in expired) + { + salts.TryRemove(day, out _); + uses.TryRemove(day, out _); } } } diff --git a/src/ZoneTree.Analytics/Storage/DimensionIndexWriter.cs b/src/ZoneTree.Analytics/Storage/DimensionIndexWriter.cs index e8f032b..45bdf04 100644 --- a/src/ZoneTree.Analytics/Storage/DimensionIndexWriter.cs +++ b/src/ZoneTree.Analytics/Storage/DimensionIndexWriter.cs @@ -51,7 +51,7 @@ public void Delete(RawEvent record) IZoneTree, Memory> Tree(long ticks) { var store = owner.Store; - var name = store.Route(ticks); + var name = store.RouteWrite(ticks); if (lease?.Name != name) { lease?.Dispose(); diff --git a/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs b/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs index 510c8ef..40671ac 100644 --- a/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs +++ b/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs @@ -1,5 +1,6 @@ -using System.Text.Json; -using System.Text.Json.Serialization; +using System.Buffers.Binary; +using System.Security.Cryptography; +using System.Text; using ZoneTree.Analytics.Query; namespace ZoneTree.Analytics.Storage; @@ -30,33 +31,67 @@ sealed class JointBackfillReplacement public byte[] Value { get; set; } } -[JsonSourceGenerationOptions(IncludeFields = true)] -[JsonSerializable(typeof(JointBackfillState))] -partial class JointBackfillJson : JsonSerializerContext; - /// /// The cursor, active sessions and absolute replacements are committed together. A crash /// during the writes replays replacements instead of applying additive contributions twice. -/// Dedicated two-generation metadata files keep this temporary journal out of engine metadata. /// +/// +/// The journal is binary: a flush of replacements holds serialized rollup rows, whose dense +/// sketches are 16 KiB each. Two files are written alternately, like , +/// so a crash can only damage the file being written; loading picks the valid file with the +/// highest generation. File layout: +/// +/// magic(8) generation(8, LE) body length(8, LE) sha256 of body(32) body +/// +/// sealed class JointBackfillCheckpoint { - readonly MetaStore meta; + const int Version = 1; + + const int HeaderLength = 56; + + static readonly byte[] Magic = "ZTAJRNL1"u8.ToArray(); + + readonly string[] paths; + + long generation; public JointBackfillState State { get; } public bool Resuming { get; } public JointBackfillCheckpoint(string root, string configuration, long coverage) { - meta = new MetaStore(Path.Combine(root, "backfill", "joint-rollups")); - var json = meta.Get("state"); - var loaded = string.IsNullOrEmpty(json) ? null : JsonSerializer.Deserialize(json, JointBackfillJson.Default.JointBackfillState); + var directory = Path.Combine(root, "backfill", "joint-rollups"); + Directory.CreateDirectory(directory); + paths = [Path.Combine(directory, "journal-a.bin"), Path.Combine(directory, "journal-b.bin")]; + var loaded = Load(); Resuming = loaded?.Configuration == configuration; State = Resuming ? loaded : new() { Configuration = configuration, Coverage = coverage }; if (!Resuming) Save(); } - public void Save() => meta.Set("state", JsonSerializer.Serialize(State, JointBackfillJson.Default.JointBackfillState)); + public void Save() + { + var path = paths[(generation + 1) % 2]; + using (var stream = new FileStream(path, FileMode.Create, FileAccess.Write, FileShare.None, 1 << 16)) + { + // The header is written last: a torn write never carries a matching checksum. + stream.Write(new byte[HeaderLength]); + using var hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256); + using (var body = new BinaryWriter(new HashingStream(stream, hash), Encoding.UTF8, leaveOpen: true)) + Write(body, State); + var length = stream.Position - HeaderLength; + var header = new byte[HeaderLength]; + Magic.CopyTo(header, 0); + BinaryPrimitives.WriteInt64LittleEndian(header.AsSpan(8), generation + 1); + BinaryPrimitives.WriteInt64LittleEndian(header.AsSpan(16), length); + hash.GetHashAndReset().CopyTo(header, 24); + stream.Position = 0; + stream.Write(header); + stream.Flush(flushToDisk: true); + } + ++generation; + } public void Replay(RollupStore rollups) { @@ -74,5 +109,208 @@ public void Replay(RollupStore rollups) Save(); } - public void Complete() => meta.Delete("state"); + public void Complete() + { + foreach (var path in paths) + File.Delete(path); + generation = 0; + } + + JointBackfillState Load() + { + JointBackfillState newest = null; + foreach (var path in paths) + { + if (!File.Exists(path)) + continue; + using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, 1 << 16); + if (TryReadHeader(stream, out var fileGeneration) && fileGeneration > generation) + { + stream.Position = HeaderLength; + using var reader = new BinaryReader(stream, Encoding.UTF8, leaveOpen: true); + newest = Read(reader); + generation = fileGeneration; + } + } + return newest; + } + + /// Validates the header and the checksum of the whole body. + static bool TryReadHeader(FileStream stream, out long fileGeneration) + { + fileGeneration = 0; + var header = new byte[HeaderLength]; + if (stream.Length < HeaderLength || stream.ReadAtLeast(header, HeaderLength, throwOnEndOfStream: false) != HeaderLength || + !header.AsSpan(0, 8).SequenceEqual(Magic)) + return false; + fileGeneration = BinaryPrimitives.ReadInt64LittleEndian(header.AsSpan(8)); + var length = BinaryPrimitives.ReadInt64LittleEndian(header.AsSpan(16)); + if (fileGeneration < 1 || length != stream.Length - HeaderLength) + return false; + using var hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256); + var buffer = new byte[1 << 16]; + int read; + while ((read = stream.Read(buffer)) > 0) + hash.AppendData(buffer, 0, read); + return hash.GetHashAndReset().AsSpan().SequenceEqual(header.AsSpan(24, 32)); + } + + static void Write(BinaryWriter writer, JointBackfillState state) + { + writer.Write(Version); + writer.Write(state.Configuration); + writer.Write(state.Copying); + writer.Write(state.Coverage); + WriteString(writer, state.Site); + writer.Write(state.SiteCompleted); + WriteString(writer, state.Partition); + WriteBytes(writer, state.Key); + writer.Write(state.Events); + writer.Write(state.Sessions); + writer.Write(state.Updates); + writer.Write(state.Flushes); + writer.Write(state.Sites); + writer.Write(state.Active.Count); + foreach (var (id, session) in state.Active) + { + writer.Write(id); + writer.Write(session.Start); + writer.Write(session.Last); + writer.Write(session.PageViews); + writer.Write(session.Interactions); + writer.Write(session.Visitor); + writer.Write(session.Allowed); + writer.Write(session.Dimensions?.Count ?? -1); + if (session.Dimensions != null) + { + foreach (var (name, value) in session.Dimensions) + { + writer.Write(name); + WriteString(writer, value); + } + } + writer.Write(session.HasStartEvent); + writer.Write(session.FirstIsPageView); + WriteString(writer, session.FirstPage); + WriteString(writer, session.FirstHostname); + WriteString(writer, session.LastPage); + WriteString(writer, session.LastHostname); + writer.Write(session.LastPageTicks); + } + writer.Write(state.Redo.Count); + foreach (var row in state.Redo) + { + writer.Write((int)row.Granularity); + writer.Write(row.Partition); + WriteBytes(writer, row.Key); + WriteBytes(writer, row.Value); + } + } + + static JointBackfillState Read(BinaryReader reader) + { + if (reader.ReadInt32() != Version) + throw new AnalyticsConfigurationException("The joint rollup backfill journal has an unknown version."); + var state = new JointBackfillState + { + Configuration = reader.ReadString(), + Copying = reader.ReadBoolean(), + Coverage = reader.ReadInt64(), + Site = ReadString(reader), + SiteCompleted = reader.ReadBoolean(), + Partition = ReadString(reader), + Key = ReadBytes(reader), + Events = reader.ReadInt64(), + Sessions = reader.ReadInt64(), + Updates = reader.ReadInt64(), + Flushes = reader.ReadInt64(), + Sites = reader.ReadInt32() + }; + var active = reader.ReadInt32(); + for (var i = 0; i < active; ++i) + { + var id = reader.ReadUInt64(); + var session = new RawSession + { + Start = reader.ReadInt64(), + Last = reader.ReadInt64(), + PageViews = reader.ReadInt32(), + Interactions = reader.ReadInt32(), + Visitor = reader.ReadUInt64(), + Allowed = reader.ReadBoolean() + }; + var dimensions = reader.ReadInt32(); + if (dimensions >= 0) + { + session.Dimensions = new(dimensions, StringComparer.Ordinal); + for (var j = 0; j < dimensions; ++j) + session.Dimensions[reader.ReadString()] = ReadString(reader); + } + session.HasStartEvent = reader.ReadBoolean(); + session.FirstIsPageView = reader.ReadBoolean(); + session.FirstPage = ReadString(reader); + session.FirstHostname = ReadString(reader); + session.LastPage = ReadString(reader); + session.LastHostname = ReadString(reader); + session.LastPageTicks = reader.ReadInt64(); + state.Active[id] = session; + } + var redo = reader.ReadInt32(); + state.Redo = new(redo); + for (var i = 0; i < redo; ++i) + { + state.Redo.Add(new JointBackfillReplacement + { + Granularity = (Granularity)reader.ReadInt32(), + Partition = reader.ReadString(), + Key = ReadBytes(reader), + Value = ReadBytes(reader) + }); + } + return state; + } + + static void WriteString(BinaryWriter writer, string value) + { + writer.Write(value != null); + if (value != null) + writer.Write(value); + } + + static string ReadString(BinaryReader reader) => reader.ReadBoolean() ? reader.ReadString() : null; + + static void WriteBytes(BinaryWriter writer, byte[] value) + { + writer.Write(value?.Length ?? -1); + if (value != null) + writer.Write(value); + } + + static byte[] ReadBytes(BinaryReader reader) + { + var length = reader.ReadInt32(); + return length < 0 ? null : reader.ReadBytes(length); + } + + /// Forwards writes to a stream and hashes them. + sealed class HashingStream(Stream inner, IncrementalHash hash) : Stream + { + public override bool CanRead => false; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public override void Flush() => inner.Flush(); + 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(); + + public override void Write(byte[] buffer, int offset, int count) => Write(buffer.AsSpan(offset, count)); + + public override void Write(ReadOnlySpan buffer) + { + hash.AppendData(buffer); + inner.Write(buffer); + } + } } diff --git a/src/ZoneTree.Analytics/Storage/PartitionedStore.cs b/src/ZoneTree.Analytics/Storage/PartitionedStore.cs index 97b1863..b107edb 100644 --- a/src/ZoneTree.Analytics/Storage/PartitionedStore.cs +++ b/src/ZoneTree.Analytics/Storage/PartitionedStore.cs @@ -1,4 +1,5 @@ using System.Globalization; +using System.Text; using ZoneTree.Analytics.Internal; using ZoneTree.Logger; @@ -63,17 +64,28 @@ sealed class Partition public Exception CleanupFailure; } + /// + /// The latest time routed to a partition for a write, or when it + /// is unknown. Maps share the instances, so a write never updates a replaced copy. + /// + sealed class HighWater(long ticks = long.MinValue) + { + public long Ticks = ticks; + } + /// /// The partitions in time order. Replaced, never modified, so readers route without a lock. /// - sealed class PartitionMap(long[] starts, string[] names) + sealed class PartitionMap(long[] starts, string[] names, HighWater[] marks) { - public static PartitionMap Empty { get; } = new([], []); + public static PartitionMap Empty { get; } = new([], [], []); public long[] Starts { get; } = starts; public string[] Names { get; } = names; + public HighWater[] Marks { get; } = marks; + public int Count => Starts.Length; public long From(int index) => index == 0 ? long.MinValue : Starts[index]; @@ -89,17 +101,31 @@ public int IndexOf(long ticks) return index >= 0 ? index : Math.Max(0, ~index - 1); } - public PartitionMap Add(long start, string name) => new([.. Starts, start], [.. Names, name]); + public PartitionMap Add(long start, string name) => Insert(Count, start, name); + + public PartitionMap Insert(int index, long start, string name) => new( + [.. Starts[..index], start, .. Starts[index..]], + [.. Names[..index], name, .. Names[index..]], + [.. Marks[..index], new HighWater(), .. Marks[index..]]); public PartitionMap Remove(string name) { var index = Array.IndexOf(Names, name); if (index < 0) return this; - return new([.. Starts[..index], .. Starts[(index + 1)..]], [.. Names[..index], .. Names[(index + 1)..]]); + return new([.. Starts[..index], .. Starts[(index + 1)..]], [.. Names[..index], .. Names[(index + 1)..]], + [.. Marks[..index], .. Marks[(index + 1)..]]); } } + /// + /// The file in the root directory that holds the high-water marks of the partitions: + /// one "name ticks" line each. + /// + const string HighWaterFile = "high-water"; + + string savedHighWater; + readonly object sync = new(); readonly Dictionary partitions = new(StringComparer.Ordinal); @@ -162,7 +188,7 @@ public int OpenPartitionCount static string NameOf(long start) => new DateTime(start).ToString(NameFormat, CultureInfo.InvariantCulture); - static PartitionMap LoadMap(string rootDirectory) + PartitionMap LoadMap(string rootDirectory) { var starts = new SortedDictionary(); foreach (var directory in Directory.EnumerateDirectories(rootDirectory)) @@ -171,7 +197,47 @@ static PartitionMap LoadMap(string rootDirectory) if (DateTime.TryParseExact(name, NameFormat, CultureInfo.InvariantCulture, DateTimeStyles.None, out var start)) starts[start.Ticks] = name; } - return new([.. starts.Keys], [.. starts.Values]); + // A damaged or missing file leaves the marks unknown. + var marks = new Dictionary(StringComparer.Ordinal); + var path = Path.Combine(rootDirectory, HighWaterFile); + if (File.Exists(path)) + { + foreach (var line in File.ReadAllLines(path)) + { + var parts = line.Split(' '); + if (parts.Length == 2 && long.TryParse(parts[1], NumberStyles.None, CultureInfo.InvariantCulture, out var ticks)) + marks[parts[0]] = ticks; + } + } + var names = starts.Values.ToArray(); + var result = new PartitionMap([.. starts.Keys], names, + [.. names.Select(name => new HighWater(marks.GetValueOrDefault(name, long.MinValue)))]); + savedHighWater = HighWaterText(result); + return result; + } + + static string HighWaterText(PartitionMap map) => string.Join('\n', Enumerable.Range(0, map.Count) + .Where(i => Volatile.Read(ref map.Marks[i].Ticks) != long.MinValue) + .Select(i => map.Names[i] + " " + Volatile.Read(ref map.Marks[i].Ticks).ToString(CultureInfo.InvariantCulture))); + + /// + /// Persists the high-water marks that changed. Marks raised after the last save are lost in + /// a crash; the day boundary after a mark and the clock bound of new partitions cover them. + /// + void SaveHighWater() + { + var text = HighWaterText(map); + if (text == savedHighWater) + return; + var path = Path.Combine(RootDirectory, HighWaterFile); + var temporary = path + ".tmp"; + using (var stream = new FileStream(temporary, FileMode.Create, FileAccess.Write, FileShare.None)) + { + stream.Write(Encoding.UTF8.GetBytes(text)); + stream.Flush(flushToDisk: true); + } + File.Move(temporary, path, overwrite: true); + savedHighWater = text; } /// @@ -199,6 +265,28 @@ public string Route(long ticks) } } + /// + /// for a record about to be written: raises the partition's high-water + /// mark, which keeps later partition boundaries after the record. + /// + public string RouteWrite(long ticks) + { + var current = map; + if (current.Count == 0) + { + Route(ticks); + current = map; + } + var index = current.IndexOf(ticks); + var mark = current.Marks[index]; + for (var seen = Volatile.Read(ref mark.Ticks); ticks > seen; seen = Volatile.Read(ref mark.Ticks)) + { + if (Interlocked.CompareExchange(ref mark.Ticks, ticks, seen) == seen) + break; + } + return current.Names[index]; + } + /// /// The partitions that hold records in [fromTicks, toTicks), in time order, with the time /// range each one holds. @@ -310,11 +398,15 @@ internal void Release(string name) } /// - /// Starts a new partition when the latest one holds the configured number of records. The new - /// partition starts at the day boundary at or after , which - /// must be later than any record written so far: records written before the new partition - /// existed then still belong to the range of the previous one. Only an open latest partition - /// is counted; a closed one receives no records. Returns true when a partition was started. + /// Starts a new partition when the partition that receives new records holds the configured + /// number of records. That is the latest partition, unless a clock that ran ahead started a + /// later one: then it is the partition holding , provided its + /// high-water mark is known. The new partition starts at the day boundary at or after + /// , which must be later than any record the engine can still + /// receive, and after the day of the partition's high-water mark: records written before the + /// new partition existed, even by a clock that ran ahead, still belong to the range of the + /// previous one. Only an open partition is counted; a closed one receives no records. Also + /// persists the high-water marks. Returns true when a partition was started. /// public bool RollOver(long earliestStart) { @@ -322,19 +414,40 @@ public bool RollOver(long earliestStart) { if (isDisposed) return false; + try + { + SaveHighWater(); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) + { + // The marks in memory are exact; a later call saves them. + logger.LogError(exception); + } var current = map; - if (current.Count == 0 || - !partitions.TryGetValue(current.Names[^1], out var latest) || - latest.State != PartitionState.Open || - latest.DropRequested || - latest.Tree.Maintenance.TotalRecordCount < recordsPerPartition) + if (current.Count == 0) + return false; + var index = current.IndexOf(earliestStart); + var written = Volatile.Read(ref current.Marks[index].Ticks); + if (index < current.Count - 1 && written == long.MinValue) + { + index = current.Count - 1; + written = Volatile.Read(ref current.Marks[index].Ticks); + } + if (!partitions.TryGetValue(current.Names[index], out var target) || + target.State != PartitionState.Open || + target.DropRequested || + target.Tree.Maintenance.TotalRecordCount < recordsPerPartition) return false; var start = Math.Max( ReportingCalendar.FloorTicks(earliestStart + TimeSpan.TicksPerDay - 1, TimeSpan.TicksPerDay), - current.Starts[^1] + TimeSpan.TicksPerDay); + current.Starts[index] + TimeSpan.TicksPerDay); + if (written != long.MinValue) + start = Math.Max(start, ReportingCalendar.FloorTicks(written, TimeSpan.TicksPerDay) + TimeSpan.TicksPerDay); + if (index < current.Count - 1 && start >= current.Starts[index + 1]) + return false; var name = NameOf(start); Directory.CreateDirectory(PathOf(name)); - map = current.Add(start, name); + map = current.Insert(index + 1, start, name); return true; } } @@ -620,6 +733,14 @@ public void Dispose() if (CloseOrDrop(partition, partition.DropRequested, disposingStore: true) is { } failure) failures.Add(failure); } + try + { + SaveHighWater(); + } + catch (Exception exception) + { + failures.Add(exception); + } Disposal.ThrowFailures(failures); } } diff --git a/src/ZoneTree.Analytics/Storage/RawEventCodec.cs b/src/ZoneTree.Analytics/Storage/RawEventCodec.cs index 92b6811..b0c3ebe 100644 --- a/src/ZoneTree.Analytics/Storage/RawEventCodec.cs +++ b/src/ZoneTree.Analytics/Storage/RawEventCodec.cs @@ -186,6 +186,9 @@ public static Memory Encode(RawEvent rawEvent) foreach (var (name, value) in rawEvent.Dimensions) { Text.ValidateStoredName(name, nameof(rawEvent)); + // Ingestion never stores control characters; rollup keys reserve some of them. + if (value != null && value.Any(char.IsControl)) + throw new ArgumentException("Dimension values cannot contain control characters.", nameof(rawEvent)); Text.ValidateUnicode(value, nameof(rawEvent)); } foreach (var (name, _) in rawEvent.Values) @@ -307,6 +310,8 @@ internal static bool TryDecode( if (requireEnd && (string.IsNullOrWhiteSpace(key) || key.Any(char.IsControl))) throw new FormatException("A raw event dimension has an invalid name."); var value = reader.ReadString(strings); + if (requireEnd && value != null && value.Any(char.IsControl)) + throw new FormatException("A raw event dimension value contains a control character."); target.Dimensions.Add(new(key, value)); continue; } diff --git a/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs b/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs index ae830a3..913d671 100644 --- a/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs +++ b/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs @@ -212,7 +212,7 @@ int CopyCore(Granularity granularity, Memory originalKey, RollupValue valu if (!copies.TryGetValue((granularity, definitionId), out var destinations)) return 0; var key = StorageKeyCodec.DecodeRollup(originalKey.Span); var store = destinations[0].Store; - using var lease = store.Acquire(store.Route(key.BucketTicks)); + using var lease = store.Acquire(store.RouteWrite(key.BucketTicks)); if (lease == null) return 0; var written = 0; foreach (var index in destinations) diff --git a/src/ZoneTree.Analytics/Storage/RollupStore.cs b/src/ZoneTree.Analytics/Storage/RollupStore.cs index e33dfb7..53db637 100644 --- a/src/ZoneTree.Analytics/Storage/RollupStore.cs +++ b/src/ZoneTree.Analytics/Storage/RollupStore.cs @@ -100,7 +100,7 @@ public void Write(RollupWriteBatch batch) foreach (var (key, accumulator) in rows) { var store = GetStore(key.Granularity); - var partition = store.Route(key.Bucket); + var partition = store.RouteWrite(key.Bucket); if (!groups.TryGetValue((key.Granularity, partition), out var writes)) { writes = new(key.Granularity, partition); diff --git a/src/ZoneTree.Analytics/Storage/SessionState.cs b/src/ZoneTree.Analytics/Storage/SessionState.cs index 29435a8..8b5e5d6 100644 --- a/src/ZoneTree.Analytics/Storage/SessionState.cs +++ b/src/ZoneTree.Analytics/Storage/SessionState.cs @@ -175,9 +175,10 @@ public Memory Serialize(in SessionState entry) writer.WriteString(entry.LastPage); writer.WriteString(entry.LastHostname); writer.WriteVarInt64(entry.LastPageTicks - entry.StartTicks); - var dimensions = entry.Dimensions; + // Positions missing from a shorter array, such as the default one, were not collected. + var dimensions = entry.Dimensions ?? []; for (var i = 0; i < SessionSchema.StandardCount; ++i) - writer.WriteString(dimensions[i]); + writer.WriteString(i < dimensions.Length ? dimensions[i] : null); var customCount = 0; for (var i = SessionSchema.StandardCount; i < dimensions.Length; ++i) { From 97079942c03c0cab73d67ae6556617985043969a Mon Sep 17 00:00:00 2001 From: Ahmed Yasin Koculu Date: Thu, 15 Oct 2026 22:54:16 +0200 Subject: [PATCH 2/2] Fix the test identified the newest file by its write time which makes it flaky. --- .../StorageFormatRegressionTests.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs b/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs index d9bc3df..60d0d7e 100644 --- a/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs @@ -273,7 +273,10 @@ public void ATornJointBackfillJournalFallsBackToThePreviousCheckpoint() checkpoint.Save(); checkpoint.State.Events = 2; checkpoint.Save(); - var newest = Directory.GetFiles(Path.Combine(directory, "backfill", "joint-rollups")).OrderByDescending(File.GetLastWriteTimeUtc).First(); + // The newest file has the highest generation, stored after the 8-byte magic. Write times + // cannot tell the files apart: on Linux both saves can share a timestamp. + var newest = Directory.GetFiles(Path.Combine(directory, "backfill", "joint-rollups")) + .OrderByDescending(path => BitConverter.ToInt64(File.ReadAllBytes(path), 8)).First(); var bytes = File.ReadAllBytes(newest); File.WriteAllBytes(newest, bytes[..(bytes.Length - 1)]); Assert.That(new JointBackfillCheckpoint(directory, "configuration", 0).State.Events, Is.EqualTo(1));