pear-git / lib / p2p / sync / git-process-queue.js

/**
 * GitProcessQueue — Bounded-concurrency process queue for git operations
 *
 * Replaces unbounded process spawning (e.g., 65 concurrent git fetches)
 * with a queue that limits concurrent processes and enforces timeouts.
 *
 * Used by ServerSync and future Protomux-based git sync.
 */

import { spawn } from 'bare-subprocess'

let nextJobId = 0

export class GitProcessQueue {
  constructor (opts = {}) {
    this.maxConcurrent = opts.maxConcurrent || 5
    this.timeout = opts.timeout || 120000
    this._active = new Map()   // jobId -> { child, timer, reject }
    this._queued = []           // { jobId, cmd, args, hooks, resolve, reject }
    this._destroyed = false
  }

  /**
   * Queue a process for execution. Returns a promise that resolves with
   * { jobId, code } on success or rejects on error/timeout/cancel.
   */
  spawn (cmd, args, hooks = {}) {
    if (this._destroyed) {
      return Promise.reject(new Error('Queue destroyed'))
    }

    const jobId = 'job-' + (++nextJobId)

    return new Promise((resolve, reject) => {
      this._queued.push({ jobId, cmd, args, hooks, resolve, reject })
      this._drain()
    })
  }

  /**
   * Cancel a running or queued job. Returns true if found, false otherwise.
   */
  cancel (jobId) {
    // Check active jobs
    const active = this._active.get(jobId)
    if (active) {
      this._killJob(jobId, new Error('Cancelled: job ' + jobId))
      return true
    }

    // Check queued jobs
    const idx = this._queued.findIndex(j => j.jobId === jobId)
    if (idx !== -1) {
      const job = this._queued.splice(idx, 1)[0]
      job.reject(new Error('Cancelled: job ' + jobId))
      return true
    }

    return false
  }

  /**
   * Returns array of currently active job IDs.
   */
  activeJobs () {
    return Array.from(this._active.keys())
  }

  /**
   * Kill all active processes, reject all queued jobs.
   */
  destroy () {
    this._destroyed = true

    // Reject all queued
    for (const job of this._queued) {
      job.reject(new Error('Queue destroyed'))
    }
    this._queued = []

    // Kill all active
    for (const jobId of Array.from(this._active.keys())) {
      this._killJob(jobId, new Error('Queue destroyed'))
    }
  }

  /** @private */
  _drain () {
    while (this._queued.length > 0 && this._active.size < this.maxConcurrent) {
      const job = this._queued.shift()
      this._startJob(job)
    }
  }

  /** @private */
  _startJob (job) {
    const { jobId, cmd, args, hooks, resolve, reject } = job
    let done = false

    const child = spawn(cmd, args, {
      stdio: ['ignore', 'pipe', 'pipe']
    })

    const timer = setTimeout(() => {
      if (done) return
      done = true
      this._cleanup(jobId)
      try { child.kill('SIGKILL') } catch {}
      reject(new Error('Timeout after ' + this.timeout + 'ms: job ' + jobId))
      this._drain()
    }, this.timeout)

    const entry = { child, timer, reject }
    this._active.set(jobId, entry)

    if (hooks.onStart) {
      try { hooks.onStart(child) } catch {}
    }

    child.on('exit', (code) => {
      if (done) return
      done = true
      this._cleanup(jobId)

      if (hooks.onExit) {
        try { hooks.onExit(code) } catch {}
      }

      if (code === 0) {
        resolve({ jobId, code })
      } else {
        reject(new Error('Process exited with code ' + code + ': job ' + jobId))
      }

      this._drain()
    })

    child.on('error', (err) => {
      if (done) return
      done = true
      this._cleanup(jobId)

      if (hooks.onExit) {
        try { hooks.onExit(-1) } catch {}
      }

      reject(err)
      this._drain()
    })
  }

  /** @private */
  _cleanup (jobId) {
    const entry = this._active.get(jobId)
    if (entry) {
      clearTimeout(entry.timer)
      this._active.delete(jobId)
    }
  }

  /** @private */
  _killJob (jobId, err) {
    const entry = this._active.get(jobId)
    if (!entry) return
    this._cleanup(jobId)
    try { entry.child.kill('SIGKILL') } catch {}
    entry.reject(err)
    this._drain()
  }
}