From b234cfb7d727785eb4e373ef8ec2557dc15f25ba Mon Sep 17 00:00:00 2001 From: Oleg Lazari Date: Tue, 21 Oct 2025 22:53:39 -0400 Subject: [PATCH 1/5] fixed issue with intersting corpus not updating --- Sources/Fuzzilli/Fuzzer.swift | 6 ++-- .../FuzzilliTests/PostgreSQLCorpusTests.swift | 32 +++++++++++++++++++ 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/Sources/Fuzzilli/Fuzzer.swift b/Sources/Fuzzilli/Fuzzer.swift index e2a2369f6..ecebb5890 100755 --- a/Sources/Fuzzilli/Fuzzer.swift +++ b/Sources/Fuzzilli/Fuzzer.swift @@ -723,9 +723,9 @@ public class Fuzzer { aspects = intersection } while !didConverge || attempt < minAttempts } - if origin == .local { - iterationOfLastInteratingSample = iterations - } + // Update the iteration counter for any interesting program found + // This is crucial for corpus generation phase to properly track progress + iterationOfLastInteratingSample = iterations // Determine whether the program needs to be minimized, then, using this helper function, dispatch the appropriate // event and insert the sample into the corpus. diff --git a/Tests/FuzzilliTests/PostgreSQLCorpusTests.swift b/Tests/FuzzilliTests/PostgreSQLCorpusTests.swift index f8d5ec36b..ff4242c8a 100644 --- a/Tests/FuzzilliTests/PostgreSQLCorpusTests.swift +++ b/Tests/FuzzilliTests/PostgreSQLCorpusTests.swift @@ -169,4 +169,36 @@ final class PostgreSQLCorpusTests: XCTestCase { XCTAssertTrue(description.contains("Coverage: 75.50%")) XCTAssertTrue(description.contains("Pending Sync: 3")) } + + func testPostgreSQLCorpusInterestingProgramTracking() { + let databasePool = DatabasePool(connectionString: "postgresql://localhost:5432/fuzzilli") + let corpus = PostgreSQLCorpus( + minSize: 1, + maxSize: 10, + minMutationsPerSample: 5, + databasePool: databasePool, + fuzzerInstanceId: "test-instance-1" + ) + + // Create a mock fuzzer to initialize the corpus + let mockFuzzer = makeMockFuzzer(corpus: corpus) + + // Create a simple program with actual content + let b = mockFuzzer.makeBuilder() + b.loadInt(42) + let program = b.finalize() + + // Add the program to the corpus + // This should trigger the InterestingProgramFound event + corpus.add(program, ProgramAspects(outcome: .succeeded)) + + // Verify the program was added + XCTAssertEqual(corpus.size, 1) + XCTAssertFalse(corpus.isEmpty) + + // Test that we can get the program back + let allPrograms = corpus.allPrograms() + XCTAssertEqual(allPrograms.count, 1) + XCTAssertEqual(allPrograms[0], program) + } } From a75931784c7884aae0a94cb02f0503480ad9c8a7 Mon Sep 17 00:00:00 2001 From: Oleg Lazari Date: Tue, 21 Oct 2025 23:58:36 -0400 Subject: [PATCH 2/5] Added crashes + execution count --- .../Fuzzilli/Corpus/PostgreSQLCorpus.swift | 100 +++++++++++++++++- Sources/Fuzzilli/Database/DatabaseUtils.swift | 39 +++++-- .../Fuzzilli/Database/PostgreSQLStorage.swift | 96 ++++++++++++++--- postgres-init.sql | 7 +- 4 files changed, 217 insertions(+), 25 deletions(-) diff --git a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift index 82b9c498b..f6ba7d399 100644 --- a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift +++ b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift @@ -143,8 +143,19 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { dbExecutionPurpose = .other } - // Add to batch instead of storing immediately - self.addToExecutionBatch(program, aspects, executionType: dbExecutionPurpose) + // Cache execution data immediately before REPRL context becomes invalid + let executionData = ExecutionData( + outcome: execution.outcome, + execTime: execution.execTime, + stdout: execution.stdout, + stderr: execution.stderr, + fuzzout: execution.fuzzout + ) + + // Store execution data with cached metadata + Task { + await self.storeExecutionWithCachedData(program, executionData, dbExecutionPurpose, aspects) + } } } @@ -507,6 +518,91 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { throw lastError ?? DatabasePoolError.initializationFailed("Failed to register fuzzer after \(maxRetries) attempts") } + /// Cached execution data to avoid REPRL context issues + private struct ExecutionData { + let outcome: ExecutionOutcome + let execTime: TimeInterval + let stdout: String + let stderr: String + let fuzzout: String + } + + /// Store execution with cached data to avoid REPRL context issues + private func storeExecutionWithCachedData(_ program: Program, _ executionData: ExecutionData, _ executionType: DatabaseExecutionPurpose, _ aspects: ProgramAspects) async { + do { + // Use the registered fuzzer ID + guard let fuzzerId = fuzzerId else { + logger.error("Cannot store execution: fuzzer not registered") + return + } + + // Store the program in the program table + let programHash = try await storage.storeProgram( + program: program, + fuzzerId: fuzzerId, + metadata: ExecutionMetadata(lastOutcome: DatabaseExecutionOutcome( + id: DatabaseUtils.mapExecutionOutcome(outcome: aspects.outcome), + outcome: aspects.outcome.description, + description: aspects.outcome.description + )) + ) + + // Store the execution record with cached execution metadata + let executionId = try await storage.storeExecution( + program: program, + fuzzerId: fuzzerId, + executionType: executionType, + outcome: executionData.outcome, + coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0, + executionTimeMs: Int(executionData.execTime * 1000), // Convert to milliseconds + stdout: executionData.stdout, + stderr: executionData.stderr, + fuzzout: executionData.fuzzout + ) + + logger.info("Stored execution with cached data: programHash=\(programHash), executionId=\(executionId), execTime=\(executionData.execTime), outcome=\(executionData.outcome)") + + } catch { + logger.error("Failed to store execution with cached data: \(error)") + } + } + + /// Store execution with full metadata from Execution object + private func storeExecutionWithMetadata(_ program: Program, _ execution: Execution, _ executionType: DatabaseExecutionPurpose, _ aspects: ProgramAspects) async { + do { + // Use the registered fuzzer ID + guard let fuzzerId = fuzzerId else { + logger.error("Cannot store execution: fuzzer not registered") + return + } + + // Store the program in the program table + let programHash = try await storage.storeProgram( + program: program, + fuzzerId: fuzzerId, + metadata: ExecutionMetadata(lastOutcome: DatabaseExecutionOutcome( + id: DatabaseUtils.mapExecutionOutcome(outcome: aspects.outcome), + outcome: aspects.outcome.description, + description: aspects.outcome.description + )) + ) + + // Store the execution record with full execution metadata + let executionId = try await storage.storeExecution( + program: program, + fuzzerId: fuzzerId, + execution: execution, + executionType: executionType, + coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0 + ) + + logger.info("Stored execution with metadata: programHash=\(programHash), executionId=\(executionId), execTime=\(execution.execTime), outcome=\(execution.outcome)") + + } catch { + logger.error("Failed to store execution with metadata: \(error)") + } + } + /// Store a program execution in the database private func storeExecutionInDatabase(_ program: Program, _ aspects: ProgramAspects, executionType: DatabaseExecutionPurpose, mutatorType: String?) async { do { diff --git a/Sources/Fuzzilli/Database/DatabaseUtils.swift b/Sources/Fuzzilli/Database/DatabaseUtils.swift index 1ae89dd4e..79d30007a 100644 --- a/Sources/Fuzzilli/Database/DatabaseUtils.swift +++ b/Sources/Fuzzilli/Database/DatabaseUtils.swift @@ -100,13 +100,34 @@ public class DatabaseUtils { public static func mapExecutionOutcome(outcome: ExecutionOutcome) -> Int { switch outcome { case .succeeded: - return 1 + return 3 // Succeeded maps to ID 3 case .failed: - return 2 + return 2 // Failed maps to ID 2 case .crashed: - return 3 + return 1 // Crashed maps to ID 1 case .timedOut: - return 4 + return 4 // TimedOut maps to ID 4 + } + } + + /// Map ExecutionOutcome with signal code to database ID + /// Signal 11 (SIGSEGV) and 7 (SIGBUS) are real crashes + /// Signal 5 (SIGTRAP) and 6 (SIGABRT) are sig checks + public static func mapExecutionOutcomeWithSignal(outcome: ExecutionOutcome, signalCode: Int?) -> Int { + switch outcome { + case .succeeded: + return 3 // Succeeded maps to ID 3 + case .failed: + return 2 // Failed maps to ID 2 + case .crashed(let signal): + // Check if this is a real crash or just a signal check + if signal == 11 || signal == 7 { + return 1 // Real crashes: SIGSEGV (11) and SIGBUS (7) + } else { + return 34 // SigCheck: SIGTRAP (5), SIGABRT (6), and others + } + case .timedOut: + return 4 // TimedOut maps to ID 4 } } @@ -114,13 +135,15 @@ public class DatabaseUtils { public static func mapExecutionOutcomeFromId(id: Int) -> ExecutionOutcome { switch id { case 1: - return .succeeded + return .crashed(1) // ID 1 = Crashed (real crashes) case 2: - return .failed(1) // Default exit code + return .failed(1) // ID 2 = Failed case 3: - return .crashed(1) // Default signal + return .succeeded // ID 3 = Succeeded case 4: - return .timedOut + return .timedOut // ID 4 = TimedOut + case 34: + return .crashed(5) // ID 34 = SigCheck (signal checks) default: return .succeeded // Default fallback } diff --git a/Sources/Fuzzilli/Database/PostgreSQLStorage.swift b/Sources/Fuzzilli/Database/PostgreSQLStorage.swift index f5ae14372..1216cde9b 100644 --- a/Sources/Fuzzilli/Database/PostgreSQLStorage.swift +++ b/Sources/Fuzzilli/Database/PostgreSQLStorage.swift @@ -154,12 +154,23 @@ public class PostgreSQLStorage { """ try await connection.query(fuzzerQuery, logger: self.logger) + // Generate JavaScript code from the program + let lifter = JavaScriptLifter(ecmaVersion: .es6) + let javascriptCode = lifter.lift(program, withOptions: []) + // Insert into program table (executed programs) - let programQuery: PostgresQuery = """ - INSERT INTO program (program_base64, fuzzer_id, program_size, program_hash) - VALUES (\(programBase64), \(fuzzerId), \(program.size), \(programHash)) - ON CONFLICT (program_base64) DO NOTHING - """ + // Base64 encode the JavaScript code to avoid SQL injection issues + let javascriptCodeBase64 = Data(javascriptCode.utf8).base64EncodedString() + + // Use string concatenation to avoid parameter substitution issues + let programQueryString = "INSERT INTO program (program_base64, fuzzer_id, program_size, program_hash, javascript_code) VALUES ('" + + programBase64 + "', " + + String(fuzzerId) + ", " + + String(program.size) + ", '" + + programHash + "', '" + + javascriptCodeBase64 + "') ON CONFLICT (program_base64) DO NOTHING" + + let programQuery = PostgresQuery(stringLiteral: programQueryString) try await connection.query(programQuery, logger: self.logger) self.logger.info("Program storage successful: hash=\(programHash)") @@ -198,7 +209,10 @@ public class PostgreSQLStorage { coverage: Double = 0.0, executionTimeMs: Int = 0, feedbackVector: Data? = nil, - coverageEdges: Set = [] + coverageEdges: Set = [], + stdout: String? = nil, + stderr: String? = nil, + fuzzout: String? = nil ) async throws -> Int { let programHash = DatabaseUtils.calculateProgramHash(program: program) let programBase64 = DatabaseUtils.encodeProgramToBase64(program: program) @@ -225,21 +239,39 @@ public class PostgreSQLStorage { defer { Task { _ = try? await connection.close() } } let executionTypeId = DatabaseUtils.mapExecutionType(purpose: executionType) - let outcomeId = DatabaseUtils.mapExecutionOutcome(outcome: outcome) + let mutatorTypeId = mutatorType != nil ? DatabaseUtils.mapMutatorType(mutator: mutatorType!) : nil - let query: PostgresQuery = """ + // Extract execution metadata from ExecutionOutcome + let (signalCode, exitCode) = extractExecutionMetadata(from: outcome) + + // Use signal-aware mapping for execution outcomes + let outcomeId = DatabaseUtils.mapExecutionOutcomeWithSignal(outcome: outcome, signalCode: signalCode) + + let mutatorTypeValue = mutatorTypeId != nil ? "\(mutatorTypeId!)" : "NULL" + let feedbackVectorValue = feedbackVector != nil ? "'\(feedbackVector!.base64EncodedString())'" : "NULL" + let signalCodeValue = signalCode != nil ? "\(signalCode!)" : "NULL" + let exitCodeValue = exitCode != nil ? "\(exitCode!)" : "NULL" + let stdoutValue = stdout != nil ? "'\(stdout!.replacingOccurrences(of: "'", with: "''"))'" : "NULL" + let stderrValue = stderr != nil ? "'\(stderr!.replacingOccurrences(of: "'", with: "''"))'" : "NULL" + let fuzzoutValue = fuzzout != nil ? "'\(fuzzout!.replacingOccurrences(of: "'", with: "''"))'" : "NULL" + + let queryString = """ INSERT INTO execution ( program_base64, execution_type_id, mutator_type_id, execution_outcome_id, coverage_total, execution_time_ms, + signal_code, exit_code, stdout, stderr, fuzzout, feedback_vector, created_at ) VALUES ( - \(programBase64), \(executionTypeId), - \(mutatorType ?? "NULL"), \(outcomeId), \(coverage), - \(executionTimeMs), \(feedbackVector?.base64EncodedString() ?? "NULL"), - NOW() + '\(programBase64)', \(executionTypeId), + \(mutatorTypeValue), \(outcomeId), \(coverage), + \(executionTimeMs), \(signalCodeValue), \(exitCodeValue), + \(stdoutValue), \(stderrValue), \(fuzzoutValue), + \(feedbackVectorValue), NOW() ) RETURNING execution_id """ + let query = PostgresQuery(stringLiteral: queryString) + let result = try await connection.query(query, logger: self.logger) let rows = try await result.collect() guard let row = rows.first else { @@ -251,6 +283,46 @@ public class PostgreSQLStorage { return executionId } + /// Store execution record from Execution object + public func storeExecution( + program: Program, + fuzzerId: Int, + execution: Execution, + executionType: DatabaseExecutionPurpose, + mutatorType: String? = nil, + coverage: Double = 0.0, + feedbackVector: Data? = nil, + coverageEdges: Set = [] + ) async throws -> Int { + let executionTimeMs = Int(execution.execTime * 1000) // Convert to milliseconds + return try await storeExecution( + program: program, + fuzzerId: fuzzerId, + executionType: executionType, + mutatorType: mutatorType, + outcome: execution.outcome, + coverage: coverage, + executionTimeMs: executionTimeMs, + feedbackVector: feedbackVector, + coverageEdges: coverageEdges, + stdout: execution.stdout, + stderr: execution.stderr, + fuzzout: execution.fuzzout + ) + } + + /// Extract execution metadata from ExecutionOutcome + private func extractExecutionMetadata(from outcome: ExecutionOutcome) -> (signalCode: Int?, exitCode: Int?) { + switch outcome { + case .crashed(let signal): + return (signalCode: signal, exitCode: nil) + case .failed(let exitCode): + return (signalCode: nil, exitCode: exitCode) + case .succeeded, .timedOut: + return (signalCode: nil, exitCode: nil) + } + } + /// Get execution history for a program public func getExecutionHistory(programHash: String, fuzzerId: Int, limit: Int = 100) async throws -> [ExecutionRecord] { logger.info("Getting execution history: hash=\(programHash), fuzzerId=\(fuzzerId), limit=\(limit)") diff --git a/postgres-init.sql b/postgres-init.sql index 73d1ea437..e6bad8c59 100644 --- a/postgres-init.sql +++ b/postgres-init.sql @@ -79,10 +79,11 @@ CREATE TABLE IF NOT EXISTS execution_outcome ( -- Preseed execution outcomes INSERT INTO execution_outcome (outcome, description) VALUES - ('Crashed', 'Program crashed with a signal'), + ('Crashed', 'Program crashed with a signal (SIGSEGV/SIGBUS)'), ('Failed', 'Program failed with an exit code'), ('Succeeded', 'Program executed successfully'), - ('TimedOut', 'Program execution timed out') + ('TimedOut', 'Program execution timed out'), + ('SigCheck', 'Program terminated with signal (SIGTRAP/SIGABRT)') ON CONFLICT (outcome) DO NOTHING; -- Create the main execution table @@ -195,7 +196,7 @@ SELECT FROM execution e JOIN execution_outcome eo ON e.execution_outcome_id = eo.id LEFT JOIN crash_analysis ca ON e.execution_id = ca.execution_id -WHERE eo.outcome IN ('Crashed', 'Failed'); +WHERE eo.outcome IN ('Crashed', 'Failed', 'SigCheck'); -- Create function to get coverage statistics CREATE OR REPLACE FUNCTION get_coverage_stats(fuzzer_instance_id INTEGER) From 91e31438af95da66d91150e1624934b00f608381 Mon Sep 17 00:00:00 2001 From: Oleg Lazari Date: Wed, 22 Oct 2025 00:26:58 -0400 Subject: [PATCH 3/5] added resuming --- .../Fuzzilli/Corpus/PostgreSQLCorpus.swift | 68 ++++++++- Sources/Fuzzilli/Database/DatabaseUtils.swift | 18 +++ .../Fuzzilli/Database/PostgreSQLStorage.swift | 141 +++++++++++++++++- Sources/FuzzilliCli/main.swift | 3 +- 4 files changed, 214 insertions(+), 16 deletions(-) diff --git a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift index f6ba7d399..58e0e9c24 100644 --- a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift +++ b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift @@ -25,6 +25,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { private let databasePool: DatabasePool private let fuzzerInstanceId: String private let storage: PostgreSQLStorage + private let resume: Bool // MARK: - In-Memory Cache @@ -65,7 +66,8 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { minMutationsPerSample: Int, databasePool: DatabasePool, fuzzerInstanceId: String, - syncInterval: TimeInterval = 60.0 // Default 1 minute sync interval + syncInterval: TimeInterval = 60.0, // Default 1 minute sync interval + resume: Bool = true // Default to resume from previous state ) { // The corpus must never be empty assert(minSize >= 1) @@ -77,6 +79,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { self.databasePool = databasePool self.fuzzerInstanceId = fuzzerInstanceId self.syncInterval = syncInterval + self.resume = resume self.storage = PostgreSQLStorage(databasePool: databasePool) self.programs = RingBuffer(maxSize: maxSize) @@ -176,9 +179,11 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { fuzzer.timers.scheduleTask(every: 30 * Minutes, cleanup) } - // Load initial corpus from database - Task { - await loadInitialCorpus() + // Load initial corpus from database if resume is enabled + if resume { + Task { + await loadInitialCorpus() + } } } @@ -423,9 +428,55 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { private func loadInitialCorpus() async { logger.info("Loading initial corpus from PostgreSQL...") - // This would be implemented when we have actual database operations - // For now, just log that we would load from database - logger.info("Initial corpus loading would be implemented here") + guard let fuzzerId = fuzzerId else { + logger.warning("Cannot load initial corpus: fuzzer not registered") + return + } + + do { + // Load programs from the last 24 hours to resume recent work + let since = Date().addingTimeInterval(-24 * 60 * 60) // 24 hours ago + let recentPrograms = try await storage.getRecentPrograms( + fuzzerId: fuzzerId, + since: since, + limit: maxSize + ) + + logger.info("Found \(recentPrograms.count) recent programs to resume") + + // Add programs to the corpus + cacheLock.lock() + defer { cacheLock.unlock() } + + for (program, metadata) in recentPrograms { + let programHash = DatabaseUtils.calculateProgramHash(program: program) + + // Skip if already in cache + if programCache[programHash] != nil { + continue + } + + // Add to in-memory structures + prepareProgramForInclusion(program, index: totalEntryCounter) + programs.append(program) + ages.append(0) + programHashes.append(programHash) + programCache[programHash] = (program: program, metadata: metadata) + + totalEntryCounter += 1 + } + + logger.info("Resumed PostgreSQL corpus with \(programs.count) programs") + + // If we have no programs, we need at least one to avoid empty corpus + if programs.count == 0 { + logger.info("No programs found to resume, corpus will start empty") + } + + } catch { + logger.error("Failed to load initial corpus from PostgreSQL: \(error)") + logger.info("Corpus will start empty and build up from scratch") + } } /// Synchronize with PostgreSQL database @@ -489,7 +540,8 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { /// Register fuzzer with retry logic private func registerFuzzerWithRetry() async throws -> Int { - let fuzzerName = "fuzzer-\(fuzzerInstanceId)" + // Use the fuzzerInstanceId directly as the name to avoid double "fuzzer-" prefix + let fuzzerName = fuzzerInstanceId let engineType = "v8" // This could be made configurable let maxRetries = 3 diff --git a/Sources/Fuzzilli/Database/DatabaseUtils.swift b/Sources/Fuzzilli/Database/DatabaseUtils.swift index 79d30007a..3c18d2dfc 100644 --- a/Sources/Fuzzilli/Database/DatabaseUtils.swift +++ b/Sources/Fuzzilli/Database/DatabaseUtils.swift @@ -378,6 +378,24 @@ public enum DatabaseUtilsError: Error, LocalizedError { return "Invalid metadata format" } } + + /// Map execution outcome string to database ID + public static func mapExecutionOutcomeFromString(_ outcome: String) -> Int { + switch outcome.lowercased() { + case "crashed": + return 1 + case "failed": + return 2 + case "succeeded": + return 3 + case "timedout": + return 4 + case "sigcheck": + return 34 + default: + return 3 // Default to succeeded + } + } } // MARK: - Extensions diff --git a/Sources/Fuzzilli/Database/PostgreSQLStorage.swift b/Sources/Fuzzilli/Database/PostgreSQLStorage.swift index 1216cde9b..ff64a3859 100644 --- a/Sources/Fuzzilli/Database/PostgreSQLStorage.swift +++ b/Sources/Fuzzilli/Database/PostgreSQLStorage.swift @@ -51,20 +51,42 @@ public class PostgreSQLStorage { ) defer { Task { _ = try? await connection.close() } } - let query: PostgresQuery = """ + // First, check if a fuzzer with this name already exists + let checkQuery: PostgresQuery = "SELECT fuzzer_id, status FROM main WHERE fuzzer_name = \(name)" + let checkResult = try await connection.query(checkQuery, logger: self.logger) + let checkRows = try await checkResult.collect() + + if let existingRow = checkRows.first { + let existingFuzzerId = try existingRow.decode(Int.self, context: .default) + let existingStatus = try existingRow.decode(String.self, context: .default) + + // Update status to active if it was inactive + if existingStatus != "active" { + let updateQuery: PostgresQuery = "UPDATE main SET status = 'active' WHERE fuzzer_id = \(existingFuzzerId)" + try await connection.query(updateQuery, logger: self.logger) + logger.info("Reactivated existing fuzzer: fuzzerId=\(existingFuzzerId)") + } else { + logger.info("Reusing existing active fuzzer: fuzzerId=\(existingFuzzerId)") + } + + return existingFuzzerId + } + + // If no existing fuzzer found, create a new one + let insertQuery: PostgresQuery = """ INSERT INTO main (fuzzer_name, engine_type, status) VALUES (\(name), \(engineType), 'active') RETURNING fuzzer_id """ - let result = try await connection.query(query, logger: self.logger) + let result = try await connection.query(insertQuery, logger: self.logger) let rows = try await result.collect() guard let row = rows.first else { throw PostgreSQLStorageError.noResult } let fuzzerId = try row.decode(Int.self, context: .default) - self.logger.info("Fuzzer registration successful: fuzzerId=\(fuzzerId)") + self.logger.info("Created new fuzzer: fuzzerId=\(fuzzerId)") return fuzzerId } @@ -362,10 +384,115 @@ public class PostgreSQLStorage { public func getRecentPrograms(fuzzerId: Int, since: Date, limit: Int = 100) async throws -> [(Program, ExecutionMetadata)] { logger.info("Getting recent programs: fuzzerId=\(fuzzerId), since=\(since), limit=\(limit)") - // For now, return empty array - // TODO: Implement actual database query when PostgreSQL is set up - logger.info("Mock recent programs lookup: no programs found") - return [] + guard let eventLoopGroup = databasePool.getEventLoopGroup() else { + throw PostgreSQLStorageError.noResult + } + + let connection = try await PostgresConnection.connect( + on: eventLoopGroup.next(), + configuration: PostgresConnection.Configuration( + host: "localhost", + port: 5433, + username: "fuzzilli", + password: "fuzzilli123", + database: "fuzzilli", + tls: .disable + ), + id: 0, + logger: logger + ) + defer { Task { _ = try? await connection.close() } } + + // Query for recent programs with their latest execution metadata + let queryString = """ + SELECT + p.program_base64, + p.program_size, + p.program_hash, + p.created_at, + eo.outcome, + eo.description, + e.execution_time_ms, + e.coverage_total, + e.signal_code, + e.exit_code + FROM program p + LEFT JOIN execution e ON p.program_base64 = e.program_base64 + LEFT JOIN execution_outcome eo ON e.execution_outcome_id = eo.id + WHERE p.fuzzer_id = \(fuzzerId) + AND p.created_at >= '\(since.ISO8601Format())' + ORDER BY p.created_at DESC + LIMIT \(limit) + """ + + let query = PostgresQuery(stringLiteral: queryString) + let result = try await connection.query(query, logger: self.logger) + let rows = try await result.collect() + + var programs: [(Program, ExecutionMetadata)] = [] + + for row in rows { + let programBase64 = try row.decode(String.self, context: .default) + let programSize = try row.decode(Int.self, context: .default) + let programHash = try row.decode(String.self, context: .default) + let createdAt = try row.decode(Date.self, context: .default) + let outcome = try row.decode(String?.self, context: .default) + let description = try row.decode(String?.self, context: .default) + let executionTimeMs = try row.decode(Int?.self, context: .default) + let coverageTotal = try row.decode(Double?.self, context: .default) + let signalCode = try row.decode(Int?.self, context: .default) + let exitCode = try row.decode(Int?.self, context: .default) + + // Decode the program from base64 + guard let programData = Data(base64Encoded: programBase64) else { + logger.warning("Failed to decode base64 data for program: \(programHash)") + continue + } + + let program: Program + do { + let protobuf = try Fuzzilli_Protobuf_Program(serializedBytes: programData) + program = try Program(from: protobuf) + } catch { + logger.warning("Failed to decode program from protobuf: \(programHash), error: \(error)") + continue + } + + // Create execution metadata + + // Map outcome string to database ID + let outcomeId: Int + switch (outcome ?? "Succeeded").lowercased() { + case "crashed": + outcomeId = 1 + case "failed": + outcomeId = 2 + case "succeeded": + outcomeId = 3 + case "timedout": + outcomeId = 4 + case "sigcheck": + outcomeId = 34 + default: + outcomeId = 3 // Default to succeeded + } + + let dbOutcome = DatabaseExecutionOutcome( + id: outcomeId, + outcome: outcome ?? "Succeeded", + description: description ?? "Program executed successfully" + ) + + var metadata = ExecutionMetadata(lastOutcome: dbOutcome) + if let coverage = coverageTotal { + metadata.lastCoverage = coverage + } + + programs.append((program, metadata)) + } + + logger.info("Loaded \(programs.count) recent programs from database") + return programs } /// Update program metadata diff --git a/Sources/FuzzilliCli/main.swift b/Sources/FuzzilliCli/main.swift index 6180ac08f..5ca6524f2 100755 --- a/Sources/FuzzilliCli/main.swift +++ b/Sources/FuzzilliCli/main.swift @@ -518,7 +518,8 @@ func makeFuzzer(with configuration: Configuration) -> Fuzzer { } let databasePool = DatabasePool(connectionString: postgresUrl) - let fuzzerInstanceId = "fuzzer-\(UUID().uuidString.prefix(8))" + // Use a consistent fuzzer instance ID so we can resume with the same fuzzer + let fuzzerInstanceId = "fuzzer-main" corpus = PostgreSQLCorpus( minSize: minCorpusSize, From a05f8b4c1c2adf9a1b6a016743e8749efcf96acc Mon Sep 17 00:00:00 2001 From: Oleg Lazari Date: Wed, 22 Oct 2025 01:34:53 -0400 Subject: [PATCH 4/5] Added resuming + batching --- .../Fuzzilli/Corpus/PostgreSQLCorpus.swift | 129 ++++++--- .../Fuzzilli/Database/PostgreSQLStorage.swift | 247 ++++++++++++++++-- Sources/FuzzilliCli/main.swift | 28 +- 3 files changed, 336 insertions(+), 68 deletions(-) diff --git a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift index 58e0e9c24..17a02406a 100644 --- a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift +++ b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift @@ -55,7 +55,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { /// Batch execution storage private var pendingExecutions: [(Program, ProgramAspects, DatabaseExecutionPurpose)] = [] - private let executionBatchSize = 10 + private let executionBatchSize: Int private let executionBatchLock = NSLock() // MARK: - Initialization @@ -82,19 +82,82 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { self.resume = resume self.storage = PostgreSQLStorage(databasePool: databasePool) + // Set fixed batch size to 1 million for optimal performance + self.executionBatchSize = 1_000_000 + self.programs = RingBuffer(maxSize: maxSize) self.ages = RingBuffer(maxSize: maxSize) self.programHashes = RingBuffer(maxSize: maxSize) super.init(name: "PostgreSQLCorpus") + + // Setup signal handlers for graceful shutdown + setupSignalHandlers() + } + + deinit { + // Unregister this instance + PostgreSQLCorpus.unregisterInstance(self) + + // Commit any pending batches when the corpus is deallocated + Task { + await commitPendingBatches() + } + } + + + // MARK: - Signal Handling and Early Exit + + private func setupSignalHandlers() { + // Handle SIGINT (Ctrl+C) and SIGTERM for graceful shutdown + // Note: Signal handling is simplified for cross-platform compatibility + // The deinit method will handle cleanup when the object is deallocated + } + + // Instance tracking for cleanup (simplified without signal handling) + private static var allInstances: [PostgreSQLCorpus] = [] + private static let instancesLock = NSLock() + + private static func registerInstance(_ instance: PostgreSQLCorpus) { + instancesLock.lock() + defer { instancesLock.unlock() } + allInstances.append(instance) + } + + private static func unregisterInstance(_ instance: PostgreSQLCorpus) { + instancesLock.lock() + defer { instancesLock.unlock() } + allInstances.removeAll { $0 === instance } + } + + private func commitPendingBatches() async { + guard let fuzzerId = fuzzerId else { return } + + // Commit pending executions + executionBatchLock.lock() + let queuedExecutions = pendingExecutions + pendingExecutions.removeAll() + executionBatchLock.unlock() + + if !queuedExecutions.isEmpty { + do { + try await processExecutionBatch(queuedExecutions) + // logger.debug("Committed \(queuedExecutions.count) pending executions on exit") + } catch { + logger.error("Failed to commit pending executions on exit: \(error)") + } + } } override func initialize() { + // Register this instance for signal handling + PostgreSQLCorpus.registerInstance(self) + // Initialize database pool and register fuzzer (only once) Task { do { try await databasePool.initialize() - logger.info("Database pool initialized successfully") + // logger.debug("Database pool initialized successfully") // Register this fuzzer instance in the database (only once) if !fuzzerRegistered { @@ -102,10 +165,10 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { let id = try await registerFuzzerWithRetry() fuzzerId = id fuzzerRegistered = true - logger.info("Fuzzer registered in database with ID: \(id)") + // logger.debug("Fuzzer registered in database with ID: \(id)") } catch { logger.error("Failed to register fuzzer after retries: \(error)") - logger.info("Fuzzer will continue without database registration - executions will be queued") + // logger.debug("Fuzzer will continue without database registration - executions will be queued") } } @@ -164,15 +227,15 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { // Schedule periodic synchronization with PostgreSQL fuzzer.timers.scheduleTask(every: syncInterval, syncWithDatabase) - logger.info("Scheduled database sync every \(syncInterval) seconds") + // logger.debug("Scheduled database sync every \(syncInterval) seconds") // Schedule periodic flush of execution batch fuzzer.timers.scheduleTask(every: 5.0, flushExecutionBatch) - logger.info("Scheduled execution batch flush every 5 seconds") + // logger.debug("Scheduled execution batch flush every 5 seconds") // Schedule periodic retry of fuzzer registration if it failed fuzzer.timers.scheduleTask(every: 30.0, retryFuzzerRegistration) - logger.info("Scheduled fuzzer registration retry every 30 seconds") + // logger.debug("Scheduled fuzzer registration retry every 30 seconds") // Schedule cleanup task (similar to BasicCorpus) if !fuzzer.config.staticCorpus { @@ -233,7 +296,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { return } - logger.info("Processing batch of \(batch.count) executions") + // logger.debug("Processing batch of \(batch.count) executions") for (program, aspects, executionType) in batch { do { @@ -258,14 +321,14 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0 ) - logger.info("Stored execution in database: programHash=\(programHash), executionId=\(executionId)") + // logger.debug("Stored execution in database: programHash=\(programHash), executionId=\(executionId)") } catch { logger.error("Failed to store execution in database: \(error)") } } - logger.info("Completed processing batch of \(batch.count) executions") + // logger.debug("Completed processing batch of \(batch.count) executions") } /// Flush any pending executions in the batch @@ -276,7 +339,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { executionBatchLock.unlock() if !batch.isEmpty { - logger.info("Flushing \(batch.count) pending executions") + // logger.debug("Flushing \(batch.count) pending executions") Task { await processExecutionBatch(batch) } @@ -292,7 +355,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { let id = try await registerFuzzerWithRetry() fuzzerId = id fuzzerRegistered = true - logger.info("Successfully registered fuzzer on retry with ID: \(id)") + // logger.debug("Successfully registered fuzzer on retry with ID: \(id)") // Process any queued executions executionBatchLock.lock() @@ -301,7 +364,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { executionBatchLock.unlock() if !queuedExecutions.isEmpty { - logger.info("Processing \(queuedExecutions.count) queued executions after successful registration") + // logger.debug("Processing \(queuedExecutions.count) queued executions after successful registration") await processExecutionBatch(queuedExecutions) } @@ -352,8 +415,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { // Mark for database sync markForSync(programHash) - logger.info("Added program to PostgreSQL corpus: hash=\(programHash), size=\(program.size), total=\(programs.count)") - logger.info("Program marked for sync. Pending sync operations: \(pendingSyncOperations.count)") + // Program added to corpus silently for performance } public func randomElementForSplicing() -> Program { @@ -400,7 +462,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { defer { cacheLock.unlock() } let res = try encodeProtobufCorpus(Array(programs)) - logger.info("Successfully serialized \(programs.count) programs from PostgreSQL corpus") + // logger.debug("Successfully serialized \(programs.count) programs from PostgreSQL corpus") return res } @@ -419,14 +481,14 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { addInternal(program) } - logger.info("Imported \(newPrograms.count) programs into PostgreSQL corpus") + // logger.debug("Imported \(newPrograms.count) programs into PostgreSQL corpus") } // MARK: - Database Operations /// Load initial corpus from PostgreSQL database private func loadInitialCorpus() async { - logger.info("Loading initial corpus from PostgreSQL...") + // logger.debug("Loading initial corpus from PostgreSQL...") guard let fuzzerId = fuzzerId else { logger.warning("Cannot load initial corpus: fuzzer not registered") @@ -442,7 +504,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { limit: maxSize ) - logger.info("Found \(recentPrograms.count) recent programs to resume") + // logger.debug("Found \(recentPrograms.count) recent programs to resume") // Add programs to the corpus cacheLock.lock() @@ -466,16 +528,16 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { totalEntryCounter += 1 } - logger.info("Resumed PostgreSQL corpus with \(programs.count) programs") + // logger.debug("Resumed PostgreSQL corpus with \(programs.count) programs") // If we have no programs, we need at least one to avoid empty corpus if programs.count == 0 { - logger.info("No programs found to resume, corpus will start empty") + // logger.debug("No programs found to resume, corpus will start empty") } } catch { logger.error("Failed to load initial corpus from PostgreSQL: \(error)") - logger.info("Corpus will start empty and build up from scratch") + // logger.debug("Corpus will start empty and build up from scratch") } } @@ -498,7 +560,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { guard !hashesToSync.isEmpty else { return } - logger.info("Syncing \(hashesToSync.count) programs with PostgreSQL...") + // Syncing programs with PostgreSQL silently // Get programs to sync from cache cacheLock.lock() @@ -524,7 +586,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { metadata: metadata ) - logger.info("Successfully synced program to database: \(programHash)") + // Program synced to database silently } catch { logger.error("Failed to sync program to database: \(error)") @@ -535,7 +597,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { } } - logger.info("Database sync completed for \(programsToSync.count) programs") + // Database sync completed silently } /// Register fuzzer with retry logic @@ -549,12 +611,12 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { for attempt in 1...maxRetries { do { - logger.info("Attempting to register fuzzer (attempt \(attempt)/\(maxRetries))") + // logger.debug("Attempting to register fuzzer (attempt \(attempt)/\(maxRetries))") let id = try await storage.registerFuzzer( name: fuzzerName, engineType: engineType ) - logger.info("Successfully registered fuzzer with ID: \(id)") + // logger.debug("Successfully registered fuzzer with ID: \(id)") return id } catch { lastError = error @@ -584,8 +646,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { do { // Use the registered fuzzer ID guard let fuzzerId = fuzzerId else { - logger.error("Cannot store execution: fuzzer not registered") - return + return // Silent fail for performance } // Store the program in the program table @@ -612,10 +673,10 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { fuzzout: executionData.fuzzout ) - logger.info("Stored execution with cached data: programHash=\(programHash), executionId=\(executionId), execTime=\(executionData.execTime), outcome=\(executionData.outcome)") + // No logging for performance - just store silently } catch { - logger.error("Failed to store execution with cached data: \(error)") + // Silent fail for performance - errors are not critical for fuzzing } } @@ -648,7 +709,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0 ) - logger.info("Stored execution with metadata: programHash=\(programHash), executionId=\(executionId), execTime=\(execution.execTime), outcome=\(execution.outcome)") + // logger.debug("Stored execution with metadata: programHash=\(programHash), executionId=\(executionId), execTime=\(execution.execTime), outcome=\(execution.outcome)") } catch { logger.error("Failed to store execution with metadata: \(error)") @@ -685,7 +746,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0 ) - logger.info("Stored execution in database: programHash=\(programHash), executionId=\(executionId)") + // logger.debug("Stored execution in database: programHash=\(programHash), executionId=\(executionId)") } catch { logger.error("Failed to store execution in database: \(error)") @@ -758,7 +819,7 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { } } - logger.info("PostgreSQL corpus cleanup finished: \(self.programs.count) -> \(newPrograms.count)") + // logger.debug("PostgreSQL corpus cleanup finished: \(self.programs.count) -> \(newPrograms.count)") programs = newPrograms ages = newAges programHashes = newHashes diff --git a/Sources/Fuzzilli/Database/PostgreSQLStorage.swift b/Sources/Fuzzilli/Database/PostgreSQLStorage.swift index ff64a3859..b68006c2d 100644 --- a/Sources/Fuzzilli/Database/PostgreSQLStorage.swift +++ b/Sources/Fuzzilli/Database/PostgreSQLStorage.swift @@ -29,7 +29,7 @@ public class PostgreSQLStorage { /// Register a new fuzzer instance in the database public func registerFuzzer(name: String, engineType: String, hostname: String? = nil) async throws -> Int { - logger.info("Registering fuzzer: name=\(name), engineType=\(engineType), hostname=\(hostname ?? "none")") + logger.debug("Registering fuzzer: name=\(name), engineType=\(engineType), hostname=\(hostname ?? "none")") // Use direct connection to avoid connection pool deadlock guard let eventLoopGroup = databasePool.getEventLoopGroup() else { @@ -64,9 +64,9 @@ public class PostgreSQLStorage { if existingStatus != "active" { let updateQuery: PostgresQuery = "UPDATE main SET status = 'active' WHERE fuzzer_id = \(existingFuzzerId)" try await connection.query(updateQuery, logger: self.logger) - logger.info("Reactivated existing fuzzer: fuzzerId=\(existingFuzzerId)") + logger.debug("Reactivated existing fuzzer: fuzzerId=\(existingFuzzerId)") } else { - logger.info("Reusing existing active fuzzer: fuzzerId=\(existingFuzzerId)") + logger.debug("Reusing existing active fuzzer: fuzzerId=\(existingFuzzerId)") } return existingFuzzerId @@ -86,13 +86,13 @@ public class PostgreSQLStorage { } let fuzzerId = try row.decode(Int.self, context: .default) - self.logger.info("Created new fuzzer: fuzzerId=\(fuzzerId)") + self.logger.debug("Created new fuzzer: fuzzerId=\(fuzzerId)") return fuzzerId } /// Get fuzzer instance by name public func getFuzzer(name: String) async throws -> FuzzerInstance? { - logger.info("Getting fuzzer: name=\(name)") + logger.debug("Getting fuzzer: name=\(name)") // Use direct connection to avoid connection pool deadlock guard let eventLoopGroup = databasePool.getEventLoopGroup() else { @@ -136,17 +136,83 @@ public class PostgreSQLStorage { status: status ) - self.logger.info("Fuzzer found: \(fuzzerName) (ID: \(fuzzerId))") + self.logger.debug("Fuzzer found: \(fuzzerName) (ID: \(fuzzerId))") return fuzzer } // MARK: - Program Management + /// Store multiple programs in batch for better performance + public func storeProgramsBatch(programs: [(Program, ExecutionMetadata)], fuzzerId: Int) async throws -> [String] { + guard !programs.isEmpty else { return [] } + + // Use direct connection to avoid connection pool deadlock + guard let eventLoopGroup = databasePool.getEventLoopGroup() else { + throw PostgreSQLStorageError.noResult + } + + let connection = try await PostgresConnection.connect( + on: eventLoopGroup.next(), + configuration: PostgresConnection.Configuration( + host: "localhost", + port: 5433, + username: "fuzzilli", + password: "fuzzilli123", + database: "fuzzilli", + tls: .disable + ), + id: 0, + logger: logger + ) + defer { Task { _ = try? await connection.close() } } + + var programHashes: [String] = [] + var fuzzerValues: [String] = [] + var programValues: [String] = [] + + // Prepare batch data + for (program, _) in programs { + let programHash = DatabaseUtils.calculateProgramHash(program: program) + let programBase64 = DatabaseUtils.encodeProgramToBase64(program: program) + programHashes.append(programHash) + + // Generate JavaScript code from the program + let lifter = JavaScriptLifter(ecmaVersion: .es6) + let javascriptCode = lifter.lift(program, withOptions: []) + let javascriptCodeBase64 = Data(javascriptCode.utf8).base64EncodedString() + + // Escape single quotes in strings + let escapedProgramBase64 = programBase64.replacingOccurrences(of: "'", with: "''") + let escapedJavascriptCodeBase64 = javascriptCodeBase64.replacingOccurrences(of: "'", with: "''") + + fuzzerValues.append("('\(escapedProgramBase64)', \(fuzzerId), \(program.size), '\(programHash)')") + programValues.append("('\(escapedProgramBase64)', \(fuzzerId), \(program.size), '\(programHash)', '\(escapedJavascriptCodeBase64)')") + } + + // Batch insert into fuzzer table + if !fuzzerValues.isEmpty { + let fuzzerQueryString = "INSERT INTO fuzzer (program_base64, fuzzer_id, program_size, program_hash) VALUES " + + fuzzerValues.joined(separator: ", ") + " ON CONFLICT (program_base64) DO NOTHING" + let fuzzerQuery = PostgresQuery(stringLiteral: fuzzerQueryString) + try await connection.query(fuzzerQuery, logger: self.logger) + } + + // Batch insert into program table + if !programValues.isEmpty { + let programQueryString = "INSERT INTO program (program_base64, fuzzer_id, program_size, program_hash, javascript_code) VALUES " + + programValues.joined(separator: ", ") + " ON CONFLICT (program_base64) DO NOTHING" + let programQuery = PostgresQuery(stringLiteral: programQueryString) + try await connection.query(programQuery, logger: self.logger) + } + + return programHashes + } + /// Store a program in the database with execution metadata public func storeProgram(program: Program, fuzzerId: Int, metadata: ExecutionMetadata) async throws -> String { let programHash = DatabaseUtils.calculateProgramHash(program: program) let programBase64 = DatabaseUtils.encodeProgramToBase64(program: program) - logger.info("Storing program: hash=\(programHash), fuzzerId=\(fuzzerId), executionCount=\(metadata.executionCount)") + logger.debug("Storing program: hash=\(programHash), fuzzerId=\(fuzzerId), executionCount=\(metadata.executionCount)") // Use direct connection to avoid connection pool deadlock guard let eventLoopGroup = databasePool.getEventLoopGroup() else { @@ -195,32 +261,114 @@ public class PostgreSQLStorage { let programQuery = PostgresQuery(stringLiteral: programQueryString) try await connection.query(programQuery, logger: self.logger) - self.logger.info("Program storage successful: hash=\(programHash)") + self.logger.debug("Program storage successful: hash=\(programHash)") return programHash } /// Get program by hash public func getProgram(hash: String) async throws -> Program? { - logger.info("Getting program: hash=\(hash)") + logger.debug("Getting program: hash=\(hash)") // For now, return nil (program not found) // TODO: Implement actual database query when PostgreSQL is set up - logger.info("Mock program lookup: program not found") + logger.debug("Mock program lookup: program not found") return nil } /// Get program metadata for a specific fuzzer public func getProgramMetadata(programHash: String, fuzzerId: Int) async throws -> ExecutionMetadata? { - logger.info("Getting program metadata: hash=\(programHash), fuzzerId=\(fuzzerId)") + logger.debug("Getting program metadata: hash=\(programHash), fuzzerId=\(fuzzerId)") // For now, return nil (metadata not found) // TODO: Implement actual database query when PostgreSQL is set up - logger.info("Mock metadata lookup: metadata not found") + logger.debug("Mock metadata lookup: metadata not found") return nil } // MARK: - Execution Management + /// Store multiple executions in batch for better performance + public func storeExecutionsBatch(executions: [ExecutionBatchData], fuzzerId: Int) async throws -> [Int] { + guard !executions.isEmpty else { return [] } + + // Use direct connection to avoid connection pool deadlock + guard let eventLoopGroup = databasePool.getEventLoopGroup() else { + throw PostgreSQLStorageError.noResult + } + + let connection = try await PostgresConnection.connect( + on: eventLoopGroup.next(), + configuration: PostgresConnection.Configuration( + host: "localhost", + port: 5433, + username: "fuzzilli", + password: "fuzzilli123", + database: "fuzzilli", + tls: .disable + ), + id: 0, + logger: logger + ) + defer { Task { _ = try? await connection.close() } } + + var executionIds: [Int] = [] + var executionValues: [String] = [] + + // Prepare batch data + for executionData in executions { + _ = DatabaseUtils.calculateProgramHash(program: executionData.program) + let programBase64 = DatabaseUtils.encodeProgramToBase64(program: executionData.program) + + let executionTypeId = DatabaseUtils.mapExecutionType(purpose: executionData.executionType) + let mutatorTypeId = executionData.mutatorType != nil ? DatabaseUtils.mapMutatorType(mutator: executionData.mutatorType!) : nil + + // Extract execution metadata from ExecutionOutcome + let (signalCode, exitCode) = extractExecutionMetadata(from: executionData.outcome) + + // Use signal-aware mapping for execution outcomes + let outcomeId = DatabaseUtils.mapExecutionOutcomeWithSignal(outcome: executionData.outcome, signalCode: signalCode) + + let mutatorTypeValue = mutatorTypeId != nil ? "\(mutatorTypeId!)" : "NULL" + let feedbackVectorValue = executionData.feedbackVector != nil ? "'\(executionData.feedbackVector!.base64EncodedString())'" : "NULL" + let signalCodeValue = signalCode != nil ? "\(signalCode!)" : "NULL" + let exitCodeValue = exitCode != nil ? "\(exitCode!)" : "NULL" + let stdoutValue = executionData.stdout != nil ? "'\(executionData.stdout!.replacingOccurrences(of: "'", with: "''"))'" : "NULL" + let stderrValue = executionData.stderr != nil ? "'\(executionData.stderr!.replacingOccurrences(of: "'", with: "''"))'" : "NULL" + let fuzzoutValue = executionData.fuzzout != nil ? "'\(executionData.fuzzout!.replacingOccurrences(of: "'", with: "''"))'" : "NULL" + + executionValues.append(""" + ('\(programBase64.replacingOccurrences(of: "'", with: "''"))', \(executionTypeId), + \(mutatorTypeValue), \(outcomeId), \(executionData.coverage), + \(executionData.executionTimeMs), \(signalCodeValue), \(exitCodeValue), + \(stdoutValue), \(stderrValue), \(fuzzoutValue), + \(feedbackVectorValue), NOW()) + """) + } + + // Batch insert executions + if !executionValues.isEmpty { + let queryString = """ + INSERT INTO execution ( + program_base64, execution_type_id, mutator_type_id, + execution_outcome_id, coverage_total, execution_time_ms, + signal_code, exit_code, stdout, stderr, fuzzout, + feedback_vector, created_at + ) VALUES \(executionValues.joined(separator: ", ")) RETURNING execution_id + """ + + let query = PostgresQuery(stringLiteral: queryString) + let result = try await connection.query(query, logger: self.logger) + let rows = try await result.collect() + + for row in rows { + let executionId = try row.decode(Int.self, context: .default) + executionIds.append(executionId) + } + } + + return executionIds + } + /// Store execution record in the database public func storeExecution( program: Program, @@ -238,7 +386,7 @@ public class PostgreSQLStorage { ) async throws -> Int { let programHash = DatabaseUtils.calculateProgramHash(program: program) let programBase64 = DatabaseUtils.encodeProgramToBase64(program: program) - logger.info("Storing execution: hash=\(programHash), fuzzerId=\(fuzzerId), type=\(executionType), outcome=\(outcome)") + logger.debug("Storing execution: hash=\(programHash), fuzzerId=\(fuzzerId), type=\(executionType), outcome=\(outcome)") // Use direct connection to avoid connection pool deadlock guard let eventLoopGroup = databasePool.getEventLoopGroup() else { @@ -301,7 +449,7 @@ public class PostgreSQLStorage { } let executionId = try row.decode(Int.self, context: .default) - self.logger.info("Execution storage successful: executionId=\(executionId)") + self.logger.debug("Execution storage successful: executionId=\(executionId)") return executionId } @@ -347,11 +495,11 @@ public class PostgreSQLStorage { /// Get execution history for a program public func getExecutionHistory(programHash: String, fuzzerId: Int, limit: Int = 100) async throws -> [ExecutionRecord] { - logger.info("Getting execution history: hash=\(programHash), fuzzerId=\(fuzzerId), limit=\(limit)") + logger.debug("Getting execution history: hash=\(programHash), fuzzerId=\(fuzzerId), limit=\(limit)") // For now, return empty array // TODO: Implement actual database query when PostgreSQL is set up - logger.info("Mock execution history lookup: no executions found") + logger.debug("Mock execution history lookup: no executions found") return [] } @@ -369,12 +517,12 @@ public class PostgreSQLStorage { stderr: String? = nil ) async throws -> Int { let programHash = DatabaseUtils.calculateProgramHash(program: program) - logger.info("Storing crash: hash=\(programHash), fuzzerId=\(fuzzerId), executionId=\(executionId), type=\(crashType)") + logger.debug("Storing crash: hash=\(programHash), fuzzerId=\(fuzzerId), executionId=\(executionId), type=\(crashType)") // For now, return a mock crash ID // TODO: Implement actual database storage when PostgreSQL is set up let mockCrashId = Int.random(in: 1...1000) - logger.info("Mock crash storage successful: crashId=\(mockCrashId)") + logger.debug("Mock crash storage successful: crashId=\(mockCrashId)") return mockCrashId } @@ -382,7 +530,7 @@ public class PostgreSQLStorage { /// Get recent programs with metadata for a fuzzer public func getRecentPrograms(fuzzerId: Int, since: Date, limit: Int = 100) async throws -> [(Program, ExecutionMetadata)] { - logger.info("Getting recent programs: fuzzerId=\(fuzzerId), since=\(since), limit=\(limit)") + logger.debug("Getting recent programs: fuzzerId=\(fuzzerId), since=\(since), limit=\(limit)") guard let eventLoopGroup = databasePool.getEventLoopGroup() else { throw PostgreSQLStorageError.noResult @@ -433,15 +581,15 @@ public class PostgreSQLStorage { for row in rows { let programBase64 = try row.decode(String.self, context: .default) - let programSize = try row.decode(Int.self, context: .default) + _ = try row.decode(Int.self, context: .default) // programSize let programHash = try row.decode(String.self, context: .default) - let createdAt = try row.decode(Date.self, context: .default) + _ = try row.decode(Date.self, context: .default) // createdAt let outcome = try row.decode(String?.self, context: .default) let description = try row.decode(String?.self, context: .default) - let executionTimeMs = try row.decode(Int?.self, context: .default) + _ = try row.decode(Int?.self, context: .default) // executionTimeMs let coverageTotal = try row.decode(Double?.self, context: .default) - let signalCode = try row.decode(Int?.self, context: .default) - let exitCode = try row.decode(Int?.self, context: .default) + _ = try row.decode(Int?.self, context: .default) // signalCode + _ = try row.decode(Int?.self, context: .default) // exitCode // Decode the program from base64 guard let programData = Data(base64Encoded: programBase64) else { @@ -491,24 +639,24 @@ public class PostgreSQLStorage { programs.append((program, metadata)) } - logger.info("Loaded \(programs.count) recent programs from database") + logger.debug("Loaded \(programs.count) recent programs from database") return programs } /// Update program metadata public func updateProgramMetadata(programHash: String, fuzzerId: Int, metadata: ExecutionMetadata) async throws { - logger.info("Updating program metadata: hash=\(programHash), fuzzerId=\(fuzzerId), executionCount=\(metadata.executionCount)") + logger.debug("Updating program metadata: hash=\(programHash), fuzzerId=\(fuzzerId), executionCount=\(metadata.executionCount)") // For now, just log the operation // TODO: Implement actual database update when PostgreSQL is set up - logger.info("Mock metadata update successful") + logger.debug("Mock metadata update successful") } // MARK: - Statistics /// Get storage statistics public func getStorageStatistics() async throws -> StorageStatistics { - logger.info("Getting storage statistics") + logger.debug("Getting storage statistics") // For now, return mock statistics // TODO: Implement actual database statistics when PostgreSQL is set up @@ -518,13 +666,54 @@ public class PostgreSQLStorage { totalCrashes: 0, activeFuzzers: 0 ) - logger.info("Mock statistics: \(mockStats.description)") + logger.debug("Mock statistics: \(mockStats.description)") return mockStats } } // MARK: - Supporting Types +/// Batch data for execution storage +public struct ExecutionBatchData { + public let program: Program + public let executionType: DatabaseExecutionPurpose + public let mutatorType: String? + public let outcome: ExecutionOutcome + public let coverage: Double + public let executionTimeMs: Int + public let feedbackVector: Data? + public let coverageEdges: Set + public let stdout: String? + public let stderr: String? + public let fuzzout: String? + + public init( + program: Program, + executionType: DatabaseExecutionPurpose, + mutatorType: String? = nil, + outcome: ExecutionOutcome, + coverage: Double = 0.0, + executionTimeMs: Int = 0, + feedbackVector: Data? = nil, + coverageEdges: Set = [], + stdout: String? = nil, + stderr: String? = nil, + fuzzout: String? = nil + ) { + self.program = program + self.executionType = executionType + self.mutatorType = mutatorType + self.outcome = outcome + self.coverage = coverage + self.executionTimeMs = executionTimeMs + self.feedbackVector = feedbackVector + self.coverageEdges = coverageEdges + self.stdout = stdout + self.stderr = stderr + self.fuzzout = fuzzout + } +} + /// Storage statistics public struct StorageStatistics { public let totalPrograms: Int diff --git a/Sources/FuzzilliCli/main.swift b/Sources/FuzzilliCli/main.swift index 5ca6524f2..7f4867017 100755 --- a/Sources/FuzzilliCli/main.swift +++ b/Sources/FuzzilliCli/main.swift @@ -210,7 +210,7 @@ if corpusName == "markov" && (args.int(for: "--maxCorpusSize") != nil || args.in configError("--maxCorpusSize, --minCorpusSize, --minMutationsPerSample are not compatible with the Markov corpus") } -if (resume || overwrite) && storagePath == nil { +if (resume || overwrite) && storagePath == nil && corpusName != "postgresql" { configError("--resume and --overwrite require --storagePath") } @@ -517,9 +517,25 @@ func makeFuzzer(with configuration: Configuration) -> Fuzzer { logger.fatal("PostgreSQL URL is required for PostgreSQL corpus") } - let databasePool = DatabasePool(connectionString: postgresUrl) - // Use a consistent fuzzer instance ID so we can resume with the same fuzzer - let fuzzerInstanceId = "fuzzer-main" + // Generate database name based on resume flag + let databaseName: String + let fuzzerInstanceId: String + + if resume { + // Use fixed database name for resume + databaseName = "database-main" + fuzzerInstanceId = "fuzzer-main" + } else { + // Generate dynamic database name with 8-char hash + let randomHash = String(UUID().uuidString.prefix(8)) + databaseName = "database-\(randomHash)" + fuzzerInstanceId = "fuzzer-\(randomHash)" + } + + // Replace database name in the connection string + let modifiedPostgresUrl = postgresUrl.replacingOccurrences(of: "/fuzzilli", with: "/\(databaseName)") + + let databasePool = DatabasePool(connectionString: modifiedPostgresUrl) corpus = PostgreSQLCorpus( minSize: minCorpusSize, @@ -530,7 +546,9 @@ func makeFuzzer(with configuration: Configuration) -> Fuzzer { ) logger.info("Created PostgreSQL corpus with instance ID: \(fuzzerInstanceId)") - logger.info("PostgreSQL URL: \(postgresUrl)") + logger.info("Database name: \(databaseName)") + logger.info("PostgreSQL URL: \(modifiedPostgresUrl)") + logger.info("Resume mode: \(resume)") logger.info("Sync interval: \(syncInterval) seconds") logger.info("Validate before cache: \(validateBeforeCache)") logger.info("Execution history size: \(executionHistorySize)") From 600ca6066fc63889f6870c5030346bc06a4ab449 Mon Sep 17 00:00:00 2001 From: Oleg Lazari Date: Wed, 22 Oct 2025 01:54:20 -0400 Subject: [PATCH 5/5] Added better batching --- .../Fuzzilli/Corpus/PostgreSQLCorpus.swift | 122 ++++++++++++------ 1 file changed, 79 insertions(+), 43 deletions(-) diff --git a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift index 17a02406a..de55302a3 100644 --- a/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift +++ b/Sources/Fuzzilli/Corpus/PostgreSQLCorpus.swift @@ -82,8 +82,8 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { self.resume = resume self.storage = PostgreSQLStorage(databasePool: databasePool) - // Set fixed batch size to 1 million for optimal performance - self.executionBatchSize = 1_000_000 + // Set optimized batch size for better throughput (reduced from 1M to 100k for more frequent processing) + self.executionBatchSize = 100_000 self.programs = RingBuffer(maxSize: maxSize) self.ages = RingBuffer(maxSize: maxSize) @@ -93,6 +93,9 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { // Setup signal handlers for graceful shutdown setupSignalHandlers() + + // Start periodic batch flushing for better throughput + startPeriodicBatchFlush() } deinit { @@ -106,6 +109,18 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { } + // MARK: - Performance Optimizations + + private func startPeriodicBatchFlush() { + // Flush batches every 5 seconds to ensure timely processing + Task { + while true { + try? await Task.sleep(nanoseconds: 5_000_000_000) // 5 seconds + flushExecutionBatch() + } + } + } + // MARK: - Signal Handling and Early Exit private func setupSignalHandlers() { @@ -134,10 +149,11 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { guard let fuzzerId = fuzzerId else { return } // Commit pending executions - executionBatchLock.lock() - let queuedExecutions = pendingExecutions - pendingExecutions.removeAll() - executionBatchLock.unlock() + let queuedExecutions = executionBatchLock.withLock { + let queuedExecutions = pendingExecutions + pendingExecutions.removeAll() + return queuedExecutions + } if !queuedExecutions.isEmpty { do { @@ -272,17 +288,20 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { /// Add execution to batch for later processing private func addToExecutionBatch(_ program: Program, _ aspects: ProgramAspects, executionType: DatabaseExecutionPurpose) { - executionBatchLock.lock() - defer { executionBatchLock.unlock() } - - pendingExecutions.append((program, aspects, executionType)) + // Use atomic operations to avoid blocking locks + let shouldProcessBatch: [(Program, ProgramAspects, DatabaseExecutionPurpose)]? = executionBatchLock.withLock { + pendingExecutions.append((program, aspects, executionType)) + let shouldProcess = pendingExecutions.count >= executionBatchSize + if shouldProcess { + let batch = pendingExecutions + pendingExecutions.removeAll() + return batch + } + return nil + } - // Process batch if it's full - if pendingExecutions.count >= executionBatchSize { - let batch = pendingExecutions - pendingExecutions.removeAll() - executionBatchLock.unlock() - + // Process batch asynchronously if needed + if let batch = shouldProcessBatch { Task { await processExecutionBatch(batch) } @@ -298,45 +317,61 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { // logger.debug("Processing batch of \(batch.count) executions") - for (program, aspects, executionType) in batch { - do { - // Store the program in the program table - let programHash = try await storage.storeProgram( - program: program, - fuzzerId: fuzzerId, - metadata: ExecutionMetadata(lastOutcome: DatabaseExecutionOutcome( + do { + // Prepare batch data for programs (deduplicate by program hash) + var uniquePrograms: [String: (Program, ExecutionMetadata)] = [:] + var executionBatchData: [ExecutionBatchData] = [] + + for (program, aspects, executionType) in batch { + let programHash = DatabaseUtils.calculateProgramHash(program: program) + + // Only store unique programs + if uniquePrograms[programHash] == nil { + let metadata = ExecutionMetadata(lastOutcome: DatabaseExecutionOutcome( id: DatabaseUtils.mapExecutionOutcome(outcome: aspects.outcome), outcome: aspects.outcome.description, description: aspects.outcome.description )) - ) + uniquePrograms[programHash] = (program, metadata) + } - // Store the execution record - let executionId = try await storage.storeExecution( + // Prepare execution data + let executionData = ExecutionBatchData( program: program, - fuzzerId: fuzzerId, executionType: executionType, mutatorType: nil, outcome: aspects.outcome, - coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0 + coverage: aspects is CovEdgeSet ? Double((aspects as! CovEdgeSet).count) : 0.0, + coverageEdges: Set() // Empty for now ) - - // logger.debug("Stored execution in database: programHash=\(programHash), executionId=\(executionId)") - - } catch { - logger.error("Failed to store execution in database: \(error)") + executionBatchData.append(executionData) } + + // Batch store programs (only unique ones) + let programBatch = Array(uniquePrograms.values) + if !programBatch.isEmpty { + _ = try await storage.storeProgramsBatch(programs: programBatch, fuzzerId: fuzzerId) + } + + // Batch store executions + if !executionBatchData.isEmpty { + _ = try await storage.storeExecutionsBatch(executions: executionBatchData, fuzzerId: fuzzerId) + } + + // logger.debug("Completed batch processing: \(programBatch.count) unique programs, \(executionBatchData.count) executions") + + } catch { + logger.error("Failed to process execution batch: \(error)") } - - // logger.debug("Completed processing batch of \(batch.count) executions") } /// Flush any pending executions in the batch private func flushExecutionBatch() { - executionBatchLock.lock() - let batch = pendingExecutions - pendingExecutions.removeAll() - executionBatchLock.unlock() + let batch = executionBatchLock.withLock { + let batch = pendingExecutions + pendingExecutions.removeAll() + return batch + } if !batch.isEmpty { // logger.debug("Flushing \(batch.count) pending executions") @@ -358,10 +393,11 @@ public class PostgreSQLCorpus: ComponentBase, Corpus { // logger.debug("Successfully registered fuzzer on retry with ID: \(id)") // Process any queued executions - executionBatchLock.lock() - let queuedExecutions = pendingExecutions - pendingExecutions.removeAll() - executionBatchLock.unlock() + let queuedExecutions = executionBatchLock.withLock { + let queuedExecutions = pendingExecutions + pendingExecutions.removeAll() + return queuedExecutions + } if !queuedExecutions.isEmpty { // logger.debug("Processing \(queuedExecutions.count) queued executions after successful registration")