fix event pipeline correctness issues · zzstoatzz.io/prefect-server@83d4e53

+1 -1

loq.toml

··· 10 10 11 11 [[rules]] 12 12 path = "src/broker/redis.zig" 13 - max_lines = 980 13 + max_lines = 1020

+26

src/api/events.zig

··· 220 220 SubscriberHandler.close(handle); 221 221 return; 222 222 } 223 + 224 + // Check backfill flag (defaults to true like Python) 225 + const wants_backfill = if (obj.get("backfill")) |v| switch (v) { 226 + .bool => |b| b, 227 + else => true, 228 + } else true; 229 + 223 230 // Parse filter from client message (uses page_allocator for filter lifetime) 224 231 const filter = event_broadcaster.EventFilter.parseFromJson(std.heap.page_allocator, message) catch { 225 232 log.err("events-out", "failed to parse filter", .{}); 226 233 SubscriberHandler.close(handle); 227 234 return; 228 235 }; 236 + 237 + // Backfill: send recent events before registering for live 238 + if (wants_backfill) { 239 + const db = @import("../db/sqlite.zig"); 240 + if (db.queryRecentEvents(alloc, 100)) |events| { 241 + var backfill_count: usize = 0; 242 + for (events) |event_json| { 243 + const wrapped = std.fmt.allocPrint(alloc, "{{\"type\":\"event\",\"event\":{s}}}", .{event_json}) catch continue; 244 + SubscriberHandler.write(handle, wrapped, true) catch continue; 245 + backfill_count += 1; 246 + } 247 + if (backfill_count > 0) { 248 + log.debug("events-out", "sent {d} backfill events", .{backfill_count}); 249 + } 250 + } else |err| { 251 + log.warn("events-out", "backfill query failed: {}", .{err}); 252 + } 253 + } 254 + 229 255 if (!event_broadcaster.addSubscriber(handle, filter)) { 230 256 log.err("events-out", "failed to add subscriber (at capacity)", .{}); 231 257 var f = filter;

+33 -31

src/broker/core.zig

··· 145 145 self.alloc.free(sub.group); 146 146 } 147 147 entry.value_ptr.*.subscriptions.deinit(self.alloc); 148 + // Free any remaining messages in queue 149 + for (entry.value_ptr.*.queue.items.items) |msg| { 150 + msg.deinit(); 151 + } 148 152 entry.value_ptr.*.queue.deinit(); 149 153 self.alloc.destroy(entry.value_ptr.*); 150 154 } ··· 177 181 pub fn publish(self: *Self, topic_name: []const u8, id: []const u8, data: []const u8) !void { 178 182 const topic = try self.getOrCreateTopic(topic_name); 179 183 180 - var msg: StoredMessage = undefined; 181 - msg.truncated = false; 182 - 183 - // copy data into fixed buffers 184 - msg.id_len = copyField(&msg.id, id, &msg.truncated); 185 - msg.data_len = copyField(&msg.data, data, &msg.truncated); 186 - msg.timestamp = std.time.milliTimestamp(); 187 - 188 - if (msg.truncated) { 189 - log.warn("broker", "message {s} truncated", .{msg.id[0..@min(msg.id_len, 36)]}); 190 - } 184 + const msg = try StoredMessage.init(self.alloc, id, data); 185 + errdefer msg.deinit(); 191 186 192 - topic.queue.push(msg) catch { 193 - log.warn("broker", "failed to enqueue message on topic {s}", .{topic_name}); 194 - return error.QueueError; 195 - }; 187 + try topic.queue.push(msg); 196 188 } 197 189 198 190 pub fn subscribe(self: *Self, topic_name: []const u8, group: []const u8, handler: MessageHandler) !ConsumerHandle { ··· 277 269 if (!should_run and topic.queue.len() == 0) break; 278 270 279 271 if (topic.queue.popTimeout(timeout_ns)) |msg| { 280 - // create Message from StoredMessage with proper topic 272 + defer msg.deinit(); 273 + 274 + // create Message from StoredMessage 281 275 const message = Message{ 282 - .id = msg.id[0..msg.id_len], 276 + .id = msg.id, 283 277 .topic = topic_name, 284 - .data = msg.data[0..msg.data_len], 278 + .data = msg.data, 285 279 .timestamp = msg.timestamp, 286 280 }; 287 281 ··· 312 306 // Storage types for message queue 313 307 // ============================================================================ 314 308 315 - /// Stored message data with fixed-size buffers (no allocation in hot path) 309 + /// Stored message with heap-allocated data (no size limits) 316 310 pub const StoredMessage = struct { 317 - id: [64]u8, 318 - id_len: usize, 319 - data: [8192]u8, 320 - data_len: usize, 311 + id: []const u8, 312 + data: []const u8, 321 313 timestamp: i64, 322 - truncated: bool, 323 - }; 314 + alloc: Allocator, 324 315 325 - fn copyField(dest: []u8, src: []const u8, truncated: *bool) usize { 326 - const copy_len = @min(src.len, dest.len); 327 - @memcpy(dest[0..copy_len], src[0..copy_len]); 328 - if (src.len > dest.len) truncated.* = true; 329 - return copy_len; 330 - } 316 + pub fn init(alloc: Allocator, id: []const u8, data: []const u8) !StoredMessage { 317 + const id_copy = try alloc.dupe(u8, id); 318 + errdefer alloc.free(id_copy); 319 + const data_copy = try alloc.dupe(u8, data); 320 + return .{ 321 + .id = id_copy, 322 + .data = data_copy, 323 + .timestamp = std.time.milliTimestamp(), 324 + .alloc = alloc, 325 + }; 326 + } 327 + 328 + pub fn deinit(self: StoredMessage) void { 329 + self.alloc.free(self.id); 330 + self.alloc.free(self.data); 331 + } 332 + }; 331 333 332 334 // ============================================================================ 333 335 // Growable Queue - unbounded, like Python's asyncio.Queue

+32 -1

src/broker/redis.zig

··· 328 328 } 329 329 } 330 330 331 + /// XGROUP DESTROY stream_key group_name - delete a consumer group 332 + pub fn xgroupDestroy(self: *Self, stream_key: []const u8, group_name: []const u8) !void { 333 + const args = [_][]const u8{ "XGROUP", "DESTROY", stream_key, group_name }; 334 + const response = try self.sendCommand(&args); 335 + 336 + // Returns integer (number of destroyed entries) or error 337 + if (response[0] == '-') { 338 + // NOGROUP error means group doesn't exist, which is fine for cleanup 339 + if (std.mem.indexOf(u8, response, "NOGROUP")) |_| { 340 + return; 341 + } 342 + log.err("redis", "XGROUP DESTROY error: {s}", .{response}); 343 + return error.RedisError; 344 + } 345 + } 346 + 331 347 /// XREADGROUP GROUP group consumer BLOCK ms COUNT n STREAMS stream > 332 348 pub fn xreadgroup( 333 349 self: *Self, ··· 652 668 handler: broker.MessageHandler, 653 669 thread: ?Thread, 654 670 running: *bool, 671 + ephemeral: bool, // if true, destroy group on unsubscribe 655 672 }; 656 673 657 674 pub fn init(alloc: Allocator, config: broker.RedisConfig) !*Self { ··· 770 787 .handler = handler, 771 788 .thread = null, 772 789 .running = running_flag, 790 + .ephemeral = false, // durable group - don't destroy on unsubscribe 773 791 }); 774 792 775 793 const idx = self.consumers.items.len - 1; ··· 794 812 795 813 /// Subscribe ephemerally - only receive new messages (no replay) 796 814 /// Uses "$" start_id so new groups only see messages published after creation 815 + /// Group is destroyed on unsubscribe to prevent accumulation 797 816 pub fn subscribeEphemeral(self: *Self, topic: []const u8, group: []const u8, handler: broker.MessageHandler) !broker.ConsumerHandle { 798 817 self.mutex.lock(); 799 818 defer self.mutex.unlock(); ··· 824 843 .handler = handler, 825 844 .thread = null, 826 845 .running = running_flag, 846 + .ephemeral = true, // ephemeral group - destroy on unsubscribe 827 847 }); 828 848 829 849 const idx = self.consumers.items.len - 1; ··· 851 871 defer self.mutex.unlock(); 852 872 853 873 if (handle.id < self.consumers.items.len) { 854 - self.consumers.items[handle.id].running.* = false; 874 + const consumer = &self.consumers.items[handle.id]; 875 + consumer.running.* = false; 876 + 877 + // Destroy ephemeral groups to prevent accumulation 878 + if (consumer.ephemeral) { 879 + var client = self.connectAndAuth() catch return; 880 + defer client.close(); 881 + client.xgroupDestroy(consumer.topic, consumer.group) catch |err| { 882 + log.warn("redis", "failed to destroy ephemeral group '{s}': {}", .{ consumer.group, err }); 883 + }; 884 + log.debug("redis", "destroyed ephemeral group '{s}'", .{consumer.group}); 885 + } 855 886 } 856 887 } 857 888

+76

src/db/events.zig

··· 68 68 } 69 69 return 0; 70 70 } 71 + 72 + /// Event for backfill queries 73 + pub const Event = struct { 74 + id: []const u8, 75 + occurred: []const u8, 76 + event: []const u8, 77 + resource_id: []const u8, 78 + resource: []const u8, 79 + payload: []const u8, 80 + related: []const u8, 81 + }; 82 + 83 + /// Query recent events for backfill (returns JSON-serialized events) 84 + /// Returns up to `limit` events ordered by occurred DESC 85 + pub fn queryRecent(alloc: std.mem.Allocator, limit: usize) ![][]const u8 { 86 + var limit_buf: [16]u8 = undefined; 87 + const limit_str = std.fmt.bufPrint(&limit_buf, "{d}", .{limit}) catch "100"; 88 + 89 + var results = std.ArrayListUnmanaged([]const u8){}; 90 + errdefer { 91 + for (results.items) |item| alloc.free(item); 92 + results.deinit(alloc); 93 + } 94 + 95 + const Col = struct { 96 + const id: usize = 0; 97 + const occurred: usize = 1; 98 + const event: usize = 2; 99 + const resource_id: usize = 3; 100 + const resource: usize = 4; 101 + const payload: usize = 5; 102 + const related: usize = 6; 103 + }; 104 + 105 + var sql_buf: [512]u8 = undefined; 106 + const sql = std.fmt.bufPrint(&sql_buf, "SELECT id, occurred, event, resource_id, resource, payload, related FROM events ORDER BY occurred DESC LIMIT {s}", .{limit_str}) catch return error.SqlTooLong; 107 + 108 + var rows = backend.db.query(sql, .{}) catch |err| { 109 + log.err("database", "query recent events error: {}", .{err}); 110 + return err; 111 + }; 112 + defer rows.deinit(); 113 + 114 + while (rows.next()) |row| { 115 + // Build event struct for proper JSON serialization 116 + const EventJson = struct { 117 + id: []const u8, 118 + occurred: []const u8, 119 + event: []const u8, 120 + resource_id: []const u8, 121 + }; 122 + 123 + const ev = EventJson{ 124 + .id = row.text(Col.id), 125 + .occurred = row.text(Col.occurred), 126 + .event = row.text(Col.event), 127 + .resource_id = row.text(Col.resource_id), 128 + }; 129 + 130 + // Serialize base fields using proper JSON escaping 131 + const base_json = std.json.Stringify.valueAlloc(alloc, ev, .{}) catch continue; 132 + defer alloc.free(base_json); 133 + 134 + // Splice in pre-serialized JSON fields (resource, payload, related are already JSON in DB) 135 + const json = std.fmt.allocPrint(alloc, "{s},\"resource\":{s},\"payload\":{s},\"related\":{s}}}", .{ 136 + base_json[0 .. base_json.len - 1], // remove closing } 137 + row.text(Col.resource), 138 + row.text(Col.payload), 139 + row.text(Col.related), 140 + }) catch continue; 141 + 142 + try results.append(alloc, json); 143 + } 144 + 145 + return results.toOwnedSlice(alloc); 146 + }

+1

src/db/sqlite.zig

··· 40 40 pub const insertEventDeduped = events.insertDeduped; 41 41 pub const trimEvents = events.trim; 42 42 pub const countEvents = events.count; 43 + pub const queryRecentEvents = events.queryRecent; 43 44 44 45 // legacy shared state (deprecated - use backend.db instead) 45 46 pub var conn: zqlite.Conn = undefined;

+49 -90

src/services/event_persister.zig

··· 28 28 var last_flush: i64 = 0; 29 29 30 30 const ParsedEvent = struct { 31 - id: [64]u8, 32 - id_len: usize, 33 - occurred: [32]u8, 34 - occurred_len: usize, 35 - event_name: [256]u8, 36 - event_name_len: usize, 37 - resource_id: [512]u8, 38 - resource_id_len: usize, 39 - resource: [8192]u8, // 8KB for resource JSON 40 - resource_len: usize, 41 - payload: [32768]u8, // 32KB for payload JSON 42 - payload_len: usize, 43 - related: [8192]u8, // 8KB for related JSON 44 - related_len: usize, 45 - follows: [64]u8, 46 - follows_len: usize, 47 - truncated: bool, 31 + id: []const u8, 32 + occurred: []const u8, 33 + event_name: []const u8, 34 + resource_id: []const u8, 35 + resource: []const u8, 36 + payload: []const u8, 37 + related: []const u8, 38 + follows: ?[]const u8, 39 + alloc: std.mem.Allocator, 48 40 49 - fn idSlice(self: *const ParsedEvent) []const u8 { 50 - return self.id[0..self.id_len]; 51 - } 52 - fn occurredSlice(self: *const ParsedEvent) []const u8 { 53 - return self.occurred[0..self.occurred_len]; 54 - } 55 - fn eventNameSlice(self: *const ParsedEvent) []const u8 { 56 - return self.event_name[0..self.event_name_len]; 57 - } 58 - fn resourceIdSlice(self: *const ParsedEvent) []const u8 { 59 - return self.resource_id[0..self.resource_id_len]; 60 - } 61 - fn resourceSlice(self: *const ParsedEvent) []const u8 { 62 - return self.resource[0..self.resource_len]; 63 - } 64 - fn payloadSlice(self: *const ParsedEvent) []const u8 { 65 - return self.payload[0..self.payload_len]; 66 - } 67 - fn relatedSlice(self: *const ParsedEvent) []const u8 { 68 - return self.related[0..self.related_len]; 69 - } 70 - fn followsSlice(self: *const ParsedEvent) ?[]const u8 { 71 - if (self.follows_len == 0) return null; 72 - return self.follows[0..self.follows_len]; 41 + pub fn deinit(self: ParsedEvent) void { 42 + self.alloc.free(self.id); 43 + self.alloc.free(self.occurred); 44 + self.alloc.free(self.event_name); 45 + if (self.resource_id.len > 0) self.alloc.free(self.resource_id); 46 + self.alloc.free(self.resource); 47 + self.alloc.free(self.payload); 48 + self.alloc.free(self.related); 49 + if (self.follows) |f| self.alloc.free(f); 73 50 } 74 51 }; 75 52 ··· 144 121 145 122 /// Message handler callback for broker subscription 146 123 fn handleMessage(msg: *const broker.Message) anyerror!void { 147 - // Parse event from JSON data 148 - var parsed: ParsedEvent = undefined; 149 - parsed.truncated = false; 124 + const alloc = std.heap.page_allocator; 150 125 151 - // Parse JSON 152 - var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 126 + // Parse JSON using arena (freed after we extract what we need) 127 + var arena = std.heap.ArenaAllocator.init(alloc); 153 128 defer arena.deinit(); 154 - const alloc = arena.allocator(); 129 + const arena_alloc = arena.allocator(); 155 130 156 - const json_parsed = std.json.parseFromSlice(std.json.Value, alloc, msg.data, .{}) catch { 131 + const json_parsed = std.json.parseFromSlice(std.json.Value, arena_alloc, msg.data, .{}) catch { 157 132 log.err("event_persister", "failed to parse event json", .{}); 158 133 return error.ParseError; 159 134 }; ··· 186 161 const occurred = getString(obj.get("occurred")) orelse return error.MissingOccurred; 187 162 const event_name = getString(obj.get("event")) orelse return error.MissingEvent; 188 163 const resource_id = getString(obj.get("resource_id")) orelse ""; 189 - const resource_json = stringifyJson(alloc, obj.get("resource"), "{}"); 190 - const payload_json = stringifyJson(alloc, obj.get("payload"), "{}"); 191 - const related_json = stringifyJson(alloc, obj.get("related"), "[]"); 164 + const resource_json = stringifyJson(arena_alloc, obj.get("resource"), "{}"); 165 + const payload_json = stringifyJson(arena_alloc, obj.get("payload"), "{}"); 166 + const related_json = stringifyJson(arena_alloc, obj.get("related"), "[]"); 192 167 const follows = getString(obj.get("follows")); 193 168 194 - // Copy to fixed buffers 195 - const copyField = struct { 196 - fn f(dest: []u8, src: []const u8, truncated: *bool) usize { 197 - const copy_len = @min(src.len, dest.len); 198 - @memcpy(dest[0..copy_len], src[0..copy_len]); 199 - if (src.len > dest.len) truncated.* = true; 200 - return copy_len; 201 - } 202 - }.f; 203 - 204 - parsed.id_len = copyField(&parsed.id, id, &parsed.truncated); 205 - parsed.occurred_len = copyField(&parsed.occurred, occurred, &parsed.truncated); 206 - parsed.event_name_len = copyField(&parsed.event_name, event_name, &parsed.truncated); 207 - parsed.resource_id_len = copyField(&parsed.resource_id, resource_id, &parsed.truncated); 208 - parsed.resource_len = copyField(&parsed.resource, resource_json, &parsed.truncated); 209 - parsed.payload_len = copyField(&parsed.payload, payload_json, &parsed.truncated); 210 - parsed.related_len = copyField(&parsed.related, related_json, &parsed.truncated); 211 - if (follows) |f| { 212 - parsed.follows_len = copyField(&parsed.follows, f, &parsed.truncated); 213 - } else { 214 - parsed.follows_len = 0; 215 - } 169 + // Heap-allocate copies (owned by ParsedEvent, freed on flush) 170 + const parsed = ParsedEvent{ 171 + .id = try alloc.dupe(u8, id), 172 + .occurred = try alloc.dupe(u8, occurred), 173 + .event_name = try alloc.dupe(u8, event_name), 174 + .resource_id = if (resource_id.len > 0) try alloc.dupe(u8, resource_id) else "", 175 + .resource = try alloc.dupe(u8, resource_json), 176 + .payload = try alloc.dupe(u8, payload_json), 177 + .related = try alloc.dupe(u8, related_json), 178 + .follows = if (follows) |f| try alloc.dupe(u8, f) else null, 179 + .alloc = alloc, 180 + }; 216 181 217 182 // Add to batch 218 183 batch_mutex.lock(); ··· 222 187 batch[batch_count] = parsed; 223 188 batch_count += 1; 224 189 225 - // Flush if batch is full 226 190 if (batch_count >= BATCH_SIZE) { 227 191 flushBatchLocked(); 228 192 } 229 193 } else { 194 + parsed.deinit(); 230 195 log.warn("event_persister", "batch full, event dropped", .{}); 231 196 } 232 197 } ··· 265 230 log.debug("event_persister", "flushing {d} events", .{batch_count}); 266 231 267 232 var success_count: usize = 0; 268 - var truncated_count: usize = 0; 269 233 var ts_buf: [32]u8 = undefined; 270 234 const now_ts = time_util.timestamp(&ts_buf); 271 235 272 236 for (batch[0..batch_count]) |event| { 273 - if (event.truncated) truncated_count += 1; 274 - 275 - const related_resource_ids = "[]"; 237 + defer event.deinit(); 276 238 277 239 db.insertEventDeduped( 278 - event.idSlice(), 279 - event.occurredSlice(), 280 - event.eventNameSlice(), 281 - event.resourceIdSlice(), 282 - event.resourceSlice(), 283 - related_resource_ids, 284 - event.relatedSlice(), 285 - event.payloadSlice(), 240 + event.id, 241 + event.occurred, 242 + event.event_name, 243 + event.resource_id, 244 + event.resource, 245 + "[]", // related_resource_ids 246 + event.related, 247 + event.payload, 286 248 now_ts, 287 249 now_ts, 288 - event.followsSlice(), 250 + event.follows, 289 251 ) catch |err| { 290 252 log.err("event_persister", "insert failed: {}", .{err}); 291 253 continue; ··· 295 257 296 258 if (success_count > 0) { 297 259 log.debug("event_persister", "persisted {d}/{d} events", .{ success_count, batch_count }); 298 - } 299 - if (truncated_count > 0) { 300 - log.warn("event_persister", "{d} events had truncated fields", .{truncated_count}); 301 260 } 302 261 303 262 batch_count = 0;