diff --git a/.changeset/zero-swift-inspector.md b/.changeset/zero-swift-inspector.md new file mode 100644 index 0000000000..fa34638dce --- /dev/null +++ b/.changeset/zero-swift-inspector.md @@ -0,0 +1,5 @@ +--- +'zero-swift': patch +--- + +Add JS's inspector and query metrics. `zero.inspector` lists the client's and its group's queries on the server, with their client and server timings and ZQL, reads the clients the local store knows and their local data, reports the client's query metrics merged with the server's, shows zero-cache's version, and analyzes a query, a server AST, or a named query on the server. Where JS asks for zero-cache's admin password in a dialog, a request throws `InspectorError.authenticationRequired`: call `zero.inspector.authenticate(password:)` first, and the password it accepts is presented again after a reconnect. The client records JS's query metrics (`query-materialization-client`, `query-materialization-end-to-end`, `query-update-client`) in a `TDigest` that gives the same quantiles as JS's, and `ZeroOptions.slowMaterializeThreshold` (5 seconds by default) logs a warning for a query that takes longer to materialize, server and network included. diff --git a/swift/Sources/Zero/AST/ASTToZQL.swift b/swift/Sources/Zero/AST/ASTToZQL.swift new file mode 100644 index 0000000000..3aed12ada9 --- /dev/null +++ b/swift/Sources/Zero/AST/ASTToZQL.swift @@ -0,0 +1,219 @@ +/// The query builder code of an AST after its table, such as +/// `.where('id', '=', 123)` (`astToZQL` in `packages/ast-to-zql/src/ast-to-zql.ts`, +/// without `redactLiterals`). The code is JS's, as the JS inspector shows it. +func astToZQL(_ ast: AST) -> String { + var code = "" + if let condition = ast.where { + var args = Set() + code += zqlCondition(condition, where: true, &args) + } + for related in ast.related ?? [] { + if related.hidden == true { + if let nested = related.subquery.related?.first { code += zqlRelated(nested) } + } else { + code += zqlRelated(related) + } + } + // distinct and groupBy go before orderBy, which then orders their rows. + if let groupBy = ast.groupBy { + let columns = groupBy.map(jsStringLiteral).joined(separator: ", ") + if let aggregate = ast.aggregate { + code += ".groupBy(\(columns)).\(aggregate.fn.rawValue)(\(zqlAggregateArg(aggregate)))" + } else { + code += ".distinct(\(columns))" + } + } else if let distinct = ast.aggregate?.distinct { + code += ".distinct(\(distinct.map(jsStringLiteral).joined(separator: ", ")))" + } + if let orderBy = ast.orderBy, !orderBy.isEmpty { + for part in orderBy { code += ".orderBy('\(part.field)', '\(part.direction.rawValue)')" } + } + if let limit = ast.limit { code += ".limit(\(JSNumber.format(limit)))" } + if let start = ast.start { + code += + ".start(\(JSONSerializer.stringify(.object(start.row)))" + + (start.exclusive ? "" : ", { inclusive: true }") + ")" + } + // A count of the distinct values ends the chain. + if ast.aggregate?.distinct != nil { code += ".count()" } + return code +} + +/// Whether a subquery has more than a table: JS tests these properties' +/// truthiness, so an empty ordering counts and a limit of 0 doesn't. +private func hasSubqueryProps(_ ast: AST) -> Bool { + ast.where != nil || !(ast.related ?? []).isEmpty || ast.orderBy != nil + || (ast.limit.map { $0 != 0 && !$0.isNaN } ?? false) +} + +/// `wrap`: `code`, a builder-callback expression, at the top level of a +/// `.where` (a callback of its own) or inside a callback (its names join the +/// enclosing callback's). +private func zqlWrap( + _ isWhere: Bool, _ args: inout Set, _ needed: [String], _ code: String +) -> String { + if isWhere { + let names = Set(needed).sorted(by: StringCompare.utf16Less).joined(separator: ", ") + return ".where(({\(names)}) => \(code))" + } + args.formUnion(needed) + return code +} + +private func zqlCondition( + _ condition: AST.Condition, where isWhere: Bool, _ args: inout Set +) -> String { + let prefix = isWhere ? ".where" : "cmp" + switch condition { + case .simple(let op, let left, let right): + let leftCode = zqlValue(left) + let rightCode = zqlValue(right) + let argsCode = + op == .eq ? "\(leftCode), \(rightCode)" : "\(leftCode), '\(op.rawValue)', \(rightCode)" + if case .json = left { + // A JSON path operand only exists through the expression builder's `json()`. + return zqlWrap(isWhere, &args, ["cmp", "json"], "cmp(\(argsCode))") + } + return "\(prefix)(\(argsCode))" + case .and(let conditions), .or(let conditions): + let type = condition.isAnd ? "and" : "or" + if conditions.count == 1 { return zqlCondition(conditions[0], where: isWhere, &args) } + if condition.isAnd { + let parts = conditions.map { zqlCondition($0, where: isWhere, &args) } + if isWhere { return parts.joined() } + args.insert("and") + return "and(" + parts.joined(separator: ", ") + ")" + } + // An OR gets a callback: of its own at the top level, the enclosing one + // inside a callback. + var own = Set() + let code: String + if isWhere { + code = conditions.map { zqlCondition($0, where: false, &own) }.joined(separator: ", ") + } else { + code = conditions.map { zqlCondition($0, where: false, &args) }.joined(separator: ", ") + own = args + } + return zqlWrap(isWhere, &args, Array(own) + ["cmp", type], "\(type)(\(code))") + case .correlatedSubquery(let related, let op, let flip, let scalar): + let relationship = zqlRelationshipName(related) + let next = nextExistsSubquery(related) + let props = hasSubqueryProps(next) + var options: [String] = [] + if let flip { options.append("flip: \(flip)") } + if let scalar { options.append("scalar: \(scalar)") } + let optionsCode = options.isEmpty ? "" : ", {\(options.joined(separator: ", "))}" + let subquery = props ? ", q => q\(astToZQL(next))" : "" + switch op { + case .exists: + if isWhere { return ".whereExists('\(relationship)'\(subquery)\(optionsCode))" } + args.insert("exists") + return "exists('\(relationship)'\(subquery)\(optionsCode))" + case .notExists: + return zqlWrap( + isWhere, &args, ["exists", "not"], + "not(exists('\(relationship)'\(subquery)\(optionsCode)))") + } + } +} + +extension AST.Condition { + fileprivate var isAnd: Bool { + if case .and = self { return true } + return false + } +} + +/// `getNextExistsSubquery`: an `exists` over a junction edge has both hops +/// as exists conditions with the same alias; the code names the far one. +private func nextExistsSubquery(_ related: AST.CorrelatedSubquery) -> AST { + if case .correlatedSubquery(let inner, _, _, _)? = related.subquery.where, + inner.subquery.alias?.contains(subqueryAliasPrefix + "zhidden_") == true + { + return nextExistsSubquery(inner) + } + return related.subquery +} + +private func zqlRelationshipName(_ related: AST.CorrelatedSubquery) -> String { + let alias = related.subquery.alias ?? "" + return alias.hasPrefix(subqueryAliasPrefix) + ? String(alias.dropFirst(subqueryAliasPrefix.count)) : alias +} + +private func zqlRelated(_ related: AST.CorrelatedSubquery) -> String { + guard let alias = related.subquery.alias else { return "" } + var code = ".related('\(alias)'" + if hasSubqueryProps(related.subquery) { code += ", q => q" + astToZQL(related.subquery) } + return code + ")" +} + +private func zqlValue(_ value: AST.ValuePosition) -> String { + switch value { + case .literal(let v): return zqlLiteral(v) + case .column(let name): return "'\(name)'" + case .json(let column, let path): + return "json('\(column)', \(zqlPath(path)))" + case .static(_, let field): + switch field { + case .single(let f): return "authParam('\(f)')" + case .path(let fields): + return "authParam([\(fields.map { "'\($0)'" }.joined(separator: ", "))])" + } + } +} + +private func zqlPath(_ path: [AST.JSONPathSegment]) -> String { + path.map { segment in + switch segment { + case .key(let k): jsStringLiteral(k) + case .index(let i): String(i) + } + }.joined(separator: ", ") +} + +/// `transformLiteral`: what `String()` prints, with strings quoted. +private func zqlLiteral(_ value: JSONValue) -> String { + switch value { + case .null: "null" + case .array: JSONSerializer.stringify(value) + case .string(let s): jsStringLiteral(s) + case .number(let n): JSNumber.format(n) + case .bool(let b): b ? "true" : "false" + case .object: "[object Object]" + } +} + +/// `transformAggregateArg`: nothing for a count, a column, or a callback +/// that picks a leaf of a `json()` column. +private func zqlAggregateArg(_ aggregate: AST.Aggregate) -> String { + guard let field = aggregate.field else { return "" } + let name = jsStringLiteral(field) + guard let path = aggregate.path else { return name } + return "({json}) => json(\(name), \(zqlPath(path)))" +} + +/// A single-quoted JS string literal of `s`, escaped as `JSON.stringify` +/// escapes it (`jsStringLiteral`). +private func jsStringLiteral(_ s: String) -> String { + let quoted = Array(JSONSerializer.stringify(.string(s)).unicodeScalars) + var out = String.UnicodeScalarView() + var i = 1 + let end = quoted.count - 1 + while i < end { + let c = quoted[i] + if c == "\\", i + 1 < end, quoted[i + 1] == "\"" { + out.append("\"") + i += 2 + continue + } + if c == "'" { + out.append("\\") + out.append("'") + } else { + out.append(c) + } + i += 1 + } + return "'" + String(out) + "'" +} diff --git a/swift/Sources/Zero/Client/Engine.swift b/swift/Sources/Zero/Client/Engine.swift index 3b21f53d5f..694956ac93 100644 --- a/swift/Sources/Zero/Client/Engine.swift +++ b/swift/Sources/Zero/Client/Engine.swift @@ -75,6 +75,10 @@ public struct ZeroOptions: Sendable { /// hydration), with its listener calls. It must call the function it is /// given, once, before it returns. public var batchViewUpdates: (@MainActor @Sendable (() -> Void) -> Void)? + /// How long, in milliseconds, a query may take to materialize, until the + /// server has sent its rows, before a warning is logged + /// (`slowMaterializeThreshold`). Faster ones are logged at debug level. + public var slowMaterializeThreshold: Double = 5_000 public init( cacheURL: String? = nil, userID: String? = nil, auth: String? = nil, @@ -145,6 +149,8 @@ final class Engine { private(set) var mutationTracker: MutationTracker! private(set) var queryManager: QueryManager! private var pokeHandler: PokeHandler! + /// The inspector's requests (`zero.inspector`). + let inspectRPCs = InspectRPCs() /// The current token: `options.auth`, then the last one `connect(auth:)` set. private(set) var auth: String? @@ -302,12 +308,17 @@ final class Engine { closedReason: { [weak self] in guard case .closed(let reason) = self?.connectionManager.state else { return nil } return reason - }) + }, + addMetric: { [weak self] metric, value, queryID, ast in + self?.queryManager.addMetric(metric, value, queryID: queryID, ast: ast) + }, + monotonicNow: { [scheduler] in scheduler.monotonicNow }) _ = connectionManager.subscribe { [weak self] state in self?.onConnectionStateChange(state) } onConnectionStateChange(connectionManager.state) queryManager = QueryManager( clientID: clientID, schema: schema, mutationTracker: mutationTracker, scheduler: scheduler, queryChangeThrottleMs: options.queryChangeThrottleMs, recentQueriesMaxSize: 0, + slowMaterializeThreshold: options.slowMaterializeThreshold, log: log, send: { [weak self] body in guard let self else { return } var body = body @@ -806,7 +817,8 @@ final class Engine { userID: userID, baseCookie: connectCookie.stringValue ?? connectCookie.numberValue.map(JSNumber.format), timestamp: scheduler.monotonicNow, lastMutationID: lastMutationIDReceived, wsid: wsid, - profileID: profileID, additionalParams: additionalConnectParams ?? []) + profileID: profileID, debugPerf: log.level == .debug, + additionalParams: additionalConnectParams ?? []) addConnectEvent(attempt, "reading deleted clients") var deleted = deletedClientsBody() addConnectEvent(attempt, "reading desired queries") @@ -915,6 +927,7 @@ final class Engine { connectedAt = nil socket?.close(code: closeCode) socket = nil + inspectRPCs.connectionLost(reason) lastMutationIDsSent.removeAll() pokeHandler.handleDisconnect() queryManager.clearGotQueriesAuthoritative() @@ -1047,8 +1060,8 @@ final class Engine { try mutationTracker.processPushResponse(body) case .transformError(let errors): queryManager.handleTransformErrors(errors) - case .inspect: - break + case .inspect(let body): + inspectRPCs.receive(body) } } catch { disconnect(.client(.internal, message: errorMessage(error))) @@ -1265,6 +1278,44 @@ final class Engine { database.release() database.removeWriteFailureListener(clientID) connectionManager.closed() + inspectRPCs.clientClosed() + } + + /// The socket of the connection, once the current attempt has made it + /// (the inspector's `getSocket`). + func inspectorSocket(_ done: @escaping (Result) -> Void) { + if repClosed { return done(.failure(.clientClosed)) } + // Swift only: JS waits for a connection that never comes. + if socketOrigin == nil { + return done(.failure(.connectionLost("No server socket origin provided"))) + } + let waiter = inspectRPCs.addSocketWaiter(done) + connectionReady.onSettle(loop) { [weak self] connected in + guard let self, let done = self.inspectRPCs.takeSocketWaiter(waiter) else { return } + if connected, let socket = self.socket { + done(.success(socket)) + } else { + let reason = + Self.connectionError(self.connectionManager.state) + ?? self.connectionManager.state.reason + done(.failure(.connectionLost(reason?.message ?? "Not connected"))) + } + } + } + + /// Sends an inspector request on `socket`, which must still be the + /// connection's, and calls `done` with its response. + func sendInspect( + _ socket: SocketHandle, _ body: InspectUpBody, + _ done: @escaping (Result) -> Void + ) { + if repClosed { return done(.failure(.clientClosed)) } + guard socket === self.socket else { + return done( + .failure(.connectionLost(connectionManager.state.reason?.message ?? "Disconnected"))) + } + inspectRPCs.expect(body.id, done) + socket.send(Upstream.inspect(body).text) } // MARK: - Storage and the client lifecycle diff --git a/swift/Sources/Zero/Client/Inspector.swift b/swift/Sources/Zero/Client/Inspector.swift new file mode 100644 index 0000000000..091f22bd4c --- /dev/null +++ b/swift/Sources/Zero/Client/Inspector.swift @@ -0,0 +1,718 @@ +/// Inspects the client's queries, local data, and metrics, and the +/// server's, for debugging (`zero.inspector`, the `Inspector` of +/// `packages/zero-client/src/client/inspector/inspector.ts`). +/// +/// The requests to the server go over the sync connection, once it is +/// connected. Unlike JS, which asks for zero-cache's admin password in an +/// HTML dialog when the server needs it, a request then throws +/// ``InspectorError/authenticationRequired``: call +/// ``authenticate(password:)`` first. The password it accepts is presented +/// again on later connections, which the server authenticates separately. +@MainActor +public final class Inspector { + let state: InspectorState + + init(engine: @escaping @MainActor () -> Engine?) { + state = InspectorState(engine: engine) + } + + /// This client. + public var client: InspectorClient { + InspectorClient(state, id: state.clientID, clientGroupID: state.clientGroupID) + } + + /// This client's group: the clients that share its local data. + public var clientGroup: InspectorClientGroup { client.clientGroup } + + /// The client's query metrics with the server's (`inspector.metrics()`). + public func metrics() async throws -> Metrics { + try await state.run { engine, done in + let client = engine.queryManager.metrics + engine.inspect(["op": "metrics"], expect: "metrics") { r in + done( + r.flatMap { v in + Result { + Metrics( + queryMaterializationClient: client.materializationClient, + queryMaterializationEndToEnd: client.materializationEndToEnd, + queryUpdateClient: client.updateClient, + queryMaterializationServer: try digest(v["query-materialization-server"]), + queryUpdateServer: try digest(v["query-update-server"])) + } + }) + } + } + } + + /// Every client the local store has a record of. + public func clients() async throws -> [InspectorClient] { + try await state.run { _, done in done(.success(self.state.clients(ofGroup: nil))) } + } + + /// The clients the local store has a record of that have queries on the + /// server, in the order the server answers. + public func clientsWithQueries() async throws -> [InspectorClient] { + try await state.run { engine, done in + self.state.withQueries(engine, self.state.clients(ofGroup: nil), done) + } + } + + /// zero-cache's version. + public func serverVersion() async throws -> String { + try await state.run { engine, done in + engine.inspect(["op": "version"], expect: "version") { r in + done(r.map { $0.stringValue ?? "" }) + } + } + } + + /// How the server runs a query: a named one by its name and arguments, + /// an unnamed one by its AST with the server's names. + public func analyzeQuery( + _ query: AnyQuery, options: AnalyzeQueryOptions? = nil + ) async throws -> AnalyzeQueryResult { + if let id = query.customQueryID { + return try await analyzeNamedQuery(name: id.name, args: id.args, options: options) + } + return try await analyze(serverAST: { $0.queryManager.serverAST(query.ast) }, options) + } + + public func analyzeQuery( + _ query: Query, options: AnalyzeQueryOptions? = nil + ) async throws -> AnalyzeQueryResult { + try await analyzeQuery(query.any, options: options) + } + + public func analyzeQuery( + _ request: QueryRequest, options: AnalyzeQueryOptions? = nil + ) async throws -> AnalyzeQueryResult { + try await analyzeNamedQuery(name: request.name, args: request.args, options: options) + } + + /// How the server runs a query given by its AST, which is sent as is: + /// it must already have the server's table and column names. + public func analyzeServerAST( + _ ast: AST, options: AnalyzeQueryOptions? = nil + ) async throws -> AnalyzeQueryResult { + try await analyze(serverAST: { _ in ast }, options) + } + + /// How the server runs a named query, which it resolves from the name + /// and arguments. + public func analyzeNamedQuery( + name: String, args: [JSONValue], options: AnalyzeQueryOptions? = nil + ) async throws -> AnalyzeQueryResult { + var request: JSONObject = [ + "op": "analyze-query", "name": .string(name), "args": .array(args), + ] + request["options"] = options?.json + return try await state.analyze(request) + } + + private func analyze( + serverAST: @escaping (Engine) -> AST, _ options: AnalyzeQueryOptions? + ) async throws -> AnalyzeQueryResult { + try await state.run { engine, done in + var request: JSONObject = ["op": "analyze-query", "ast": serverAST(engine).json] + request["options"] = options?.json + engine.inspect(request, expect: "analyze-query") { r in + done(r.map(AnalyzeQueryResult.init(json:))) + } + } + } + + /// Presents zero-cache's admin password. True when the server accepts it + /// (or runs in a development mode that doesn't check it); it is then + /// presented again on later connections. + public func authenticate(password: String) async throws -> Bool { + try await state.run { engine, done in engine.authenticate(password, done) } + } + + /// The planner's debug events (`AnalyzeQueryResult.joinPlans`) as text. + public func formatPlannerEvents(_ events: [JSONValue]) -> String { + formatPlanDebugEvents(events) + } +} + +/// A client the local store has a record of (`Client` in +/// `packages/zero-client/src/client/inspector/client.ts`). +@MainActor +public struct InspectorClient: Sendable { + let state: InspectorState + public let id: String + public let clientGroup: InspectorClientGroup + + init(_ state: InspectorState, id: String, clientGroupID: String) { + self.state = state + self.id = id + self.clientGroup = InspectorClientGroup(state, id: clientGroupID) + } + + /// The client's queries on the server. + public func queries() async throws -> [InspectorQuery] { + try await state.run { engine, done in + self.state.queries(engine, ["op": "queries", "clientID": .string(self.id)], done) + } + } + + /// The client's local data: every key and value of its replica, with its + /// pending mutations applied. + public func map() async throws -> [String: JSONValue] { + try await state.run { engine, done in + done( + Result { + var map: [String: JSONValue] = [:] + for e in try self.state.kv(engine, self.id).entries { map[e.key] = e.value } + return map + }) + } + } + + /// The client's local rows of `table`. + public func rows(_ table: String) async throws -> [Row] { + try await state.run { engine, done in + done( + Result { + try self.state.kv(engine, self.id).scan(prefix: Keys.entitiesPrefix + table + "/") + .compactMap { $0.value.objectValue.map(Row.init) } + }) + } + } +} + +/// A client group (`ClientGroup` in +/// `packages/zero-client/src/client/inspector/client-group.ts`). +@MainActor +public struct InspectorClientGroup: Sendable { + let state: InspectorState + public let id: String + + init(_ state: InspectorState, id: String) { + self.state = state + self.id = id + } + + /// The group's clients the local store has a record of. + public func clients() async throws -> [InspectorClient] { + try await state.run { _, done in done(.success(self.state.clients(ofGroup: self.id))) } + } + + /// The group's clients that have queries on the server, in the order the + /// server answers. + public func clientsWithQueries() async throws -> [InspectorClient] { + try await state.run { engine, done in + self.state.withQueries(engine, self.state.clients(ofGroup: self.id), done) + } + } + + /// The group's queries on the server. + public func queries() async throws -> [InspectorQuery] { + try await state.run { engine, done in self.state.queries(engine, ["op": "queries"], done) } + } +} + +/// A query on the server, with the client's view of it (`Query` in +/// `packages/zero-client/src/client/inspector/query.ts`). +public struct InspectorQuery: Sendable { + /// The name of a named query. + public let name: String? + /// The arguments of a named query. + public let args: [JSONValue]? + /// Whether the server has sent the query's rows. + public let got: Bool + public let ttl: TTL + /// When the query became inactive, in milliseconds since 1970. + public let inactivatedAt: Double? + public let rowCount: Double + public let deleted: Bool + /// The query's hash. + public let id: String + public let clientID: String + /// The client's metrics of the query, with the server's. + public let metrics: Metrics + /// The client's query as builder code, such as `issue.where('id', 'i1')`. + public let clientZQL: String? + /// The server's query as builder code. + public let serverZQL: String? + /// The client's materialization time, in milliseconds. + public let hydrateClient: Double? + /// The server's materialization time, in milliseconds. + public let hydrateServer: Double? + /// The time from materializing to the server's rows, in milliseconds. + public let hydrateTotal: Double? + public let updateClientP50: Double? + public let updateClientP95: Double? + public let updateServerP50: Double? + public let updateServerP95: Double? + + init(row: JSONValue, clientAST: AST?, clientMetrics: QueryClientMetrics?) throws { + clientID = row["clientID"]?.stringValue ?? "" + id = row["queryID"]?.stringValue ?? "" + // `new Date(ms)`'s time value: the whole milliseconds, NaN out of range. + inactivatedAt = row["inactivatedAt"]?.numberValue.map { ms in + ms.isFinite && abs(ms) <= 8.64e15 ? ms.rounded(.towardZero) + 0 : .nan + } + ttl = TTL.milliseconds(row["ttl"]?.numberValue ?? 0).normalized + name = row["name"]?.stringValue + args = row["args"]?.arrayValue + got = row["got"]?.boolValue ?? false + rowCount = row["rowCount"]?.numberValue ?? 0 + deleted = row["deleted"]?.boolValue ?? false + let serverAST = try row["ast"].flatMap { $0.isNull ? nil : try AST(json: $0) } + serverZQL = serverAST.map { $0.table + astToZQL($0) } + clientZQL = clientAST.map { $0.table + astToZQL($0) } + + // `mergeMetrics`. + let server: JSONValue? = row["metrics"].flatMap { $0.isNull ? JSONValue?.none : $0 } + func single(_ value: Double?) -> TDigest { + let d = TDigest() + if let value { d.add(value) } + return d + } + let merged = Metrics( + queryMaterializationClient: single(clientMetrics?.materializationClient), + queryMaterializationEndToEnd: single(clientMetrics?.materializationEndToEnd), + queryUpdateClient: clientMetrics?.updateClient ?? TDigest(), + queryMaterializationServer: single(server?["query-hydration-server-ms"]?.numberValue), + queryUpdateServer: try server.map { try digest($0["query-update-server"]) } ?? TDigest()) + metrics = merged + func percentile(_ d: TDigest, _ q: Double) -> Double? { + let n = d.quantile(q) + return n.isNaN ? nil : n + } + hydrateClient = clientMetrics?.materializationClient + hydrateServer = server?["query-hydration-server-ms"]?.numberValue + hydrateTotal = clientMetrics?.materializationEndToEnd + updateClientP50 = percentile(merged.queryUpdateClient, 0.5) + updateClientP95 = percentile(merged.queryUpdateClient, 0.95) + updateServerP50 = percentile(merged.queryUpdateServer, 0.5) + updateServerP95 = percentile(merged.queryUpdateServer, 0.95) + } +} + +/// Query metrics, by their JS names (`Metrics` in `lazy-inspector.ts`). +/// The client's digests are live: they go on recording. +public struct Metrics: Sendable { + /// `query-materialization-client`: how long views take to hydrate. + public let queryMaterializationClient: TDigest + /// `query-materialization-end-to-end`: how long queries take from + /// materializing until the server has sent their rows. + public let queryMaterializationEndToEnd: TDigest + /// `query-update-client`: how long the client takes to push a change + /// through a query. + public let queryUpdateClient: TDigest + /// `query-materialization-server`. + public let queryMaterializationServer: TDigest + /// `query-update-server`. + public let queryUpdateServer: TDigest + + /// The JS names of the metrics, in order. + public static let names = [ + "query-materialization-client", "query-materialization-end-to-end", "query-update-client", + "query-materialization-server", "query-update-server", + ] + + /// The metric of a JS name, such as `query-update-client`. + public subscript(name: String) -> TDigest? { + switch name { + case "query-materialization-client": queryMaterializationClient + case "query-materialization-end-to-end": queryMaterializationEndToEnd + case "query-update-client": queryUpdateClient + case "query-materialization-server": queryMaterializationServer + case "query-update-server": queryUpdateServer + default: nil + } + } +} + +/// What an analysis reports (`AnalyzeQueryOptions` in +/// `packages/zero-protocol/src/inspect-up.ts`); the server's defaults for +/// those left nil. +public struct AnalyzeQueryOptions: Sendable, Hashable { + /// The rows the query read from the replica (`vendedRows`). + public var vendedRows: Bool? + /// The rows the query would sync. + public var syncedRows: Bool? + /// The planner's debug events. + public var joinPlans: Bool? + + public init(vendedRows: Bool? = nil, syncedRows: Bool? = nil, joinPlans: Bool? = nil) { + self.vendedRows = vendedRows + self.syncedRows = syncedRows + self.joinPlans = joinPlans + } + + var json: JSONValue { + var o: JSONObject = [:] + o["vendedRows"] = vendedRows.map(JSONValue.bool) + o["syncedRows"] = syncedRows.map(JSONValue.bool) + o["joinPlans"] = joinPlans.map(JSONValue.bool) + return .object(o) + } +} + +/// How the server ran a query (`AnalyzeQueryResult` in +/// `packages/zero-protocol/src/analyze-query-result.ts`). Counts and rows +/// are by source (table), then by query. +public struct AnalyzeQueryResult: Sendable { + public let warnings: [String] + public let syncedRows: [String: [Row]]? + public let syncedRowCount: Double + /// When the analysis started, in milliseconds. + public let start: Double + public let end: Double + public let elapsed: Double? + /// The query after the server applied permissions, as builder code. + public let afterPermissions: String? + public let vendedRowCounts: [String: [String: Double]]? + public let vendedRows: [String: [String: [Row]]]? + /// SQLite's query plans, by query. + public let sqlitePlans: [String: [String]]? + public let readRows: [String: [String: [Row]]]? + public let readRowCountsByQuery: [String: [String: Double]]? + public let readRowCount: Double? + public let dbScansByQuery: [String: [String: Double]]? + /// The planner's debug events: see ``Inspector/formatPlannerEvents(_:)``. + public let joinPlans: [JSONValue]? + /// The result as the server sent it. + public let json: JSONValue + + init(json: JSONValue) { + self.json = json + func record(_ v: JSONValue?, _ f: (JSONValue) -> T?) -> [String: T]? { + guard let o = v?.objectValue else { return nil } + var out: [String: T] = [:] + for (k, e) in o { if let t = f(e) { out[k] = t } } + return out + } + func rows(_ v: JSONValue) -> [Row]? { + v.arrayValue?.compactMap { $0.objectValue.map(Row.init) } + } + func counts(_ v: JSONValue?) -> [String: [String: Double]]? { + record(v) { record($0, \.numberValue) } + } + func rowsBySource(_ v: JSONValue?) -> [String: [String: [Row]]]? { + record(v) { record($0, rows) } + } + warnings = json["warnings"]?.arrayValue?.compactMap(\.stringValue) ?? [] + syncedRows = record(json["syncedRows"], rows) + syncedRowCount = json["syncedRowCount"]?.numberValue ?? 0 + start = json["start"]?.numberValue ?? 0 + end = json["end"]?.numberValue ?? 0 + elapsed = json["elapsed"]?.numberValue + afterPermissions = json["afterPermissions"]?.stringValue + vendedRowCounts = counts(json["vendedRowCounts"]) + vendedRows = rowsBySource(json["vendedRows"]) + sqlitePlans = record(json["sqlitePlans"]) { $0.arrayValue?.compactMap(\.stringValue) } + readRows = rowsBySource(json["readRows"]) + readRowCountsByQuery = counts(json["readRowCountsByQuery"]) + readRowCount = json["readRowCount"]?.numberValue + dbScansByQuery = counts(json["dbScansByQuery"]) + joinPlans = json["joinPlans"]?.arrayValue + } +} + +/// Why an inspector request failed. +public enum InspectorError: Error, Sendable, Hashable, CustomStringConvertible { + /// The server needs zero-cache's admin password: call + /// ``Inspector/authenticate(password:)``. JS asks for it in a dialog. + case authenticationRequired + /// The server's error response. + case server(String) + /// A response of another kind than the request's. + case unexpectedResponse(op: String) + /// The connection ended, or couldn't be made, before the response came. + case connectionLost(String) + /// Zero was closed. + case clientClosed + /// The local store has no record of the client. + case clientNotFound(String) + + public var description: String { + switch self { + case .authenticationRequired: + "The inspector needs zero-cache's admin password: call " + + "zero.inspector.authenticate(password:) first." + case .server(let message): message + case .unexpectedResponse(let op): "Unexpected inspector response \"\(op)\"" + case .connectionLost(let reason): + "The connection to zero-cache ended before the inspector's request was answered: \(reason)" + case .clientClosed: "The inspector's request failed: Zero instance closed by user" + case .clientNotFound(let id): "Client not found: \(id)" + } + } +} + +private func digest(_ json: JSONValue?) throws -> TDigest { + try TDigest(json: json?.arrayValue?.compactMap(\.numberValue) ?? []) +} + +/// What the inspector and its clients share: the current client, and the +/// queries it dropped (`#deletedQueries`). +@MainActor +final class InspectorState { + private let engine: () -> Engine? + private weak var attached: Engine? + private(set) var clientID = "" + private(set) var clientGroupID = "" + private var deletedQueries: [String: (ast: AST, metrics: QueryClientMetrics)] = [:] + + init(engine: @escaping () -> Engine?) { + self.engine = engine + if let engine = engine() { attach(engine) } + } + + /// Follows `engine`, a new client of the `Zero`. + func attach(_ engine: Engine) { + attached = engine + clientID = engine.clientID + clientGroupID = engine.clientGroupID + engine.queryManager.onQueryEvicted = { [weak self] hash, ast, metrics in + self?.deletedQueries[hash] = (ast, metrics) + } + } + + /// Runs `body` on the engine's loop and waits for what it reports. + func run( + _ body: @escaping (Engine, @escaping (Result) -> Void) -> Void + ) async throws -> T { + try Task.checkCancellation() + guard let engine = engine() else { throw InspectorError.clientClosed } + if engine !== attached { attach(engine) } + let result = Deferred() + engine.loop.call { body(engine) { result.settle($0) } } + return try await withTaskCancellationHandler { + try await result.value + } onCancel: { + Task { @MainActor in result.reject(CancellationError()) } + } + } + + func analyze(_ request: JSONObject) async throws -> AnalyzeQueryResult { + try await run { engine, done in + engine.inspect(request, expect: "analyze-query") { r in + done(r.map(AnalyzeQueryResult.init(json:))) + } + } + } + + func clients(ofGroup id: String?) -> [InspectorClient] { + guard let engine = engine() else { return [] } + return engine.database.clients.filter { id == nil || $0.clientGroupID == id }.map { + InspectorClient(self, id: $0.id, clientGroupID: $0.clientGroupID) + } + } + + /// The `queries` request, as `InspectorQuery`s with the most expensive on + /// the server first. + func queries( + _ engine: Engine, _ request: JSONObject, + _ done: @escaping (Result<[InspectorQuery], any Error>) -> Void + ) { + engine.inspect(request, expect: "queries") { [self] r in + done( + r.flatMap { v in + Result { + let queries = try (v.arrayValue ?? []).map { row in + let id = row["queryID"]?.stringValue ?? "" + return try InspectorQuery( + row: row, + clientAST: engine.queryManager.clientAST(id) ?? deletedQueries[id]?.ast, + clientMetrics: engine.queryManager.queryMetrics(id) ?? deletedQueries[id]?.metrics) + } + // A stable sort, as JS's. + return queries.enumerated().sorted { a, b in + let x = a.element.hydrateServer ?? 0 + let y = b.element.hydrateServer ?? 0 + return x != y ? x > y : a.offset < b.offset + }.map(\.element) + } + }) + } + } + + /// The clients of `clients` that have queries, asked about all at once + /// as JS does, in the order the answers come. + func withQueries( + _ engine: Engine, _ clients: [InspectorClient], + _ done: @escaping (Result<[InspectorClient], any Error>) -> Void + ) { + if clients.isEmpty { return done(.success([])) } + var remaining = clients.count + var found: [InspectorClient] = [] + var failed = false + for client in clients { + queries(engine, ["op": "queries", "clientID": .string(client.id)]) { r in + guard !failed else { return } + switch r { + case .failure(let e): + failed = true + done(.failure(e)) + case .success(let queries): + if !queries.isEmpty { found.append(client) } + remaining -= 1 + if remaining == 0 { done(.success(found)) } + } + } + } + } + + /// The key-value map of client `id`: its group's, as this client has it + /// for its own group, and as last stored for another. + func kv(_ engine: Engine, _ id: String) throws -> KVMap { + guard let record = engine.database.clients.first(where: { $0.id == id }) else { + throw InspectorError.clientNotFound(id) + } + if record.clientGroupID == engine.clientGroupID { return engine.store.main } + return try engine.database.storedEntries(ofGroup: record.clientGroupID) + } +} + +/// The inspector's requests in flight on the connection, and the password +/// the server last accepted (`adminPassword`). +@MainActor +final class InspectRPCs { + /// The admin password the server last accepted, presented again when a + /// later connection is unauthenticated. + var adminPassword: String? + /// Makes each request's ID (`nanoid`). Tests can make them predictable. + var makeID: () -> String = randomID + private var pending: [(id: String, done: (Result) -> Void)] = + [] + private var socketWaiters: [(id: Int, done: (Result) -> Void)] = + [] + private var nextWaiterID = 0 + + func expect(_ id: String, _ done: @escaping (Result) -> Void) { + pending.append((id, done)) + } + + /// An `inspect` message from the server: answers the request of its ID. + func receive(_ body: InspectDownBody) { + guard let i = pending.firstIndex(where: { $0.id == body.id }) else { return } + pending.remove(at: i).done(.success(body)) + } + + func addSocketWaiter( + _ done: @escaping (Result) -> Void + ) -> Int { + nextWaiterID += 1 + socketWaiters.append((nextWaiterID, done)) + return nextWaiterID + } + + func takeSocketWaiter(_ id: Int) -> ((Result) -> Void)? { + guard let i = socketWaiters.firstIndex(where: { $0.id == id }) else { return nil } + return socketWaiters.remove(at: i).done + } + + /// The connection ended: the requests on it fail. + func connectionLost(_ reason: ConnectionError) { + fail(reason.clientKind == .clientClosed ? .clientClosed : .connectionLost(reason.message)) + } + + /// The client closed: every request fails, also those waiting to connect. + func clientClosed() { + fail(.clientClosed) + let waiters = socketWaiters + socketWaiters = [] + for w in waiters { w.done(.failure(.clientClosed)) } + } + + private func fail(_ error: InspectorError) { + let failed = pending + pending = [] + for p in failed { p.done(.failure(error)) } + } +} + +extension Engine { + /// `rpcNoAuthTry`: sends one request on `socket` and reports the value + /// of a response of kind `op`. An unauthenticated answer is + /// `InspectorError.authenticationRequired`. + func inspectOnce( + _ socket: SocketHandle, _ request: JSONObject, expect op: String, + _ done: @escaping (Result) -> Void + ) { + var request = request + request["id"] = .string(inspectRPCs.makeID()) + let body: InspectUpBody + do { + body = try InspectUpBody(json: .object(request)) + } catch { + return done(.failure(error)) + } + sendInspect(socket, body) { r in + switch r { + case .failure(let e): + done(.failure(e)) + case .success(let down): + if down.op == "error" { + done(.failure(InspectorError.server(down.value.stringValue ?? ""))) + } else if down.op == op { + done(.success(down.value)) + } else if down.op == "authenticated", down.value == false { + done(.failure(InspectorError.authenticationRequired)) + } else { + done(.failure(InspectorError.unexpectedResponse(op: down.op))) + } + } + } + } + + /// `rpc`: a request, which, when the connection is unauthenticated, is + /// sent again after the password the server last accepted, if it still + /// does. + func inspect( + _ request: JSONObject, expect op: String, + _ done: @escaping (Result) -> Void + ) { + inspectorSocket { r in + let socket: SocketHandle + switch r { + case .failure(let e): return done(.failure(e)) + case .success(let s): socket = s + } + self.inspectOnce(socket, request, expect: op) { r in + guard case .failure(let e as InspectorError) = r, e == .authenticationRequired, + let remembered = self.inspectRPCs.adminPassword + else { return done(r) } + self.authenticateOn(socket, remembered) { ok in + switch ok { + case .failure(let e): + done(.failure(e)) + case .success(true): + self.inspectOnce(socket, request, expect: op, done) + case .success(false): + self.inspectRPCs.adminPassword = nil + done(r) + } + } + } + } + } + + private func authenticateOn( + _ socket: SocketHandle, _ password: String, + _ done: @escaping (Result) -> Void + ) { + inspectOnce( + socket, ["op": "authenticate", "value": .string(password)], expect: "authenticated" + ) { r in done(r.map { $0.boolValue ?? false }) } + } + + /// `authenticate`: remembers the password once the server accepts it. + func authenticate(_ password: String, _ done: @escaping (Result) -> Void) { + inspectorSocket { r in + switch r { + case .failure(let e): + done(.failure(e)) + case .success(let socket): + self.authenticateOn(socket, password) { r in + if case .success(true) = r { self.inspectRPCs.adminPassword = password } + done(r) + } + } + } + } +} diff --git a/swift/Sources/Zero/Client/PlannerDebug.swift b/swift/Sources/Zero/Client/PlannerDebug.swift new file mode 100644 index 0000000000..0c454295d6 --- /dev/null +++ b/swift/Sources/Zero/Client/PlannerDebug.swift @@ -0,0 +1,223 @@ +#if canImport(Darwin) + import Darwin +#elseif canImport(Glibc) + import Glibc +#elseif canImport(Musl) + import Musl +#endif + +/// The planner's debug events (`AnalyzeQueryResult.joinPlans`) as text, one +/// summary per planning attempt and the plan chosen (`formatPlannerEvents` +/// in `packages/zql/src/planner/planner-debug.ts`). +func formatPlanDebugEvents(_ events: [JSONValue]) -> String { + var lines: [String] = [] + var attempts: [Double] = [] + var byAttempt: [Double: [JSONValue]] = [:] + var best: JSONValue? + for event in events { + if event.objectValue?.index(forKey: "attemptNumber") != nil { + if let attempt = event["attemptNumber"]?.numberValue { + if byAttempt[attempt] == nil { attempts.append(attempt) } + byAttempt[attempt, default: []].append(event) + } + } else if event["type"] == "best-plan-selected" { + best = event + } + } + for attempt in attempts { + lines += attemptSummary(attempt, byAttempt[attempt]!) + lines.append("") + } + if let best { + let rule = String(repeating: "─", count: 60) + lines.append(rule) + lines.append( + "✓ Best plan: Attempt \(JSNumber.format(number(best, "bestAttemptNumber") + 1)) " + + "(cost=\(toFixed(number(best, "totalCost"), 2)))") + let joins = best["joinStates"]?.arrayValue ?? [] + if !joins.isEmpty { + lines.append(" Join types:") + for j in joins { + lines.append(" \(j["join"]?.stringValue ?? ""): \(j["type"]?.stringValue ?? "")") + } + } + lines.append(rule) + } + return lines.joined(separator: "\n") +} + +private func number(_ v: JSONValue?, _ key: String) -> Double { + v?[key]?.numberValue ?? .nan +} + +/// `formatAttemptSummary`. +private func attemptSummary(_ attempt: Double, _ events: [JSONValue]) -> [String] { + var lines: [String] = [] + let start = events.first { $0["type"] == "attempt-start" } + let total = start?["totalAttempts"]?.numberValue + var bits = 1.0 + if let total { + let b = log2(total).rounded(.up) + bits = b == 0 || b.isNaN ? 1 : b + } + var pattern = binary(attempt) + if bits.isFinite, Double(pattern.count) < bits { + pattern = String(repeating: "0", count: Int(bits) - pattern.count) + pattern + } + lines.append( + "[Attempt \(JSNumber.format(attempt + 1))/\(total.map(JSNumber.format) ?? "?")] " + + "Pattern \(JSNumber.format(attempt)) (\(pattern))") + + func nodeCosts(_ nodeType: String) -> [JSONValue] { + events.filter { $0["type"] == "node-cost" && $0["nodeType"]?.stringValue == nodeType } + } + func costLines(_ cost: JSONValue) -> [String] { + let estimate = cost["costEstimate"] + let limit = estimate?["limit"]?.numberValue.map(JSNumber.format) ?? "none" + return [ + " cost=\(toFixed(number(estimate, "cost"), 2)), " + + "startup=\(toFixed(number(estimate, "startupCost"), 2)), " + + "scan=\(toFixed(number(estimate, "scanEst"), 2))", + " rows=\(toFixed(number(estimate, "returnedRows"), 2)), " + + "selectivity=\(toFixed(number(estimate, "selectivity"), 8)), limit=\(limit)", + " downstreamChildSelectivity=\(toFixed(number(cost, "downstreamChildSelectivity"), 8))", + ] + } + + let connections = nodeCosts("connection") + let constraints = events.filter { + $0["type"] == "node-constraint" && $0["nodeType"]?.stringValue == "connection" + } + if !connections.isEmpty { + lines.append(" Connections:") + for cost in connections { + let pattern = branchPattern(cost) + let constraint = constraints.first { + $0["node"] == cost["node"] && branchPattern($0) == pattern + }?["constraint"] + lines.append(" \(cost["node"]?.stringValue ?? ""):") + lines += costLines(cost) + lines.append(" constraints=\(formatConstraint(constraint))") + lines.append(" filters=\(cost["filters"].map(formatFilter) ?? "none")") + lines.append(" ordering=\(formatOrdering(cost["ordering"]))") + } + } + let joins = nodeCosts("join") + if !joins.isEmpty { + lines.append(" Joins:") + for cost in joins { + let joinType = cost["joinType"]?.stringValue ?? "" + lines.append( + " \(cost["node"]?.stringValue ?? "")\(joinType.isEmpty ? "" : " (\(joinType))"):") + lines += costLines(cost) + } + } + if let complete = events.first(where: { $0["type"] == "plan-complete" }) { + lines.append(" ✓ Plan complete: total cost = \(toFixed(number(complete, "totalCost"), 2))") + } else if let failed = events.first(where: { $0["type"] == "plan-failed" }) { + lines.append(" ✗ Plan failed: \(failed["reason"]?.stringValue ?? "")") + } + return lines +} + +/// `branchPattern.join(',')`. +private func branchPattern(_ event: JSONValue) -> String { + (event["branchPattern"]?.arrayValue ?? []).map { $0.numberValue.map(JSNumber.format) ?? "" } + .joined(separator: ",") +} + +/// `attemptNum.toString(2)`, for the integers attempts are numbered with. +private func binary(_ n: Double) -> String { + guard n == n.rounded(), abs(n) < 9_007_199_254_740_992 else { return JSNumber.format(n) } + let digits = String(Int64(abs(n)), radix: 2) + return n < 0 ? "-" + digits : digits +} + +private func formatConstraint(_ constraint: JSONValue?) -> String { + guard let keys = constraint?.objectValue?.map(\.key), !keys.isEmpty else { return "{}" } + return "{" + keys.joined(separator: ", ") + "}" +} + +private func formatValuePosition(_ value: JSONValue) -> String { + switch value["type"]?.stringValue { + case "column": + return value["name"]?.stringValue ?? "" + case "json": + // `formatJsonPathReference`. + let path = (value["path"]?.arrayValue ?? []).map { s in + s.numberValue.map { "[\(JSNumber.format($0))]" } ?? ".\(s.stringValue ?? "")" + } + return (value["value"]?["name"]?.stringValue ?? "") + path.joined() + case "literal": + let literal = value["value"] ?? .null + if let s = literal.stringValue { return "'\(s)'" } + return JSONSerializer.stringify(literal) + case "static": + let field = value["field"] + let name = + field?.arrayValue.map { $0.map { $0.stringValue ?? "" }.joined(separator: ".") } + ?? field?.stringValue ?? "" + return "@\(value["anchor"]?.stringValue ?? "").\(name)" + default: + return JSONSerializer.stringify(value) + } +} + +private func formatFilter(_ filter: JSONValue) -> String { + switch filter["type"]?.stringValue { + case "simple": + return "\(formatValuePosition(filter["left"] ?? .null)) \(filter["op"]?.stringValue ?? "") " + + formatValuePosition(filter["right"] ?? .null) + case "and", "or": + let joiner = filter["type"] == "and" ? " AND " : " OR " + return "(" + + (filter["conditions"]?.arrayValue ?? []).map(formatFilter).joined( + separator: joiner) + ")" + case "correlatedSubquery": + return "EXISTS(\(filter["related"]?["subquery"]?["table"]?.stringValue ?? ""))" + default: + return JSONSerializer.stringify(filter) + } +} + +private func formatOrdering(_ ordering: JSONValue?) -> String { + guard let parts = ordering?.arrayValue, !parts.isEmpty else { return "none" } + return parts.map { "\($0[0]?.stringValue ?? "") \($0[1]?.stringValue ?? "")" }.joined( + separator: ", ") +} + +/// `Number.prototype.toFixed(digits)`: the value rounded to `digits` +/// decimals, a tie away from zero, from its exact binary value. +func toFixed(_ x: Double, _ digits: Int) -> String { + if x.isNaN { return "NaN" } + if abs(x) >= 1e21 { return JSNumber.format(x) } + let negative = x < 0 + let v = abs(x) + // v = significand * 2^shift, exactly. + var significand = UInt128(v.significandBitPattern) + if v.isNormal { significand |= 1 << 52 } + let shift = v.isNormal ? Int(v.exponent) - 52 : -1074 + var scale: UInt128 = 1 + for _ in 0..= 0 { + n = (significand << UInt128(shift)) * scale + } else if -shift > 100 { + // Below 2^53 * 10^8 / 2^100: rounds to 0. + n = 0 + } else { + let down = UInt128(-shift) + let scaled = significand * scale + n = (scaled + (1 << (down - 1))) >> down + } + var text = String(n) + if digits > 0 { + if text.count <= digits { + text = String(repeating: "0", count: digits + 1 - text.count) + text + } + text.insert(".", at: text.index(text.endIndex, offsetBy: -digits)) + } + return negative ? "-" + text : text +} diff --git a/swift/Sources/Zero/Client/QueryContext.swift b/swift/Sources/Zero/Client/QueryContext.swift index e35d489155..44af04d99a 100644 --- a/swift/Sources/Zero/Client/QueryContext.swift +++ b/swift/Sources/Zero/Client/QueryContext.swift @@ -18,6 +18,11 @@ final class BranchDelegate: BuilderDelegate { func aggregateKeyColumns(_ table: String, _ columns: [String]) -> [String] { branch.serverColumnNames(table, columns) } + /// Wraps each source connection of a pipeline, to time its pushes. + var decorateSource: ((any SourceInput, String) -> any Input)? + func decorateSourceInput(_ input: any SourceInput, queryID: String) -> any Input { + decorateSource?(input, queryID) ?? input + } } /// Materializes queries over an IVM branch, registers named ones with the @@ -51,6 +56,10 @@ final class QueryContext { private var warnedLocalOnly = Set() /// Counts the view notifications under way. let notifications = NotificationGate() + /// Records a query metric (`addMetric`); nil for a mutator's queries. + private let addMetric: ((ClientMetricName, Double, String, AST?) -> Void)? + /// The monotonic clock the metrics are timed with. + private let monotonicNow: () -> Double /// Whether listeners are being told about a change. JS mutators are /// async, so a mutation made from a listener commits once the @@ -65,7 +74,9 @@ final class QueryContext { microtask: @escaping (@escaping () -> Void) -> Void, log: Log? = nil, closedReason: @escaping () -> ConnectionError? = { nil }, - aggregatesFromSource: Bool = true + aggregatesFromSource: Bool = true, + addMetric: ((ClientMetricName, Double, String, AST?) -> Void)? = nil, + monotonicNow: @escaping () -> Double = { 0 } ) { self.branch = branch self.delegate = BranchDelegate(branch, aggregatesFromSource: aggregatesFromSource) @@ -75,6 +86,15 @@ final class QueryContext { self.microtask = microtask self.log = log self.closedReason = closedReason + self.addMetric = addMetric + self.monotonicNow = monotonicNow + if let addMetric { + delegate.decorateSource = { input, queryID in + MeasurePush( + input, queryID: queryID, sampleRate: Int(updateSampleRate), now: monotonicNow + ) { elapsed, queryID in addMetric(.queryUpdateClient, elapsed, queryID, nil) } + } + } } /// The error of a query waiting for the server on a closed client: the @@ -225,8 +245,12 @@ final class QueryContext { var view: ArrayView? var destroyed = false + let addMetric = self.addMetric + let monotonicNow = self.monotonicNow + var t0 = 0.0 let maybeResolveComplete = { if attached, got == .yes, case .pending = queryComplete { + addMetric?(.queryMaterializationEndToEnd, monotonicNow() - t0, queryID, ast) queryComplete = .complete settle(.complete) } @@ -253,6 +277,7 @@ final class QueryContext { var removeCommitObserver: (() -> Void)? var removePendingAttach: (() -> Void)? + t0 = monotonicNow() let removeAddedQuery = customQueryID.map { addCustomQuery(ast, $0, ttl, gotCallback) } ?? {} @@ -339,6 +364,7 @@ final class QueryContext { removePendingAttach = onPipelinesReady { [weak created] in removePendingAttach = nil if deferred.destroyed { return {} } + let t1 = monotonicNow() do { created?.holdData() try deferred.attach() @@ -350,14 +376,18 @@ final class QueryContext { settle(.failed(e)) throw error } + let hydrateMs = monotonicNow() - t1 return { if destroyed { return } attached = true created?.releaseData() + addMetric?(.queryMaterializationClient, hydrateMs, queryID, nil) maybeResolveComplete() maybeMarkCached() } } + } else { + addMetric?(.queryMaterializationClient, monotonicNow() - t0, queryID, nil) } view = created diff --git a/swift/Sources/Zero/Client/QueryManager.swift b/swift/Sources/Zero/Client/QueryManager.swift index 0d2fdff5de..513a4ad3d0 100644 --- a/swift/Sources/Zero/Client/QueryManager.swift +++ b/swift/Sources/Zero/Client/QueryManager.swift @@ -7,6 +7,27 @@ enum Got: Equatable { typealias GotCallback = (Got, ErroredQuery?) -> Void +/// The client's query metrics (`ClientMetricMap` in +/// `packages/zql/src/query/metrics-delegate.ts`). +enum ClientMetricName: String { + case queryMaterializationClient = "query-materialization-client" + case queryMaterializationEndToEnd = "query-materialization-end-to-end" + case queryUpdateClient = "query-update-client" +} + +/// `UPDATE_SAMPLE_RATE`: client query updates are timed for only the first +/// of every this many pushes to each source connection, and each time is +/// recorded with this weight. +let updateSampleRate: Double = 32 + +/// One query's client metrics (`PerQueryClientMetric`): its materialization +/// times, recorded once (the last value wins), and its update times. +final class QueryClientMetrics { + var materializationClient: Double? + var materializationEndToEnd: Double? + let updateClient = TDigest() +} + /// A dictionary that iterates in insertion order, like a JS `Map`. struct OrderedMap { private(set) var keys: [Key] = [] @@ -74,15 +95,31 @@ final class QueryManager { private var batchTimer: (any ScheduledTimer)? private var closedError: ConnectionError? private var nextCallbackID = 0 + private let serverToClient: NameMapper + private let log: Log? + private let slowMaterializeThreshold: Double + /// Every query's materialization and update times (`metrics`). + let metrics = ( + materializationClient: TDigest(), materializationEndToEnd: TDigest(), + updateClient: TDigest() + ) + private var queryMetrics: [String: QueryClientMetrics] = [:] + /// Hears each query dropped from the client, with its client AST and + /// metrics (`setQueryEvictedCallback`). + var onQueryEvicted: ((String, AST, QueryClientMetrics) -> Void)? init( clientID: String, schema: ZeroSchema, mutationTracker: MutationTracker, scheduler: any Scheduler, queryChangeThrottleMs: Double, recentQueriesMaxSize: Int, + slowMaterializeThreshold: Double = 5_000, log: Log? = nil, send: @escaping (ChangeDesiredQueriesBody) -> Void, onFatalError: @escaping (ConnectionError) -> Void ) { self.clientID = clientID self.clientToServer = NameMapper.clientToServer(schema) + self.serverToClient = NameMapper.serverToClient(schema) + self.slowMaterializeThreshold = slowMaterializeThreshold + self.log = log self.mutationTracker = mutationTracker self.scheduler = scheduler self.queryChangeThrottleMs = queryChangeThrottleMs @@ -309,9 +346,63 @@ final class QueryManager { guard entry.count == 0 else { return } recentQueries[hash] = () if recentQueries.count > recentQueriesMaxSize, let lru = recentQueries.keys.first { + let evictedAST = clientAST(lru.string) + let evictedMetrics = queryMetrics[lru.string] ?? QueryClientMetrics() queries[lru] = nil recentQueries[lru] = nil + queryMetrics[lru.string] = nil queueQueryChange(.del(hash: lru.string)) + if let evictedAST { onQueryEvicted?(lru.string, evictedAST, evictedMetrics) } + } + } + + // MARK: - Metrics + + /// The client AST of a registered query (`getAST`). + func clientAST(_ queryID: String) -> AST? { + guard let ast = queries[JSKey(queryID)]?.normalized else { return nil } + return (try? serverToClient.map(ast)) ?? ast + } + + /// `mapClientASTToServer`. + func serverAST(_ ast: AST) -> AST { + (try? clientToServer.map(ast)) ?? ast + } + + func queryMetrics(_ queryID: String) -> QueryClientMetrics? { queryMetrics[queryID] } + + /// Records a query metric, client-wide and for the query (`addMetric`). + /// An end-to-end materialization over `slowMaterializeThreshold` is + /// logged as a warning. + func addMetric(_ metric: ClientMetricName, _ value: Double, queryID: String, ast: AST?) { + let weight = metric == .queryUpdateClient ? updateSampleRate : 1 + switch metric { + case .queryMaterializationClient: metrics.materializationClient.add(value, weight: weight) + case .queryMaterializationEndToEnd: metrics.materializationEndToEnd.add(value, weight: weight) + case .queryUpdateClient: metrics.updateClient.add(value, weight: weight) + } + if metric == .queryMaterializationEndToEnd, let log { + let details = { () -> String in + let astText = ast.map { JSONSerializer.stringify($0.json) } ?? "undefined" + return "\(queryID) \(astText) \(JSNumber.format(value))" + } + if value > slowMaterializeThreshold { + log.warn("Slow query materialization (including server/network) \(details())") + } else { + log.debug("Materialized query (including server/network) \(details())") + } + } + let existing: QueryClientMetrics + if let m = queryMetrics[queryID] { + existing = m + } else { + existing = QueryClientMetrics() + queryMetrics[queryID] = existing + } + switch metric { + case .queryUpdateClient: existing.updateClient.add(value, weight: weight) + case .queryMaterializationClient: existing.materializationClient = value + case .queryMaterializationEndToEnd: existing.materializationEndToEnd = value } } } diff --git a/swift/Sources/Zero/Client/ZeroClient.swift b/swift/Sources/Zero/Client/ZeroClient.swift index 517b2954aa..c930e87bcf 100644 --- a/swift/Sources/Zero/Client/ZeroClient.swift +++ b/swift/Sources/Zero/Client/ZeroClient.swift @@ -23,6 +23,7 @@ public final class Zero { private var preloads: [PreloadRecord] = [] private var outcomeListeners: [(id: Int, listener: (MutationServerOutcome) -> Void)] = [] private var nextOutcomeListenerID = 0 + private var inspectorStorage: Inspector? #if canImport(Darwin) private var lifecycle: AppLifecycle? #endif @@ -79,7 +80,17 @@ public final class Zero { public var schema: ZeroSchema { engine.schema } public var closed: Bool { engine.closed } + /// Inspects the client's queries, local data, and metrics, and the + /// server's, for debugging (`zero.inspector`). + public var inspector: Inspector { + if let inspectorStorage { return inspectorStorage } + let inspector = Inspector { [weak self] in self?.engine } + inspectorStorage = inspector + return inspector + } + private func attach(_ engine: Engine) { + inspectorStorage?.state.attach(engine) if options.onClientStateNotFound == nil { engine.onClientStateNotFound = { [weak self] in self?.replaceClient() } } diff --git a/swift/Sources/Zero/IVM/MeasurePush.swift b/swift/Sources/Zero/IVM/MeasurePush.swift new file mode 100644 index 0000000000..bbdf3f4f3a --- /dev/null +++ b/swift/Sources/Zero/IVM/MeasurePush.swift @@ -0,0 +1,56 @@ +/// Times the pushes and reconciles of a source connection for the query's +/// `query-update-client` metric (`MeasurePushOperator` in +/// `packages/zql/src/query/measure-push-operator.ts`). +/// +/// With a `sampleRate` of n, only the first of every n pushes (and +/// reconciles) is timed and reported; the metric weights each time by n. +final class MeasurePush: Operator { + let input: any Input + private let queryID: String + private let sampleRate: Int + private let now: () -> Double + private let report: (Double, String) -> Void + /// Pushes and reconciles so far, sampled or not. + private var count = 0 + weak var output: (any Output)? + + init( + _ input: any Input, queryID: String, sampleRate: Int, now: @escaping () -> Double, + report: @escaping (Double, String) -> Void + ) { + self.input = input + self.queryID = queryID + self.sampleRate = sampleRate + self.now = now + self.report = report + input.setOutput(self) + } + + func setOutput(_ output: any Output) { self.output = output } + func getSchema() -> SourceSchema { input.getSchema() } + func destroy() { input.destroy() } + func fetch(_ req: FetchRequest) -> NodeStream { input.fetch(req) } + + func push(_ change: Change, pusher: any InputBase) { + guard let output else { return } + measure(sample()) { output.push(change, pusher: self) } + } + + func reconcile(pusher: any InputBase) { + // A view has no `reconcile` in JS, which then neither counts nor times one. + guard let output, !(output is ArrayView) else { return } + measure(sample()) { output.reconcile(pusher: self) } + } + + private func sample() -> Bool { + defer { count += 1 } + return count % sampleRate == 0 + } + + private func measure(_ sampled: Bool, _ body: () -> Void) { + guard sampled else { return body() } + let start = now() + body() + report(now() - start, queryID) + } +} diff --git a/swift/Sources/Zero/Storage/ReplicaDatabase.swift b/swift/Sources/Zero/Storage/ReplicaDatabase.swift index 49c3867058..362167ebd4 100644 --- a/swift/Sources/Zero/Storage/ReplicaDatabase.swift +++ b/swift/Sources/Zero/Storage/ReplicaDatabase.swift @@ -386,6 +386,12 @@ final class ReplicaDatabase { } } + /// The stored snapshot of group `id`, for the inspector's look at a + /// client of another group. + func storedEntries(ofGroup id: String) throws -> KVMap { + try loadGroup(id).entries + } + /// Applies `ops` to the in-memory state now and queues them for SQLite. func apply(_ ops: [StorageOp]) { guard !dropped else { return } diff --git a/swift/Sources/Zero/Util/TDigest.swift b/swift/Sources/Zero/Util/TDigest.swift new file mode 100644 index 0000000000..f6a869aaa7 --- /dev/null +++ b/swift/Sources/Zero/Util/TDigest.swift @@ -0,0 +1,544 @@ +import Synchronization + +#if canImport(Darwin) + import Darwin +#elseif canImport(Glibc) + import Glibc +#elseif canImport(Musl) + import Musl +#endif + +/// A t-digest: an online summary of a distribution that answers rank +/// queries, such as percentiles (`TDigest` in `packages/shared/src/tdigest.ts`, +/// after influxdata/tdigest, Apache License 2.0). +/// +/// It is the JS digest's port, value for value: the same values, added in +/// the same order, give the same centroids, quantiles, and JSON as +/// `@rocicorp/zero`, which sends its metrics in that JSON. As in JS, values +/// are buffered and merged into the centroids when the buffer fills or the +/// digest is read, so reading between adds changes what later reads return. +/// +/// A reference type like the JS class: the inspector's ``Metrics`` share +/// the client's live digests. It is safe to use from any thread. +public final class TDigest: Sendable { + /// About how many centroids the digest keeps. + public let compression: Double + private let state: Mutex + + /// A centroid: the mean of the values merged into it, and their weight. + public struct Centroid: Sendable, Hashable { + public var mean: Double + public var weight: Double + + public init(mean: Double, weight: Double) { + self.mean = mean + self.weight = weight + } + } + + /// JSON that isn't a digest's (`fromJSON`'s "Invalid centroids array"). + public struct InvalidJSON: Error, Sendable, Hashable, CustomStringConvertible { + public var description: String { "Invalid centroids array" } + } + + /// A digest that keeps roughly `compression` centroids, and buffers up to + /// eight times as many values between merges. + public init(compression: Double = 100) { + self.compression = compression + state = Mutex(State(maxUnprocessed: (compression.rounded(.up)) * 8)) + } + + /// The digest of `toJSON()`'s form: the compression, then each + /// centroid's mean and weight (`fromJSON`). + public convenience init(json: [Double]) throws(InvalidJSON) { + guard json.count % 2 == 1 else { throw InvalidJSON() } + self.init(compression: json[0]) + var i = 1 + while i < json.count { + add(json[i], weight: json[i + 1]) + i += 2 + } + } + + /// Empties the digest. + public func reset() { + state.withLock { $0.reset() } + } + + /// Adds a value with the given weight. A NaN value, and a weight that is + /// not finite or not positive, are ignored. + public func add(_ mean: Double, weight: Double = 1) { + state.withLock { $0.add(mean, weight, compression: compression) } + } + + /// Adds the centroids of `other` (`merge`). `other` may be this digest. + public func merge(_ other: TDigest) { + let centroids = other.state.withLock { s in + s.process(compression: other.compression) + return (s.means, s.weights) + } + state.withLock { s in + for i in centroids.0.indices { + s.add(centroids.0[i], centroids.1[i], compression: compression) + } + } + } + + /// The centroids, in order of mean. + public func centroids() -> [Centroid] { + state.withLock { s in + s.process(compression: compression) + return zip(s.means, s.weights).map { Centroid(mean: $0, weight: $1) } + } + } + + /// The total weight of the values added. + public func count() -> Double { + state.withLock { s in + s.process(compression: compression) + return s.processedWeight + } + } + + /// The compression, then each centroid's mean and weight (`toJSON`). + public func toJSON() -> [Double] { + state.withLock { s in + s.process(compression: compression) + var data = [compression] + data.reserveCapacity(1 + 2 * s.means.count) + for i in s.means.indices { + data.append(s.means[i]) + data.append(s.weights[i]) + } + return data + } + } + + /// The approximate `q` quantile of the values, `q` between 0 and 1. NaN + /// when the digest is empty or `q` is out of range. + public func quantile(_ q: Double) -> Double { + state.withLock { $0.quantile(q, compression: compression) } + } + + /// The approximate fraction of the values at or below `x`. + public func cdf(_ x: Double) -> Double { + state.withLock { $0.cdf(x, compression: compression) } + } +} + +extension TDigest { + fileprivate struct State { + let maxUnprocessed: Double + /// Means of the processed centroids, ascending, and their weights. + var means: [Double] = [] + var weights: [Double] = [] + /// The values added since the last `process`, in order. + var unprocessedMeans: [Double] = [] + var unprocessedWeights: [Double] = [] + var cumulative: [Double] = [] + var processedWeight: Double = 0 + var unprocessedWeight: Double = 0 + var min = Double.greatestFiniteMagnitude + var max = -Double.greatestFiniteMagnitude + + init(maxUnprocessed: Double) { + self.maxUnprocessed = maxUnprocessed + } + + mutating func reset() { + self = State(maxUnprocessed: maxUnprocessed) + } + + mutating func add(_ mean: Double, _ weight: Double, compression: Double) { + if mean.isNaN || weight <= 0 || !weight.isFinite { return } + unprocessedMeans.append(mean) + unprocessedWeights.append(weight) + unprocessedWeight += weight + if Double(unprocessedMeans.count) > maxUnprocessed { process(compression: compression) } + } + + /// Merges the unprocessed values into the centroids: walks both in + /// order of mean, folding each into the last centroid while that stays + /// within its size limit, and starting a new centroid otherwise. + mutating func process(compression: Double) { + if unprocessedMeans.isEmpty { return } + let (newMeans, newWeights) = sortedUnprocessed() + let oldMeans = means + let oldWeights = weights + let n = newMeans.count + let p = oldMeans.count + unprocessedMeans.removeAll(keepingCapacity: true) + unprocessedWeights.removeAll(keepingCapacity: true) + processedWeight += unprocessedWeight + unprocessedWeight = 0 + let total = processedWeight + + var outMeans: [Double] = [] + var outWeights: [Double] = [] + var mean = 0.0 + var weight = 0.0 + var soFar = 0.0 + var limit = 0.0 + var i = 0 + var j = 0 + while i < n || j < p { + // On a tie the new value goes first. + let m: Double + let w: Double + if i < n && (j == p || newMeans[i] <= oldMeans[j]) { + m = newMeans[i] + w = newWeights[i] + i += 1 + } else { + m = oldMeans[j] + w = oldWeights[j] + j += 1 + } + if soFar + w <= limit { + weight += w + mean += (w * (m - mean)) / weight + } else { + if weight > 0 { + outMeans.append(mean) + outWeights.append(weight) + } + let k = integratedLocation(soFar / total, compression) + limit = total * integratedQ(k + 1, compression) + mean = m + weight = w + } + soFar += w + } + outMeans.append(mean) + outWeights.append(weight) + means = outMeans + weights = outWeights + min = jsMin(min, outMeans[0]) + max = jsMax(max, outMeans[outMeans.count - 1]) + } + + /// The unprocessed means and weights, sorted by mean as JS sorts them. + private func sortedUnprocessed() -> ([Double], [Double]) { + let means = unprocessedMeans + let weights = unprocessedWeights + if isSorted(means) { return (means, weights) } + // One weight for all the values: JS sorts the means alone, with a + // typed array's numeric sort, which puts -0 before +0. + if weights.allSatisfy({ $0 == weights[0] }) { + return ( + means.sorted { $0 < $1 || ($0 == $1 && $0.sign == .minus && $1.sign == .plus) }, + weights + ) + } + // A stable sort by mean, as JS's with `(a, b) => means[a] - means[b]`. + let order = means.indices.sorted { a, b in + means[a] < means[b] || (!(means[b] < means[a]) && a < b) + } + return (order.map { means[$0] }, order.map { weights[$0] }) + } + + private func isSorted(_ values: [Double]) -> Bool { + var i = 1 + while i < values.count { + if values[i] < values[i - 1] { return false } + i += 1 + } + return true + } + + private mutating func updateCumulative() { + // Weight only grows, so an unchanged last value means nothing changed. + if let last = cumulative.last, last == processedWeight { return } + var out: [Double] = [] + out.reserveCapacity(means.count + 1) + var prev = 0.0 + for i in means.indices { + let cur = weights[i] + out.append(prev + cur / 2) + prev += cur + } + out.append(prev) + cumulative = out + } + + mutating func quantile(_ q: Double, compression: Double) -> Double { + process(compression: compression) + updateCumulative() + if q < 0 || q > 1 || means.isEmpty { return .nan } + if means.count == 1 { return means[0] } + // JS reads past the arrays' ends here, which makes NaN. + if q.isNaN { return .nan } + let index = q * processedWeight + if index <= weights[0] / 2 { + return min + ((2 * index) / weights[0]) * (means[0] - min) + } + let cumulative = self.cumulative + let lower = binarySearch(cumulative.count) { -cumulative[$0] + index } + // Past the last cumulative weight, which rounding can leave below the + // total weight: JS reads past the arrays' ends, which makes NaN. + if lower == cumulative.count { return .nan } + if lower + 1 != cumulative.count { + let z1 = index - cumulative[lower - 1] + let z2 = cumulative[lower] - index + return weightedAverage(means[lower - 1], z2, means[lower], z1) + } + let z1 = index - processedWeight - weights[lower - 1] / 2 + let z2 = weights[lower - 1] / 2 - z1 + return weightedAverage(means[means.count - 1], z1, max, z2) + } + + mutating func cdf(_ x: Double, compression: Double) -> Double { + process(compression: compression) + updateCumulative() + switch means.count { + case 0: + return 0 + case 1: + let width = max - min + if x <= min { return 0 } + if x >= max { return 1 } + if x - min <= width { return 0.5 } + return (x - min) / width + default: + break + } + if x <= min { return 0 } + if x >= max { return 1 } + // JS reads past the arrays' ends here, which makes NaN. + if x.isNaN { return .nan } + let m0 = means[0] + if x <= m0 { + if m0 - min > 0 { + return (((x - min) / (m0 - min)) * weights[0]) / processedWeight / 2 + } + return 0 + } + let mn = means[means.count - 1] + if x >= mn { + if max - mn > 0 { + return 1 - (((max - x) / (max - mn)) * weights[weights.count - 1]) / processedWeight / 2 + } + return 1 + } + let means = self.means + // `x - means[i] || 1`: equal counts as greater. + let upper = binarySearch(means.count) { i in + let d = x - means[i] + return d == 0 || d.isNaN ? 1 : d + } + let z1 = x - means[upper - 1] + let z2 = means[upper] - x + return weightedAverage(cumulative[upper - 1], z2, cumulative[upper], z1) / processedWeight + } + + private func integratedQ(_ k: Double, _ compression: Double) -> Double { + (fdlibmSin((jsMin(k, compression) * Double.pi) / compression - Double.pi / 2) + 1) / 2 + } + + private func integratedLocation(_ q: Double, _ compression: Double) -> Double { + (compression * (fdlibmAsin(2 * q - 1) + Double.pi / 2)) / Double.pi + } + } +} + +/// `binarySearch` (`packages/shared/src/binary-search.ts`): the index of the +/// first element `compare` doesn't place after, where `compare` is positive +/// for an element before the one sought and 0 for it. +private func binarySearch(_ high: Int, _ compare: (Int) -> Double) -> Int { + var low = 0 + var high = high + while low < high { + let mid = low + ((high - low) >> 1) + let c = compare(mid) + if c == 0 { return mid } + if c > 0 { + low = mid + 1 + } else { + high = mid + } + } + return low +} + +private func weightedAverage(_ x1: Double, _ w1: Double, _ x2: Double, _ w2: Double) -> Double { + x1 <= x2 + ? weightedAverageSorted(x1, w1, x2, w2) : weightedAverageSorted(x2, w2, x1, w1) +} + +private func weightedAverageSorted( + _ x1: Double, _ w1: Double, _ x2: Double, _ w2: Double +) -> Double { + let x = (x1 * w1 + x2 * w2) / (w1 + w2) + return jsMax(x1, jsMin(x, x2)) +} + +/// `Math.min` of two numbers: NaN if either is, and -0 below +0. +private func jsMin(_ a: Double, _ b: Double) -> Double { + if a.isNaN || b.isNaN { return .nan } + if a == b { return a.sign == .minus ? a : b } + return a < b ? a : b +} + +/// `Math.max` of two numbers: NaN if either is, and +0 above -0. +private func jsMax(_ a: Double, _ b: Double) -> Double { + if a.isNaN || b.isNaN { return .nan } + if a == b { return a.sign == .plus ? a : b } + return a > b ? a : b +} + +// `Math.asin` and `Math.sin` as Node 22 computes them: fdlibm, as V8 12's +// `src/base/ieee754.cc` has it. The platform's functions round some results +// the other way, which moves the digest's centroid limits. + +private func highWord(_ x: Double) -> Int32 { Int32(truncatingIfNeeded: x.bitPattern >> 32) } + +private func lowWord(_ x: Double) -> UInt32 { UInt32(truncatingIfNeeded: x.bitPattern) } + +private func withLowWordZero(_ x: Double) -> Double { + Double(bitPattern: x.bitPattern & 0xFFFF_FFFF_0000_0000) +} + +/// fdlibm's `asin`. +func fdlibmAsin(_ x: Double) -> Double { + let one = 1.0 + let huge = 1.000e+300 + let pio2Hi = Double(bitPattern: 0x3FF9_21FB_5444_2D18) + let pio2Lo = Double(bitPattern: 0x3C91_A626_3314_5C07) + let pio4Hi = Double(bitPattern: 0x3FE9_21FB_5444_2D18) + let pS0 = Double(bitPattern: 0x3FC5_5555_5555_5555) + let pS1 = Double(bitPattern: 0xBFD4_D612_03EB_6F7D) + let pS2 = Double(bitPattern: 0x3FC9_C155_0E88_4455) + let pS3 = Double(bitPattern: 0xBFA4_8228_B568_8F3B) + let pS4 = Double(bitPattern: 0x3F49_EFE0_7501_B288) + let pS5 = Double(bitPattern: 0x3F02_3DE1_0DFD_F709) + let qS1 = Double(bitPattern: 0xC003_3A27_1C8A_2D4B) + let qS2 = Double(bitPattern: 0x4000_2AE5_9C59_8AC8) + let qS3 = Double(bitPattern: 0xBFE6_066C_1B8D_0159) + let qS4 = Double(bitPattern: 0x3FB3_B8C5_B12E_9282) + + var t = 0.0 + let hx = highWord(x) + let ix = hx & 0x7FFF_FFFF + if ix >= 0x3FF0_0000 { + // |x| >= 1 + if (UInt32(bitPattern: ix - 0x3FF0_0000) | lowWord(x)) == 0 { + return x * pio2Hi + x * pio2Lo + } + return .nan + } else if ix < 0x3FE0_0000 { + // |x| < 0.5 + if ix < 0x3E40_0000 { + if huge + x > one { return x } + } else { + t = x * x + } + let p = t * (pS0 + t * (pS1 + t * (pS2 + t * (pS3 + t * (pS4 + t * pS5))))) + let q = one + t * (qS1 + t * (qS2 + t * (qS3 + t * qS4))) + let w = p / q + return x + x * w + } + // 1 > |x| >= 0.5 + var w = one - abs(x) + t = w * 0.5 + var p = t * (pS0 + t * (pS1 + t * (pS2 + t * (pS3 + t * (pS4 + t * pS5))))) + var q = one + t * (qS1 + t * (qS2 + t * (qS3 + t * qS4))) + let s = t.squareRoot() + if ix >= 0x3FEF_3333 { + // |x| > 0.975 + w = p / q + t = pio2Hi - (2.0 * (s + s * w) - pio2Lo) + } else { + w = withLowWordZero(s) + let c = (t - w * w) / (s + w) + let r = p / q + p = 2.0 * s * r - (pio2Lo - 2.0 * c) + q = pio4Hi - 2.0 * w + t = pio4Hi - (p - q) + } + return hx > 0 ? t : -t +} + +/// fdlibm's `sin`, for |x| < 3π/4, which covers the digest's arguments +/// (within [-π/2, π/2]). Beyond, the platform's. +func fdlibmSin(_ x: Double) -> Double { + let hx = highWord(x) + let ix = hx & 0x7FFF_FFFF + if ix <= 0x3FE9_21FB { + // |x| ~< π/4 + return kernelSin(x, 0, 0) + } + guard ix < 0x4002_D97C else { return sin(x) } + // |x| < 3π/4: `__ieee754_rem_pio2` with n = ±1. + let pio2_1 = Double(bitPattern: 0x3FF9_21FB_5440_0000) + let pio2_1t = Double(bitPattern: 0x3DD0_B461_1A62_6331) + let pio2_2 = Double(bitPattern: 0x3DD0_B461_1A60_0000) + let pio2_2t = Double(bitPattern: 0x3BA3_198A_2E03_7073) + var y0: Double + var y1: Double + if hx > 0 { + var z = x - pio2_1 + if ix != 0x3FF9_21FB { + y0 = z - pio2_1t + y1 = (z - y0) - pio2_1t + } else { + z -= pio2_2 + y0 = z - pio2_2t + y1 = (z - y0) - pio2_2t + } + // n = 1 + return kernelCos(y0, y1) + } + var z = x + pio2_1 + if ix != 0x3FF9_21FB { + y0 = z + pio2_1t + y1 = (z - y0) + pio2_1t + } else { + z += pio2_2 + y0 = z + pio2_2t + y1 = (z - y0) + pio2_2t + } + // n = -1, so n & 3 = 3 + return -kernelCos(y0, y1) +} + +/// fdlibm's `__kernel_sin`. +private func kernelSin(_ x: Double, _ y: Double, _ iy: Int) -> Double { + let half = 0.5 + let s1 = Double(bitPattern: 0xBFC5_5555_5555_5549) + let s2 = Double(bitPattern: 0x3F81_1111_1110_F8A6) + let s3 = Double(bitPattern: 0xBF2A_01A0_19C1_61D5) + let s4 = Double(bitPattern: 0x3EC7_1DE3_57B1_FE7D) + let s5 = Double(bitPattern: 0xBE5A_E5E6_8A2B_9CEB) + let s6 = Double(bitPattern: 0x3DE5_D93A_5ACF_D57C) + let ix = highWord(x) & 0x7FFF_FFFF + if ix < 0x3E40_0000 { return x } + let z = x * x + let v = z * x + let r = s2 + z * (s3 + z * (s4 + z * (s5 + z * s6))) + if iy == 0 { return x + v * (s1 + z * r) } + return x - ((z * (half * y - v * r) - y) - v * s1) +} + +/// fdlibm's `__kernel_cos`. +private func kernelCos(_ x: Double, _ y: Double) -> Double { + let one = 1.0 + let c1 = Double(bitPattern: 0x3FA5_5555_5555_554C) + let c2 = Double(bitPattern: 0xBF56_C16C_16C1_5177) + let c3 = Double(bitPattern: 0x3EFA_01A0_19CB_1590) + let c4 = Double(bitPattern: 0xBE92_7E4F_809C_52AD) + let c5 = Double(bitPattern: 0x3E21_EE9E_BDB4_B1C4) + let c6 = Double(bitPattern: 0xBDA8_FAE9_BE88_38D4) + let ix = highWord(x) & 0x7FFF_FFFF + if ix < 0x3E40_0000 { return one } + let z = x * x + let r = z * (c1 + z * (c2 + z * (c3 + z * (c4 + z * (c5 + z * c6))))) + if ix < 0x3FD3_3333 { + return one - (0.5 * z - (z * r - x * y)) + } + let qx: Double = + ix > 0x3FE9_0000 + ? 0.28125 : Double(bitPattern: UInt64(UInt32(bitPattern: ix - 0x0020_0000)) << 32) + let iz = 0.5 * z - qx + let a = one - qx + return a - (iz - (z * r - x * y)) +} diff --git a/swift/Sources/Zero/Zero.docc/JSParity.md b/swift/Sources/Zero/Zero.docc/JSParity.md index a1b4d0b0a7..a029ec6b65 100644 --- a/swift/Sources/Zero/Zero.docc/JSParity.md +++ b/swift/Sources/Zero/Zero.docc/JSParity.md @@ -31,8 +31,8 @@ are arguments of | `kvStore` (`'idb'` or `'mem'`) | `storage`: `.sqlite()` or `.memory` | | `hiddenTabDisconnectDelay` | none: the client pushes pending mutations and disconnects when the app goes to the background, and reconnects when it comes back | | `onOnlineChange` (deprecated) | ``Connection/subscribe(_:)`` | -| `getTraceparent`, `batchViewUpdates` | the same | -| `slowMaterializeThreshold`, `maxRecentQueries` (deprecated) | not ported | +| `getTraceparent`, `batchViewUpdates`, `slowMaterializeThreshold` | the same | +| `maxRecentQueries` (deprecated) | not ported | ## The client @@ -50,7 +50,8 @@ are arguments of | `zero.close()` | `zero.close()`; dropping the last reference to a `Zero` closes it too | | `zero.mutate.issue.insert(...)` (CRUD mutators) | not ported: write a mutator | | `zero.delete()`, `dropAllDatabases()` | `await zero.delete()`, `await dropAllDatabases(_:)`, which return the ``DroppedDatabases`` | -| `zero.mutateBatch`, `zero.online`, `zero.onOnline`, `zero.inspector`, `zero.server`, `zero.storageKey`, `zero.idbName`, `zero.kvStore`, `zero.schemaVersion` | not ported | +| `zero.inspector`: `client`, `clientGroup`, `metrics()`, `clients()`, `clientsWithQueries()`, `serverVersion()`, `analyzeQuery`, `analyzeServerAST`, `analyzeNamedQuery`, `authenticate`, `formatPlannerEvents` | `zero.inspector`, an ``Inspector`` with the same members, whose methods are `async throws` (`formatPlannerEvents` isn't); its queries are ``InspectorQuery`` values, and the metrics ``TDigest``s. Where JS asks for the admin password in a dialog, a request throws ``InspectorError/authenticationRequired``: call `authenticate(password:)` first | +| `zero.mutateBatch`, `zero.online`, `zero.onOnline`, `zero.server`, `zero.storageKey`, `zero.idbName`, `zero.kvStore`, `zero.schemaVersion` | not ported | | `zero.version` | ``protocolVersion``, the sync protocol version | When the server no longer has the client's state, JS asks the app to diff --git a/swift/Tests/ConformanceTests/ClientConformanceTests.swift b/swift/Tests/ConformanceTests/ClientConformanceTests.swift index 8657f7c8e3..0c9436b37f 100644 --- a/swift/Tests/ConformanceTests/ClientConformanceTests.swift +++ b/swift/Tests/ConformanceTests/ClientConformanceTests.swift @@ -147,6 +147,7 @@ final class ScenarioRunner { if let v = o["disconnectTimeoutMs"]?.numberValue { options.disconnectTimeoutMs = v } if let v = o["queryChangeThrottleMs"]?.numberValue { options.queryChangeThrottleMs = v } if let v = o["maxHeaderLength"]?.numberValue { options.maxHeaderLength = Int(v) } + if let v = o["slowMaterializeThreshold"]?.numberValue { options.slowMaterializeThreshold = v } options.hiddenDisconnectDelayMs = o["hiddenTabDisconnectDelay"]?.numberValue ?? 300_000 options.mutateURL = o["mutateURL"]?.stringValue options.mutateHeaders = o["mutateHeaders"]?.objectValue?.dictionary.compactMapValues( diff --git a/swift/Tests/ZeroTests/InspectorTests.swift b/swift/Tests/ZeroTests/InspectorTests.swift new file mode 100644 index 0000000000..b0a297130a --- /dev/null +++ b/swift/Tests/ZeroTests/InspectorTests.swift @@ -0,0 +1,563 @@ +import Testing +import ZeroTesting + +@testable import Zero + +/// `zero.inspector`, the query metrics, and `slowMaterializeThreshold`, +/// against a mock zero-cache. The formatting tests expect what the JS +/// sources (`astToZQL`, `formatPlannerEvents`, `toFixed`) return for the +/// same input, run with the repo's Node. +@Suite @MainActor struct InspectorTests { + static let all = defineQuery("issues.all") { () in zql.issue.orderBy(\.id, .asc) } + static let hash = QueryHash.hashOfNameAndArgs("issues.all", []) + + let scheduler = ManualScheduler(now: 1_700_000_000_000) + let sockets = MockSocketFactory() + let log = RecordingLogSink() + + func makeZero( + storageKey: String = randomID(), sockets: MockSocketFactory? = nil, + _ configure: (inout ZeroOptions) -> Void = { _ in } + ) throws -> Zero { + var options = ZeroOptions( + cacheURL: "http://zero.test", userID: "u", storageKey: storageKey, storage: .memory) + options.logSink = log + // No pings while the clock moves. + options.pingTimeoutMs = 600_000 + configure(&options) + return try Zero( + schema: .test, options: options, scheduler: scheduler, socketFactory: sockets ?? self.sockets) + } + + /// Acknowledges the newest socket. + @discardableResult + func connect() throws -> MockSocket { + scheduler.runDueTimers() + let socket = try #require(sockets.sockets.last) + socket.receive(text: #"["connected",{"wsid":"w"}]"#) + scheduler.runDueTimers() + return socket + } + + func poke(_ socket: MockSocket, from base: String?, to cookie: String, _ part: String) { + let base = base.map { "\"\($0)\"" } ?? "null" + socket.receive(text: #"["pokeStart",{"pokeID":"\#(cookie)","baseCookie":\#(base)}]"#) + socket.receive(text: #"["pokePart",{"pokeID":"\#(cookie)",\#(part)}]"#) + socket.receive(text: #"["pokeEnd",{"pokeID":"\#(cookie)","cookie":"\#(cookie)"}]"#) + scheduler.runDueTimers() + } + + static let got = #""gotQueriesPatch":[{"op":"put","hash":"\#(hash)"}]"# + static func row(_ id: String) -> String { + #"{"op":"put","tableName":"issues","value":{"id":"\#(id)","title":"","description":"","closed":false,"owner_id":null,"createdAt":1}}"# + } + + /// Lets the inspector's tasks run, and the work they queue. + func settle() async { + for _ in 0..<5 { + await Task.yield() + scheduler.runDueTimers() + } + } + + /// The bodies of the `inspect` messages sent on `socket`. + func requests(_ socket: MockSocket) -> [JSONValue] { + socket.messages.compactMap { text in + let m = try? JSONParser.parse(Array(text.utf8)) + return m?[0] == "inspect" ? m?[1] : nil + } + } + + func answer(_ socket: MockSocket, _ request: JSONValue?, _ op: String, _ value: JSONValue) { + let message: JSONValue = [ + "inspect", ["id": request?["id"] ?? .null, "op": .string(op), "value": value], + ] + socket.receive(text: JSONSerializer.stringify(message)) + scheduler.runDueTimers() + } + + /// The request without its random ID. + func withoutID(_ request: JSONValue?) -> JSONValue? { + guard var o = request?.objectValue else { return nil } + o["id"] = nil + return .object(o) + } + + // MARK: - Metrics and slowMaterializeThreshold + + @Test func logsAMaterializationSlowerThanTheThreshold() async throws { + for (threshold, level, message) in [ + (5_000.0, LogLevel.warn, "Slow query materialization (including server/network)"), + (10_000, .debug, "Materialized query (including server/network)"), + ] { + let zero = try makeZero { + $0.slowMaterializeThreshold = threshold + $0.logLevel = .debug + } + let view = try zero.materialize(Self.all()) + let socket = try connect() + scheduler.advance(by: 6_000) + poke(socket, from: nil, to: "01", Self.got) + #expect(view.resultType == .complete) + let logged = log.entries.filter { $0.message.hasPrefix(message) }.last + #expect(logged?.level == level) + #expect(logged?.message.hasPrefix("\(message) \(Self.hash) {\"table\":\"issue\"") == true) + #expect(logged?.message.hasSuffix(" 6000") == true) + let metrics = try #require(zero.engine.queryManager.queryMetrics(Self.hash)) + #expect(metrics.materializationEndToEnd == 6_000) + // The manual clock doesn't move while the view hydrates. + #expect(metrics.materializationClient == 0) + let all = zero.engine.queryManager.metrics + #expect(all.materializationEndToEnd.count() == 1) + #expect(all.materializationEndToEnd.quantile(0.5) == 6_000) + #expect(all.materializationClient.count() == 1) + zero.close() + } + } + + @Test func weightsTheSampledUpdatesByTheSampleRate() throws { + let zero = try makeZero() + let view = try zero.materialize(Self.all()) + let socket = try connect() + poke(socket, from: nil, to: "01", Self.got) + let all = zero.engine.queryManager.metrics.updateClient + #expect(all.count() == 0) + // The first of every 32 pushes is timed, and recorded with weight 32. + var cookie = 1 + func pushRow() { + let next = cookie + 1 + let row = Self.row("i\(next)") + poke(socket, from: format2(cookie), to: format2(next), #""rowsPatch":[\#(row)]"#) + cookie = next + } + pushRow() + #expect(all.count() == 32) + #expect(all.quantile(0.5) == 0) + for _ in 0..<31 { pushRow() } + #expect(all.count() == 32) + pushRow() + #expect(all.count() == 64) + #expect(zero.engine.queryManager.queryMetrics(Self.hash)?.updateClient.count() == 64) + #expect(view.data.json?.arrayValue?.count == 33) + } + + func format2(_ n: Int) -> String { n < 10 ? "0\(n)" : "\(n)" } + + // MARK: - RPCs + + @Test func waitsForTheConnectionAndAnswersTheRequest() async throws { + let zero = try makeZero() + var ids = 0 + zero.engine.inspectRPCs.makeID = { + ids += 1 + return "r\(ids)" + } + let version = Task { try await zero.inspector.serverVersion() } + await settle() + let socket = try connect() + await settle() + #expect(requests(socket) == [["op": "version", "id": "r1"]]) + // Answers to other requests are ignored. + answer(socket, ["id": "other"], "version", "0") + answer(socket, requests(socket).last, "version", "0.26.0") + #expect(try await version.value == "0.26.0") + } + + @Test func throwsTheServersErrorsAndUnexpectedAnswers() async throws { + let zero = try makeZero() + let socket = try connect() + let failing = Task { try await zero.inspector.serverVersion() } + await settle() + answer(socket, requests(socket).last, "error", "boom") + await #expect(throws: InspectorError.server("boom")) { try await failing.value } + + let unexpected = Task { try await zero.inspector.serverVersion() } + await settle() + answer( + socket, requests(socket).last, "metrics", + ["query-materialization-server": [100], "query-update-server": [100]]) + await #expect(throws: InspectorError.unexpectedResponse(op: "metrics")) { + try await unexpected.value + } + } + + @Test func needsThePasswordAndPresentsItAgainAfterAReconnect() async throws { + let zero = try makeZero() + var socket = try connect() + let first = Task { try await zero.inspector.serverVersion() } + await settle() + answer(socket, requests(socket).last, "authenticated", false) + await #expect(throws: InspectorError.authenticationRequired) { try await first.value } + // No password yet, so none was presented. + #expect(requests(socket).map { $0["op"] } == ["version"]) + + let accepted = Task { try await zero.inspector.authenticate(password: "pw") } + await settle() + #expect(withoutID(requests(socket).last) == ["op": "authenticate", "value": "pw"]) + answer(socket, requests(socket).last, "authenticated", true) + #expect(try await accepted.value == true) + + // The server authenticates each connection. + socket.serverClose() + socket = try connect() + let second = Task { try await zero.inspector.serverVersion() } + await settle() + answer(socket, requests(socket).last, "authenticated", false) + await settle() + #expect(withoutID(requests(socket).last) == ["op": "authenticate", "value": "pw"]) + answer(socket, requests(socket).last, "authenticated", true) + await settle() + #expect(requests(socket).map { $0["op"] } == ["version", "authenticate", "version"]) + answer(socket, requests(socket).last, "version", "1") + #expect(try await second.value == "1") + + // A refused password is forgotten. + socket.serverClose() + socket = try connect() + let refused = Task { try await zero.inspector.serverVersion() } + await settle() + answer(socket, requests(socket).last, "authenticated", false) + await settle() + answer(socket, requests(socket).last, "authenticated", false) + await #expect(throws: InspectorError.authenticationRequired) { try await refused.value } + let again = Task { try await zero.inspector.serverVersion() } + await settle() + answer(socket, requests(socket).last, "authenticated", false) + await #expect(throws: InspectorError.authenticationRequired) { try await again.value } + #expect( + requests(socket).map { $0["op"] } == ["version", "authenticate", "version"]) + } + + @Test func failsThePendingRequestsWhenTheConnectionEndsOrZeroCloses() async throws { + let zero = try makeZero() + let socket = try connect() + let dropped = Task { try await zero.inspector.serverVersion() } + await settle() + socket.serverClose() + scheduler.runDueTimers() + let error = await #expect(throws: InspectorError.self) { try await dropped.value } + #expect(error == .connectionLost("WebSocket connection closed abruptly")) + + try connect() + let closing = Task { try await zero.inspector.serverVersion() } + let waiting = Task { try await zero.inspector.metrics() } + await settle() + zero.close() + await #expect(throws: InspectorError.clientClosed) { try await closing.value } + await #expect(throws: InspectorError.clientClosed) { try await waiting.value } + await #expect(throws: InspectorError.clientClosed) { + try await zero.inspector.serverVersion() + } + } + + @Test func mergesTheClientsMetricsWithTheServers() async throws { + let zero = try makeZero() + let view = try zero.materialize(Self.all()) + let socket = try connect() + scheduler.advance(by: 1_234) + poke(socket, from: nil, to: "01", Self.got) + let metrics = Task { try await zero.inspector.metrics() } + await settle() + #expect(withoutID(requests(socket).last) == ["op": "metrics"]) + answer( + socket, requests(socket).last, "metrics", + ["query-materialization-server": [100, 5, 1, 7, 2], "query-update-server": [100]]) + let m = try await metrics.value + // The client's digests are its live ones. + #expect( + m.queryMaterializationEndToEnd === zero.engine.queryManager.metrics.materializationEndToEnd) + #expect(m.queryMaterializationEndToEnd.quantile(0.5) == 1_234) + #expect(m.queryMaterializationServer.toJSON() == [100, 5, 1, 7, 2]) + #expect(m.queryUpdateServer.count() == 0) + #expect(m["query-update-client"] === m.queryUpdateClient) + view.destroy() + } + + @Test func listsTheQueriesWithTheClientsViewOfThem() async throws { + let zero = try makeZero() + // The inspector keeps the queries the client drops from then on. + let inspector = zero.inspector + let view = try zero.materialize(Self.all()) + let socket = try connect() + scheduler.advance(by: 40) + poke(socket, from: nil, to: "01", Self.got) + view.destroy() + let queries = Task { try await inspector.client.queries() } + await settle() + #expect( + withoutID(requests(socket).last) == ["op": "queries", "clientID": .string(zero.clientID)]) + let cid = JSONValue.string(zero.clientID) + answer( + socket, requests(socket).last, "queries", + [ + [ + "clientID": cid, "queryID": .string(Self.hash), + "ast": ["table": "issues", "orderBy": [["id", "asc"]]], "name": "issues.all", "args": [], + "got": true, "deleted": false, "ttl": 300_000, "inactivatedAt": nil, "rowCount": 2, + "metrics": [ + "query-hydration-server-ms": 3, "query-update-server": [100, 2, 3, 4, 1, 10, 2], + ], + ], + [ + "clientID": cid, "queryID": "gone", "ast": nil, "name": nil, "args": nil, "got": false, + "deleted": true, "ttl": -1, "inactivatedAt": 1_700_000_000_000, "rowCount": 0, + "metrics": nil, + ], + [ + "clientID": cid, "queryID": "slow", "ast": ["table": "users"], "name": nil, "args": nil, + "got": false, "deleted": false, "ttl": 0, "inactivatedAt": nil, "rowCount": 0, + "metrics": ["query-hydration-server-ms": 10, "query-update-server": [100]], + ], + ]) + let result = try await queries.value + // The most expensive on the server first. + #expect(result.map(\.id) == ["slow", Self.hash, "gone"]) + let all = result[1] + #expect(all.name == "issues.all") + #expect(all.args == []) + #expect(all.got) + #expect(all.ttl == "5m") + #expect(all.rowCount == 2) + #expect(all.clientZQL == "issue.orderBy('id', 'asc')") + #expect(all.serverZQL == "issues.orderBy('id', 'asc')") + #expect(all.hydrateClient == 0) + #expect(all.hydrateServer == 3) + #expect(all.hydrateTotal == 40) + #expect(all.updateServerP50 == 3.5) + #expect(all.updateServerP95 == 10) + #expect(all.updateClientP50 == nil) + #expect(all.metrics.queryMaterializationEndToEnd.toJSON() == [100, 40, 1]) + let gone = result[2] + #expect(gone.ttl == .forever) + #expect(gone.deleted) + #expect(gone.inactivatedAt == 1_700_000_000_000) + #expect(gone.clientZQL == nil) + #expect(gone.serverZQL == nil) + #expect(gone.hydrateServer == nil) + #expect(gone.metrics.queryUpdateServer.count() == 0) + #expect(result[0].ttl == TTL.none) + #expect(result[0].serverZQL == "users") + + let group = Task { try await inspector.clientGroup.queries() } + await settle() + #expect(withoutID(requests(socket).last) == ["op": "queries"]) + answer(socket, requests(socket).last, "queries", []) + #expect(try await group.value.isEmpty) + } + + @Test func readsTheLocalClientsAndTheirData() async throws { + let key = randomID() + let zero = try makeZero(storageKey: key) + let other = try makeZero(storageKey: key, sockets: MockSocketFactory()) + defer { other.close() } + let view = try zero.materialize(Self.all()) + let socket = try connect() + poke(socket, from: nil, to: "01", Self.got + #","rowsPatch":[\#(Self.row("i1"))]"#) + #expect(view.data.json?.arrayValue?.count == 1) + + let rows = try await zero.inspector.client.rows("issue") + #expect(rows.map { $0["id"] } == ["i1"]) + let map = try await zero.inspector.client.map() + #expect(map["e/issue/i1"]?["id"] == "i1") + #expect(map["g/\(Self.hash)"] != nil) + + let clients = try await zero.inspector.clients() + #expect(Set(clients.map(\.id)) == [zero.clientID, other.clientID]) + #expect(clients.allSatisfy { $0.clientGroup.id == zero.clientGroupID }) + #expect(try await zero.inspector.clientGroup.clients().count == 2) + // The other client shares the group's data. + let otherClient = try #require(clients.first { $0.id == other.clientID }) + #expect(try await otherClient.rows("issue").count == 1) + await #expect(throws: InspectorError.clientNotFound("nobody")) { + try await InspectorClient(zero.inspector.state, id: "nobody", clientGroupID: "g").map() + } + + // Every client is asked about at once; the answers come in any order. + let withQueries = Task { try await zero.inspector.clientsWithQueries() } + await settle() + let asked = requests(socket).suffix(2) + #expect(Set(asked.map { $0["clientID"] }) == [.string(zero.clientID), .string(other.clientID)]) + let otherRequest = asked.first { $0["clientID"] == .string(other.clientID) } + let ownRequest = asked.first { $0["clientID"] == .string(zero.clientID) } + answer( + socket, otherRequest, "queries", + [ + [ + "clientID": .string(other.clientID), "queryID": "q", "ast": nil, "name": "x", + "args": [], "got": true, "deleted": false, "ttl": 0, "inactivatedAt": nil, + "rowCount": 0, + ] + ]) + answer(socket, ownRequest, "queries", []) + #expect(try await withQueries.value.map(\.id) == [other.clientID]) + } + + @Test func analyzesQueriesByNameOrByTheServersAST() async throws { + let zero = try makeZero() + let socket = try connect() + let result: JSONValue = [ + "warnings": ["w"], "syncedRowCount": 1, "start": 1, "end": 5, "elapsed": 4, + "syncedRows": ["issues": [["id": "i1"]]], "readRowCountsByQuery": ["issues": ["q": 3]], + "joinPlans": [["type": "plan-failed", "attemptNumber": 0, "reason": "x"]], + ] + + let named = Task { try await zero.inspector.analyzeQuery(Self.all()) } + await settle() + #expect( + withoutID(requests(socket).last) == ["op": "analyze-query", "name": "issues.all", "args": []]) + answer(socket, requests(socket).last, "analyze-query", result) + let analyzed = try await named.value + #expect(analyzed.warnings == ["w"]) + #expect(analyzed.syncedRows?["issues"]?.map { $0["id"] } == ["i1"]) + #expect(analyzed.readRowCountsByQuery == ["issues": ["q": 3]]) + #expect(analyzed.elapsed == 4) + #expect(analyzed.joinPlans?.count == 1) + #expect(analyzed.json == result) + + // An unnamed query goes by its AST, with the server's names. + let unnamed = Task { + try await zero.inspector.analyzeQuery( + zql.issue.where(\.ownerId, "u1"), options: AnalyzeQueryOptions(syncedRows: true)) + } + await settle() + let sent = requests(socket).last + #expect(sent?["ast"]?["table"] == "issues") + #expect(sent?["ast"]?["where"]?["left"] == ["type": "column", "name": "owner_id"]) + #expect(sent?["options"] == ["syncedRows": true]) + answer(socket, sent, "analyze-query", result) + _ = try await unnamed.value + + let server = Task { + try await zero.inspector.analyzeServerAST( + AST(table: "custom"), options: AnalyzeQueryOptions(vendedRows: false, joinPlans: true)) + } + await settle() + #expect( + withoutID(requests(socket).last) == [ + "op": "analyze-query", "ast": ["table": "custom"], + "options": ["vendedRows": false, "joinPlans": true], + ]) + answer(socket, requests(socket).last, "analyze-query", result) + _ = try await server.value + + let byName = Task { + try await zero.inspector.analyzeNamedQuery(name: "issues.byOwner", args: [["ownerId": "u"]]) + } + await settle() + #expect( + withoutID(requests(socket).last) == [ + "op": "analyze-query", "name": "issues.byOwner", "args": [["ownerId": "u"]], + ]) + answer(socket, requests(socket).last, "analyze-query", result) + _ = try await byName.value + } + + // MARK: - Formatting + + @Test(arguments: [ + ( + #"{"table":"issue","where":{"type":"and","conditions":[{"type":"simple","op":"=","left":{"type":"column","name":"id"},"right":{"type":"literal","value":"it's \"x\" \\ y"}},{"type":"simple","op":">","left":{"type":"column","name":"createdAt"},"right":{"type":"literal","value":5}}]},"orderBy":[["id","asc"]],"limit":10}"#, + #"issue.where('id', 'it\'s "x" \\ y').where('createdAt', '>', 5).orderBy('id', 'asc').limit(10)"# + ), + ( + #"{"table":"issue","where":{"type":"or","conditions":[{"type":"simple","op":"=","left":{"type":"column","name":"closed"},"right":{"type":"literal","value":true}},{"type":"correlatedSubquery","op":"EXISTS","related":{"correlation":{"parentField":["id"],"childField":["issueId"]},"subquery":{"table":"comment","alias":"zsubq_comments","where":{"type":"simple","op":"LIKE","left":{"type":"column","name":"body"},"right":{"type":"literal","value":"%a%"}}}}},{"type":"correlatedSubquery","op":"NOT EXISTS","related":{"correlation":{"parentField":["id"],"childField":["issueId"]},"subquery":{"table":"label","alias":"zsubq_labels"}},"flip":true},{"type":"and","conditions":[{"type":"simple","op":"IS","left":{"type":"column","name":"ownerId"},"right":{"type":"literal","value":null}},{"type":"simple","op":"!=","left":{"type":"column","name":"title"},"right":{"type":"static","anchor":"authData","field":"sub"}}]}]}}"#, + #"issue.where(({and, cmp, exists, not, or}) => or(cmp('closed', true), exists('comments', q => q.where('body', 'LIKE', '%a%')), not(exists('labels', {flip: true})), and(cmp('ownerId', 'IS', null), cmp('title', '!=', authParam('sub')))))"# + ), + ( + #"{"table":"issue","related":[{"correlation":{"parentField":["id"],"childField":["issueId"]},"hidden":true,"subquery":{"table":"issueLabel","alias":"labels","related":[{"correlation":{"parentField":["labelId"],"childField":["id"]},"subquery":{"table":"label","alias":"labels","orderBy":[["name","desc"]]}}]}},{"correlation":{"parentField":["id"],"childField":["issueId"]},"subquery":{"table":"comment","alias":"comments","limit":3}}],"where":{"type":"and","conditions":[{"type":"simple","op":"IN","left":{"type":"column","name":"id"},"right":{"type":"literal","value":["a","b"]}},{"type":"simple","op":"=","left":{"type":"json","value":{"type":"column","name":"meta"},"path":["a",0]},"right":{"type":"literal","value":"q"}},{"type":"simple","op":"=","left":{"type":"column","name":"ownerId"},"right":{"type":"static","anchor":"authData","field":["org","id"]}}]},"start":{"row":{"id":"x"},"exclusive":false}}"#, + #"issue.where('id', 'IN', ["a","b"]).where(({cmp, json}) => cmp(json('meta', 'a', 0), 'q')).where('ownerId', authParam(['org', 'id'])).related('labels', q => q.orderBy('name', 'desc')).related('comments', q => q.limit(3)).start({"id":"x"}, { inclusive: true })"# + ), + ( + #"{"table":"issue","groupBy":["ownerId"],"aggregate":{"fn":"count"},"orderBy":[["ownerId","asc"]]}"#, + #"issue.groupBy('ownerId').count().orderBy('ownerId', 'asc')"# + ), + ( + #"{"table":"issue","aggregate":{"fn":"count","distinct":["ownerId","closed"]}}"#, + #"issue.distinct('ownerId', 'closed').count()"# + ), + ( + #"{"table":"issue","groupBy":["ownerId"],"aggregate":{"fn":"sum","field":"meta","path":["n",1]},"start":{"row":{"ownerId":"u"},"exclusive":true}}"#, + #"issue.groupBy('ownerId').sum(({json}) => json('meta', 'n', 1)).start({"ownerId":"u"})"# + ), + ]) + func writesTheQueryAsJSsAstToZQLDoes(json: String, zql: String) throws { + let ast = try AST(json: JSONParser.parse(Array(json.utf8))) + #expect(ast.table + astToZQL(ast) == zql) + } + + @Test func formatsPlannerEventsAsJSDoes() throws { + let events = try JSONParser.parse( + Array( + #"[{"type":"attempt-start","attemptNumber":0,"totalAttempts":5},{"type":"node-cost","attemptNumber":0,"nodeType":"connection","node":"issue","branchPattern":[0,1],"downstreamChildSelectivity":0.5,"costEstimate":{"startupCost":0,"scanEst":100,"cost":101.005,"returnedRows":33.3333333,"selectivity":0.123456789,"limit":10},"filters":{"type":"and","conditions":[{"type":"simple","op":"=","left":{"type":"column","name":"id"},"right":{"type":"literal","value":"x"}},{"type":"simple","op":">","left":{"type":"json","value":{"type":"column","name":"meta"},"path":["a",0]},"right":{"type":"literal","value":3}},{"type":"or","conditions":[{"type":"simple","op":"=","left":{"type":"column","name":"o"},"right":{"type":"static","anchor":"authData","field":["a","b"]}},{"type":"correlatedSubquery","op":"EXISTS","related":{"correlation":{"parentField":["id"],"childField":["issueId"]},"subquery":{"table":"comment"}}}]}]},"ordering":[["id","asc"],["createdAt","desc"]]},{"type":"node-constraint","attemptNumber":0,"nodeType":"connection","node":"issue","branchPattern":[0,1],"constraint":{"id":null,"ownerId":null},"from":"join1"},{"type":"node-cost","attemptNumber":0,"nodeType":"connection","node":"comment","branchPattern":[],"downstreamChildSelectivity":1,"costEstimate":{"startupCost":0,"scanEst":100,"cost":2.675,"returnedRows":33.3333333,"selectivity":1e-9}},{"type":"node-cost","attemptNumber":0,"nodeType":"join","node":"join1","branchPattern":[],"downstreamChildSelectivity":1.5e-8,"costEstimate":{"startupCost":0,"scanEst":100,"cost":1.005,"returnedRows":33.3333333,"selectivity":0.123456789},"joinType":"semi"},{"type":"node-cost","attemptNumber":0,"nodeType":"join","node":"join2","branchPattern":[],"downstreamChildSelectivity":1,"costEstimate":{"startupCost":0,"scanEst":100,"cost":101.005,"returnedRows":33.3333333,"selectivity":0.123456789}},{"type":"plan-complete","attemptNumber":0,"totalCost":2.5,"flipPattern":0,"joinStates":[{"join":"join1","type":"semi"}]},{"type":"attempt-start","attemptNumber":1,"totalAttempts":5},{"type":"plan-failed","attemptNumber":1,"reason":"no plan"},{"type":"best-plan-selected","bestAttemptNumber":0,"totalCost":2.5,"flipPattern":0,"joinStates":[{"join":"join1","type":"semi"},{"join":"join2","type":"flipped"}]}]"# + .utf8)) + let rule = String(repeating: "─", count: 60) + let expected = """ + [Attempt 1/5] Pattern 0 (000) + Connections: + issue: + cost=101.00, startup=0.00, scan=100.00 + rows=33.33, selectivity=0.12345679, limit=10 + downstreamChildSelectivity=0.50000000 + constraints={id, ownerId} + filters=(id = 'x' AND meta.a[0] > 3 AND (o = @authData.a.b OR EXISTS(comment))) + ordering=id asc, createdAt desc + comment: + cost=2.67, startup=0.00, scan=100.00 + rows=33.33, selectivity=0.00000000, limit=none + downstreamChildSelectivity=1.00000000 + constraints={} + filters=none + ordering=none + Joins: + join1 (semi): + cost=1.00, startup=0.00, scan=100.00 + rows=33.33, selectivity=0.12345679, limit=none + downstreamChildSelectivity=0.00000001 + join2: + cost=101.00, startup=0.00, scan=100.00 + rows=33.33, selectivity=0.12345679, limit=none + downstreamChildSelectivity=1.00000000 + ✓ Plan complete: total cost = 2.50 + + [Attempt 2/5] Pattern 1 (001) + ✗ Plan failed: no plan + + \(rule) + ✓ Best plan: Attempt 1 (cost=2.50) + Join types: + join1: semi + join2: flipped + \(rule) + """ + let zero = try makeZero() + #expect(zero.inspector.formatPlannerEvents(events.arrayValue ?? []) == expected) + } + + @Test(arguments: [ + (1.005, "1", "1.00", "1.00500000"), + (2.5, "3", "2.50", "2.50000000"), + (-0.001, "-0", "-0.00", "-0.00100000"), + (0.5, "1", "0.50", "0.50000000"), + (1e-7, "0", "0.00", "0.00000010"), + (123_456_789.123_456_79, "123456789", "123456789.12", "123456789.12345679"), + (0, "0", "0.00", "0.00000000"), + (-0.0, "0", "0.00", "0.00000000"), + (1e21, "1e+21", "1e+21", "1e+21"), + (Double.leastNonzeroMagnitude, "0", "0.00", "0.00000000"), + (0.125, "0", "0.13", "0.12500000"), + (1.45, "1", "1.45", "1.45000000"), + (8.345, "8", "8.35", "8.34500000"), + (99.995, "100", "100.00", "99.99500000"), + (-2.5, "-3", "-2.50", "-2.50000000"), + ( + 100_000_000_000_000_065_536, "100000000000000065536", "100000000000000065536.00", + "100000000000000065536.00000000" + ), + (1.5e-8, "0", "0.00", "0.00000001"), + ]) + func roundsAsJSsToFixedDoes(value: Double, zero: String, two: String, eight: String) { + #expect(toFixed(value, 0) == zero) + #expect(toFixed(value, 2) == two) + #expect(toFixed(value, 8) == eight) + } +} diff --git a/swift/Tests/ZeroTests/TDigestTests.swift b/swift/Tests/ZeroTests/TDigestTests.swift new file mode 100644 index 0000000000..5b43c43165 --- /dev/null +++ b/swift/Tests/ZeroTests/TDigestTests.swift @@ -0,0 +1,105 @@ +import Testing + +@testable import Zero + +/// The TDigest port against JS's: every expected number is what +/// `packages/shared/src/tdigest.ts` returns for the same calls, run with +/// the repo's Node. +@Suite struct TDigestTests { + @Test func answersAsJSDoesOnceItHasMergedItsValues() { + // Enough values to merge the buffer, some with update metrics' weight. + let d = TDigest() + for i in 0..<1000 { d.add(Double((i * 7) % 101), weight: i % 3 == 0 ? 32 : 1) } + #expect( + [0, 0.25, 0.5, 0.95, 0.99, 1].map(d.quantile) == [ + 0, 24.38504613033506, 49.76361353903839, 95.57283979178716, 100, 100, + ]) + #expect([10, 50.5, 100].map(d.cdf) == [0.10489695261581822, 0.5090373478632793, 1]) + #expect(d.count() == 11354) + let json = d.toJSON() + #expect(json.count == 247) + #expect(Array(json.prefix(7)) == [100, 0, 1, 0, 32, 0, 32]) + } + + @Test func mergesWhenReadAsJSDoes() { + // Reading merges the buffer, which changes the centroids later values + // land in, as in JS. + let a = TDigest() + let b = TDigest() + for i in 0..<900 { + a.add(Double(900 - i)) + b.add(Double(900 - i)) + if i == 450 { _ = b.quantile(0.5) } + } + #expect([a.quantile(0.9), a.quantile(0.1)] == [810.5, 90.5]) + #expect([b.quantile(0.9), b.quantile(0.1)] == [810.5, 90.5]) + #expect(a.toJSON().count == 237) + #expect(b.toJSON().count == 255) + } + + @Test func keepsFewerCentroidsWithALowerCompression() { + let d = TDigest(compression: 20) + for i in 0..<500 { d.add(Double(i)) } + #expect( + d.toJSON() == [ + 20, 0, 1, 3.5, 6, 9.5, 6, 16, 7, 24, 9, 33.5, 10, 50, 23, 73.5, 24, 91.5, 12, 109, 23, + 130, 19, 152, 25, 177, 25, 201.5, 24, 225, 23, 247, 21, 274.5, 34, 309.5, 36, 344.5, 34, + 377, 31, 406, 27, 431, 23, 451.5, 18, 467, 13, 480.5, 14, 492, 9, 498, 3, + ]) + } + + @Test func readsAndWritesJSsJSON() throws { + let d = try TDigest(json: [100, 1, 1, 5, 2, 9, 3]) + #expect([d.quantile(0.5), d.quantile(0.95)] == [6.6, 9]) + #expect(d.cdf(5) == 0.3333333333333333) + #expect(d.count() == 6) + #expect(d.toJSON() == [100, 1, 1, 5, 2, 9, 3]) + let server = try TDigest(json: [100, 2, 3, 4, 1, 10, 2]) + #expect([server.quantile(0.5), server.quantile(0.95)] == [3.5, 10]) + // An even count isn't a digest's (`Invalid centroids array`). + #expect(throws: TDigest.InvalidJSON.self) { try TDigest(json: [100, 1]) } + #expect(throws: TDigest.InvalidJSON.self) { try TDigest(json: []) } + } + + @Test func ignoresWhatJSIgnores() { + let d = TDigest() + d.add(.nan) + d.add(1, weight: 0) + d.add(1, weight: -1) + d.add(1, weight: .infinity) + #expect(d.count() == 0) + #expect(d.quantile(0.5).isNaN) + d.add(3) + #expect(d.quantile(0.5) == 3) + // One update sampled out of `UPDATE_SAMPLE_RATE`. + let sampled = TDigest() + sampled.add(0, weight: 32) + #expect(sampled.count() == 32) + #expect(sampled.quantile(0.5) == 0) + } + + @Test func mergesItself() { + let d = TDigest() + for i in 0..<10 { d.add(Double(i * i)) } + d.merge(d) + #expect(d.count() == 20) + #expect( + d.toJSON() == [ + 100, 0, 1, 0, 1, 1, 1, 1, 1, 4, 1, 4, 1, 9, 1, 9, 1, 16, 1, 16, 1, 25, 1, 25, 1, 36, 1, 36, + 1, 49, 1, 49, 1, 64, 1, 64, 1, 81, 1, 81, 1, + ]) + #expect(d.quantile(0.75) == 49) + } + + /// `Math.asin` and `Math.sin` as Node 22 computes them, on arguments + /// where glibc's last bit differs. + @Test func computesNodesTrigonometry() { + #expect(fdlibmAsin(-0.8461538461538461) == -1.0087265237892695) + #expect(fdlibmAsin(-0.5294117647058824) == -0.5579070399678958) + #expect(fdlibmAsin(-0.8888888888888888) == -1.0949140771344799) + #expect(fdlibmSin(0.7960981043985682) == 0.7146321601621326) + #expect(fdlibmSin(0.8546279036617737) == 0.7543266873983547) + #expect(fdlibmSin(-1.0340198899748416) == -0.8593615719698496) + #expect(fdlibmSin(1) == 0.8414709848078965) + } +}