Greptile Overview

Greptile Summary

This PR addresses metrics data duplication during compaction by implementing a local lock mechanism to ensure only one compaction job runs per stream with daily retention. Additional changes include cleanup of unused file_list module, improved logging in PromQL, and optimization for trigger stream queries.

Key Changes:

• Introduced stream-level locking using DashSet to prevent parallel compaction jobs for daily-partitioned streams
• Added lock acquisition in run_merge() before dispatching jobs to workers
• Added lock release in worker after job completion
• Removed unused file_list.rs module and related cache status endpoint
• Added conditional check to avoid querying non-existent trigger streams in organization summary

Critical Issues Found:

• Race condition in lock acquisition: is_running() and set_running() are separate non-atomic operations that can allow multiple jobs through
• Potential deadlock: Lock cleanup won't execute if worker task panics or is cancelled during shutdown
• Debug println! statement left in organization.rs production code

Confidence Score: 1/5

• This PR has critical concurrency bugs that could cause the exact data duplication it aims to prevent, plus potential deadlocks
• The locking mechanism has a classic check-then-act race condition that defeats its purpose - two threads can both check is_running() simultaneously and both proceed. Additionally, the lock cleanup only happens on normal execution path, not on panics or cancellation, which can cause permanent deadlocks. These are fundamental concurrency bugs that make the fix unreliable.
• Critical attention needed: src/service/compact/mod.rs (race condition), src/service/compact/worker.rs (missing cleanup on panic), src/service/db/compact/stream.rs (needs atomic check-and-set API)

Important Files Changed

File Analysis

Sequence Diagram

sequenceDiagram
    participant Compactor as Compactor Thread
    participant JobQueue as Job Queue
    participant StreamLock as Stream Lock (DashSet)
    participant Worker as Worker Thread
    participant Merger as merge_by_stream

    Compactor->>JobQueue: get_pending_jobs()
    JobQueue-->>Compactor: jobs list
    
    loop For each job
        Compactor->>Compactor: Check partition_time_level == Daily
        alt Daily partition
            Compactor->>Compactor: Check node assignment
            Compactor->>StreamLock: is_running(stream)?
            StreamLock-->>Compactor: false
            Note over Compactor,StreamLock: RACE CONDITION: Another thread<br/>could check here too!
            Compactor->>StreamLock: set_running(stream)
            Compactor->>Worker: send(MergeJob)
        else Not daily
            Compactor->>Worker: send(MergeJob) directly
        end
    end
    
    Worker->>Worker: Receive MergeJob
    Worker->>Merger: merge_by_stream()
    
    alt Success or Error
        Merger-->>Worker: Result
        Worker->>StreamLock: clear_running(stream)
    else Panic or Task Cancelled
        Note over Worker,StreamLock: DEADLOCK: clear_running()<br/>never called, stream locked forever
    end

_{9 files reviewed, 3 comments}

_{Edit Code Review Agent Settings | Greptile}