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



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

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 



81
82
83
 
# File 'lib/wp2txt/corpus_jobs.rb', line 81

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
 
# File 'lib/wp2txt/corpus_jobs.rb', line 24

def start_extract(params)
  running = @mutex.synchronize { @jobs.values.find { |s| s[: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

  job_id = @mutex.synchronize { format("job-%04d", @seq += 1) }
  state = {
    job_id: job_id, status: "running", started_at: Time.now.utc.iso8601,
    params: params, titles_done: 0, titles_total: nil, cancel: false
  }
  @mutex.synchronize { @jobs[job_id] = state }

  thread = Thread.new do
    corpus = @factory.call
    begin
      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



65
66
67
68
69
70
 
# File 'lib/wp2txt/corpus_jobs.rb', line 65

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

  public_view(state)
end