Class: Wp2txt::CorpusJobManager

Inherits:
Object
  • Object
show all
Defined in:
lib/wp2txt/corpus_jobs.rb

Overview

In-process job manager for long-running corpus extractions (MCP server). Each job runs in its own thread with its own Corpus instance (separate SQLite connections and reader), so it never contends with the main request thread. Jobs are lost when the server exits (documented v1 limit).

Instance Method Summary collapse

Constructor Details

#initialize(corpus_factory) ⇒ CorpusJobManager

Returns a new instance of CorpusJobManager.

Parameters:

  • corpus_factory (#call) —

    returns a fresh Corpus for each job



13
14
15
16
17
18
 
# File 'lib/wp2txt/corpus_jobs.rb', line 13

def initialize(corpus_factory)
  @factory = corpus_factory
  @jobs = {}
  @mutex = Mutex.new
  @seq = 0
end

Instance Method Details

#cancel(job_id) ⇒ Object

Request cancellation; takes effect at the next batch boundary



75
76
77
78
79
80
81
 
# File 'lib/wp2txt/corpus_jobs.rb', line 75

def cancel(job_id)
  state = read(job_id)
  return nil unless state

  update(job_id) { |s| s[:cancel] = true }
  { job_id: job_id, status: state[:status], cancel_requested: true }
end

#list ⇒ Object 



83
84
85
 
# File 'lib/wp2txt/corpus_jobs.rb', line 83

def list
  @mutex.synchronize { @jobs.values.map { |s| public_view(s) } }
end

#start_extract(params) ⇒ Hash

Start an extraction job. params are Corpus#extract_corpus keywords. Only one job runs at a time: each job forks worker processes, and unbounded concurrent jobs would multiply workers against the same dump.

Returns:

  • (Hash) —

    { job_id:, status: "running" } or { error: ... }



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/wp2txt/corpus_jobs.rb', line 24

def start_extract(params)
  job_id = @mutex.synchronize do
    running = @jobs.values.find { |state| state[:status] == "running" }
    if running
      return { error: "another job is already running (#{running[:job_id]}); " \
                      "poll job_status or cancel_job before starting a new one" }
    end
    id = format("job-%04d", @seq += 1)
    @jobs[id] = {
      job_id: id, status: "running", started_at: Time.now.utc.iso8601,
      params: params, titles_done: 0, titles_total: nil, cancel: false
    }
    id
  end

  thread = Thread.new do
    corpus = nil
    begin
      corpus = @factory.call
      result = corpus.extract_corpus(
        **params,
        max_articles: nil,
        progress: lambda { |done, total|
          update(job_id) { |s| s[:titles_done] = done; s[:titles_total] = total }
        },
        cancel_check: -> { read(job_id)[:cancel] }
      )
      update(job_id) { |s| s[:status] = "completed"; s[:result] = result; s[:finished_at] = Time.now.utc.iso8601 }
    rescue Corpus::Cancelled
      update(job_id) { |s| s[:status] = "cancelled"; s[:finished_at] = Time.now.utc.iso8601 }
    rescue StandardError => e
      update(job_id) { |s| s[:status] = "error"; s[:error] = "#{e.class}: #{e.message}"; s[:finished_at] = Time.now.utc.iso8601 }
    ensure
      corpus&.close
    end
  end
  thread.report_on_exception = false
  update(job_id) { |s| s[:thread] = thread }

  { job_id: job_id, status: "running" }
end

#status(job_id) ⇒ Hash?

Returns public job state (no thread object), nil if unknown.

Returns:

  • (Hash, nil) —

    public job state (no thread object), nil if unknown



67
68
69
70
71
72
 
# File 'lib/wp2txt/corpus_jobs.rb', line 67

def status(job_id)
  state = read(job_id)
  return nil unless state

  public_view(state)
end