upstream: a diagnostics episode follows health, and a peer fault carries its cause

src/storage/events.zig

@@ -322,6 +322,74 @@ pub const Store = struct {
         if (!self.active.put(code, digest, id)) self.untracked_active_count += 1;
     }
 
+    /// Opens the episode of `subject_key` when none is open, and restates the
+    /// severity and the detail of the one that is.
+    ///
+    /// The projection counterpart of `report`. A reconciliation asserts the
+    /// state an endpoint is in now; it is not a new observation, so it must not
+    /// raise `occurrences` or move `last_seen`. It does own the text: a retired
+    /// generation's late report can leave a stale cause on a card the live
+    /// generation still holds open, and this is what restores the true one.
+    /// Only a recorded failure calls `report`.
+    ///
+    /// The open check is a statement rather than a mirror lookup: the mirror is
+    /// a hint that can point at a row that is gone or already closed, and
+    /// `report` recovers from that through the touch it was making anyway.
+    /// There is no write here to learn it from, so this asks. That costs one
+    /// SELECT per reconciled subject, on a path that runs at boot and at a
+    /// generation retirement.
+    pub fn ensureOpen(
+        self: *Store,
+        io: std.Io,
+        now_s: i64,
+        code: Code,
+        subject_key: []const u8,
+        subject_label: []const u8,
+        severity: Severity,
+        detail: []const u8,
+    ) void {
+        var key_buf: [max_subject_key_len]u8 = undefined;
+        const key = canonicalKey(subject_key, &key_buf);
+        const label = truncate(subject_label, max_subject_label_len);
+        const text = truncate(detail, max_detail_len);
+        const digest = digestOf(key);
+
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+
+        self.count();
+        const existing = events_repo.selectActiveId(self.database, wire(code), key) catch |err|
+            return self.recordFailure(err);
+
+        // The read is not what clears the write latch: a `resolveExcept` that
+        // failed a moment ago is still the last word on whether this store can
+        // write, and only a write of our own can answer that.
+        if (existing) |id| {
+            self.count();
+            _ = events_repo.restateActive(self.database, id, severity.text(), text) catch |err|
+                return self.recordFailure(err);
+            return self.recordSuccess();
+        }
+
+        // Nothing is open, so a mirror entry claiming otherwise is stale and
+        // would make the insert below look like a collision.
+        if (self.active.find(code, digest)) |entry| self.active.remove(entry);
+
+        self.count();
+        const id = events_repo.insertActive(
+            self.database,
+            now_s,
+            wire(code),
+            key,
+            label,
+            severity.text(),
+            text,
+        ) catch |err| return self.recordFailure(err);
+        self.recordSuccess();
+
+        if (!self.active.put(code, digest, id)) self.untracked_active_count += 1;
+    }
+
     /// Records that `subject_key` is working again, closing its episode if one
     /// is open.
     ///
@@ -930,6 +998,38 @@ test "resolving a subject with nothing open executes no SQL at all" {
     try testing.expectEqual(after_resolve, store.statements);
 }
 
+test "ensureOpen opens a missing episode and restates an open one without counting it" {
+    var fx: Fixture = .{};
+    try fx.init(1000);
+    defer fx.deinit();
+    const store = &fx.store;
+
+    // The reconciliation path: it must be able to reassert a subject that is
+    // still failing without inventing an occurrence no exchange produced.
+    store.ensureOpen(fx.io, 1000, .upstream_exchange, "https://a.example", "a.example", .warning, "Timeout (cause Timeout)");
+    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"));
+    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT occurrences FROM operational_events"));
+    try testing.expectEqual(@as(i64, 1000), try fx.count("SELECT last_seen FROM operational_events"));
+
+    // The second call is a reconciliation of a card that is already open: the
+    // text and the severity are the reconciler's to state, and the counters
+    // belong to the exchanges that actually failed.
+    store.ensureOpen(fx.io, 1500, .upstream_exchange, "https://a.example", "a.example", .@"error", "SendFailed (cause BrokenPipe)");
+    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
+    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT occurrences FROM operational_events"));
+    try testing.expectEqual(@as(i64, 1000), try fx.count("SELECT first_seen FROM operational_events"));
+    try testing.expectEqual(@as(i64, 1000), try fx.count("SELECT last_seen FROM operational_events"));
+    try testing.expectEqualStrings("error", try fx.text("SELECT severity FROM operational_events"));
+    try testing.expectEqualStrings("SendFailed (cause BrokenPipe)", try fx.text("SELECT detail FROM operational_events"));
+
+    // And once the episode is resolved it opens a second one, like any other
+    // entry point.
+    store.resolve(fx.io, 1600, .upstream_exchange, "https://a.example");
+    store.ensureOpen(fx.io, 1700, .upstream_exchange, "https://a.example", "a.example", .warning, "Timeout (cause Timeout)");
+    try testing.expectEqual(@as(i64, 2), try fx.count("SELECT count(*) FROM operational_events"));
+    try testing.expectEqual(@as(i64, 1700), try fx.count("SELECT first_seen FROM operational_events WHERE id = 2"));
+}
+
 test "a one-shot event inserts already resolved and never enters the mirror" {
     var fx: Fixture = .{};
     try fx.init(1000);