Module: Cohere::Transcribe::State

Defined in:
lib/cohere/transcribe/state/io.rb,
lib/cohere/transcribe/state/locking.rb,
lib/cohere/transcribe/state/manifest.rb,
lib/cohere/transcribe/state/contracts.rb,
lib/cohere/transcribe/state/checkpoint.rb

Defined Under Namespace

Classes: BoundDirectory, Checkpoint, CheckpointRestore, DirectoryBinding, OutputLockTarget, OutputSetLock, PublicationVerification, SourceSnapshot

Constant Summary collapse

STATE_SCHEMA_VERSION = 
1
STATE_SUFFIX = 
".cohere-transcribe.manifest.json"
CHECKPOINT_SUFFIX = 
".cohere-transcribe.asr.json"
MAX_STATE_BYTES = 
64 * 1024 * 1024
DEFERRED_PUBLICATION_EXCEPTIONS = 
{ Object => :never }.freeze
LOCK_DIRECTORY_PREFIX = 
"cohere-transcribe"
LOCK_DIRECTORY_NAME = 
".cohere-transcribe-locks"
LOCK_SUFFIX = 
".lock"
LEGACY_LOCK_REMOVAL_VERSION = 
"0.2.0"
CONTRACT_SCHEMA_VERSION = 
3
ASR_IMPLEMENTATION_FILES = 
%w[
  asr/native.rb
  audio/decoder.rb
  audio/ffmpeg_native.rb
  audio/segmentation.rb
  configuration.rb
  dense_converter.rb
  gguf_writer.rb
  model_identity.rb
  runtime/engine.rb
  runtime/model_provider.rb
  runtime/preparation.rb
  vad/silero.rb
  vad/silero_vad_v6.onnx
  vad/timestamps.rb
].freeze
RENDER_IMPLEMENTATION_FILES = 
%w[
  alignment/aligner.rb
  alignment/ctc.rb
  alignment/text.rb
  output/publication.rb
  output/rendering.rb
  output/timing.rb
  types.rb
].freeze

Class Method Summary collapse

Class Method Details

.acquire_exclusive_file_lock(handle, operation, target:, path:) ⇒ Object 



430
431
432
433
434
435
436
437
 
# File 'lib/cohere/transcribe/state/locking.rb', line 430

def acquire_exclusive_file_lock(handle, operation, target:, path:)
  handle.flock(operation)
rescue Errno::EBADF => e
  raise TranscriptionRuntimeError,
        "Cannot acquire an exclusive output lock for #{target.identity}: " \
        "the filesystem requires a writable lock file at #{path}",
        cause: e
end

.apply_lock_mode_if_supported(target, mode) ⇒ Object 



410
411
412
413
414
415
416
417
418
419
420
 
# File 'lib/cohere/transcribe/state/locking.rb', line 410

def apply_lock_mode_if_supported(target, mode)
  target.chmod(mode)
rescue SystemCallError => e
  unsupported = [Errno::EPERM::Errno]
  unsupported << Errno::EROFS::Errno
  unsupported << Errno::EOPNOTSUPP::Errno if defined?(Errno::EOPNOTSUPP)
  unsupported << Errno::ENOTSUP::Errno if defined?(Errno::ENOTSUP)
  raise unless unsupported.include?(e.errno)

  nil
end

.asr_checkpoint_payload(result:, source_snapshot:, asr_contract_key:, speech_spans:, vad_provider_options:, generation_id:) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 57

def asr_checkpoint_payload(result:, source_snapshot:, asr_contract_key:, speech_spans:,
                           vad_provider_options:, generation_id:)
  provenance = result.provenance
  {
    "kind" => "asr_complete",
    "generation_id" => generation_id,
    "asr_contract_key" => asr_contract_key,
    "source" => source_snapshot.payload,
    "updated_unix_seconds" => Time.now.to_f,
    "checkpoint" => {
      "duration" => result.duration,
      "segment_times" => result.segments.map { |segment| [segment.start, segment.end] },
      "speech_spans" => speech_spans.map { |start_time, end_time| [start_time, end_time] },
      "segment_texts" => result.segments.map(&:text),
      "generated_tokens" => provenance.generated_tokens_by_segment.sort,
      "repetition_stopped_segments" => provenance.repetition_stopped_segments.sort,
      "truncation_retried_segments" => provenance.truncation_retried_segments.sort,
      "token_limit_segments" => provenance.token_limit_segments.sort,
      "decode_backend" => provenance.decode_backend,
      "decode_fallback_reason" => provenance.decode_fallback_reason,
      "vad_engine_actual" => provenance.vad_engine_actual,
      "vad_provider" => provenance.vad_provider,
      "vad_provider_options" => vad_provider_options,
      "vad_fallback_reason" => provenance.vad_fallback_reason
    }
  }
end

.asr_contract_key(options, model_format: "dense", model_quantization: nil) ⇒ Object 



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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 39

def asr_contract_key(options, model_format: "dense", model_quantization: nil)
  silero = options.vad == "silero"
  auditok = options.vad == "auditok"
  model_id = json_value(options.model)
  model_revision = ModelIdentity.default_model_revision(model_id, options.model_revision)
  configuration = {
    "language" => options.language,
    "device" => options.device,
    "dtype" => options.dtype,
    "audio_backend" => options.audio_backend,
    "vad" => options.vad,
    "vad_engine" => silero ? options.vad_engine : nil,
    "vad_batch_size" => silero ? options.vad_batch_size : nil,
    "vad_block_frames" => silero ? options.vad_block_frames : nil,
    "vad_threads" => silero ? options.vad_threads : nil,
    "vad_merge" => silero ? options.vad_merge : nil,
    "min_dur" => options.vad == "none" ? nil : options.min_dur,
    "max_dur" => options.max_dur,
    "max_silence" => auditok ? options.max_silence : nil,
    "energy_threshold" => auditok ? options.energy_threshold : nil,
    "vad_threshold" => silero ? options.vad_threshold : nil,
    "min_silence_ms" => silero ? options.min_silence_ms : nil,
    "speech_pad_ms" => silero ? options.speech_pad_ms : nil,
    "batch_size" => options.batch_size,
    "batch_max_size" => options.batch_max_size,
    "batch_audio_seconds" => options.batch_audio_seconds,
    "batch_vram_target" => options.batch_vram_target,
    "adaptive_batch" => options.adaptive_batch,
    "pin_memory" => options.pin_memory,
    "audio_memory_gb" => options.audio_memory_gb,
    "pipeline_preparation" => options.pipeline_preparation,
    "max_new_tokens" => options.max_new_tokens,
    "max_retry_tokens" => options.max_retry_tokens,
    "truncation_policy" => options.truncation_policy,
    "stop_repetition_loops" => options.stop_repetition_loops
  }
  fingerprint(
    "contract_schema_version" => CONTRACT_SCHEMA_VERSION,
    "configuration" => configuration,
    "models" => {
      "asr" => {
        "id" => model_id,
        "revision" => model_revision,
        "format" => model_format,
        "quantization" => model_quantization,
        "adapter" => options.adapter && {
          "id" => json_value(options.adapter),
          "revision" => options.adapter_revision
        }
      },
      "silero_version" => "6.2.1"
    },
    "implementation_sha256" => implementation_fingerprint(ASR_IMPLEMENTATION_FILES)
  )
end

.canonical_json(value) ⇒ Object 



539
540
541
 
# File 'lib/cohere/transcribe/state/io.rb', line 539

def canonical_json(value)
  JSON.generate(deep_sort(value))
end

.canonical_lock_path(identity) ⇒ Object 



264
265
266
267
 
# File 'lib/cohere/transcribe/state/locking.rb', line 264

def canonical_lock_path(identity)
  identity = Pathname(identity).expand_path.cleanpath
  lock_registry_directory(identity.dirname).join(lock_filename(identity.to_s))
end

.checkpoint_path_for_outputs(output_paths) ⇒ Object 



534
535
536
537
 
# File 'lib/cohere/transcribe/state/io.rb', line 534

def checkpoint_path_for_outputs(output_paths)
  parent, stem = output_parent_and_stem(output_paths)
  parent.join(".#{stem}#{CHECKPOINT_SUFFIX}")
end

.close_atomic_resources(resources, recovery_errors, label:) ⇒ Object 



402
403
404
405
406
407
408
409
410
411
 
# File 'lib/cohere/transcribe/state/io.rb', line 402

def close_atomic_resources(resources, recovery_errors, label:)
  failures = []
  Array(resources).compact.uniq.each do |resource|
    resource.close unless resource.respond_to?(:closed?) && resource.closed?
  rescue SystemCallError, IOError, TranscriptionRuntimeError => e
    recovery_errors << "close #{label.call(resource)}: #{e.message}"
    failures << e
  end
  failures
end

.create_shared_lock_directory(path) ⇒ Object 



372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
 
# File 'lib/cohere/transcribe/state/locking.rb', line 372

def create_shared_lock_directory(path)
  parent = path.dirname.lstat
  unless parent.directory? && !parent.symlink?
    raise TranscriptionRuntimeError, "Output lock parent is not a real directory: #{path.dirname}"
  end

  mode = (parent.mode & 0o3777) | 0o700
  mode |= 0o1000 if mode.anybits?(0o022)
  begin
    Dir.mkdir(path, mode)
    apply_lock_mode_if_supported(path, mode)
  rescue Errno::EEXIST
    existing = path.lstat
    owned = Process.respond_to?(:uid) && existing.uid == Process.uid
    mode_can_be_updated = !Gem.win_platform? && existing.directory? && owned
    apply_lock_mode_if_supported(path, mode) if mode_can_be_updated && (existing.mode & 0o3777) != mode
  rescue SystemCallError => e
    raise TranscriptionRuntimeError, "Cannot prepare output lock directory #{path}: #{e.message}"
  end
end

.decode_state(path, directory_binding: nil, guard_bindings: nil) ⇒ Object 



551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
 
# File 'lib/cohere/transcribe/state/io.rb', line 551

def decode_state(path, directory_binding: nil, guard_bindings: nil)
  path = Pathname(path)
  contents = securely_read_regular_file(
    path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  decoded = JSON.parse(contents)
  return [nil, "state marker root is not an object"] unless decoded.is_a?(Hash)
  return [nil, "state marker schema is unsupported"] unless decoded["schema_version"] == STATE_SCHEMA_VERSION

  payload = decoded["payload"]
  return [nil, "state marker payload is not an object"] unless payload.is_a?(Hash)

  expected = Digest::SHA256.hexdigest(canonical_json(payload))
  return [nil, "state marker integrity check failed"] unless secure_digest_equal?(decoded["payload_sha256"], expected)

  [payload, nil]
rescue Errno::ENOENT
  [nil, "state marker is missing or not a regular file"]
rescue JSON::ParserError, JSON::GeneratorError, EncodingError, SystemCallError,
       TranscriptionRuntimeError, TypeError, ArgumentError => e
  [nil, "state marker is unreadable (#{e.class}: #{e.message})"]
end

.deep_freeze_json(value) ⇒ Object 



211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 211

def deep_freeze_json(value)
  case value
  when Hash
    value.to_h { |key, item| [key.to_s.freeze, deep_freeze_json(item)] }.freeze
  when Array
    value.map { |item| deep_freeze_json(item) }.freeze
  when String
    value.dup.freeze
  else
    value
  end
end

.deep_sort(value) ⇒ Object 



820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
 
# File 'lib/cohere/transcribe/state/io.rb', line 820

def deep_sort(value)
  case value
  when Hash
    value.keys.map(&:to_s).sort.to_h do |key|
      original_key = value.key?(key) ? key : value.keys.find { |candidate| candidate.to_s == key }
      [key, deep_sort(value.fetch(original_key))]
    end
  when Array
    value.map { |item| deep_sort(item) }
  when Pathname
    value.to_s
  else
    value
  end
end

.ensure_bound_directory(path, root_binding: nil, operation_hook: nil) ⇒ Object 



451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
 
# File 'lib/cohere/transcribe/state/io.rb', line 451

def ensure_bound_directory(path, root_binding: nil, operation_hook: nil)
  target = Pathname(path).expand_path.cleanpath
  bindings = []
  current = nil
  if root_binding
    root_binding.verify!
    relative = target.relative_path_from(root_binding.canonical_path)
    if relative.each_filename.first == ".."
      raise TranscriptionRuntimeError,
            "Publication directory escapes its bound root: #{target}"
    end
    current_binding = root_binding
    access_cursor = root_binding.canonical_path
    canonical_cursor = root_binding.canonical_path
    components = relative.each_filename.reject { |name| name == "." }
  else
    missing = []
    cursor = target
    until cursor.exist? || cursor.symlink?
      parent = cursor.dirname
      raise TranscriptionRuntimeError, "Cannot resolve publication directory #{target}" if parent == cursor

      missing.unshift(cursor.basename.to_s)
      cursor = parent
    end
    current_binding = DirectoryBinding.capture(cursor)
    access_cursor = current_binding.access_path
    canonical_cursor = current_binding.canonical_path
    components = missing
  end

  Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
    current = BoundDirectory.open(current_binding, guards: [current_binding])
    bindings << current
  end
  operation_hook&.call(:ancestor_opened, current.binding.access_path)
  components.each do |component|
    operation_hook&.call(:before_mkdir, access_cursor.join(component))
    current.mkdir(component)
    operation_hook&.call(:after_mkdir, access_cursor.join(component))
    access_cursor = access_cursor.join(component)
    canonical_cursor = canonical_cursor.join(component)
    Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
      current = current.open_child_directory(
        component,
        access_path: access_cursor,
        canonical_path: canonical_cursor
      )
      bindings << current
    end
  end
  bindings.each(&:verify!)
  current.binding
rescue ArgumentError => e
  raise TranscriptionRuntimeError, "Cannot bind publication directory #{path}: #{e.message}"
ensure
  bindings&.reverse_each(&:close)
end

.ensure_lock_acquisition_allowed!(target) ⇒ Object

Raises:



481
482
483
484
485
486
 
# File 'lib/cohere/transcribe/state/locking.rb', line 481

def ensure_lock_acquisition_allowed!(target)
  return unless lock_acquisition_unwinding?

  raise TranscriptionRuntimeError,
        "Cannot acquire output lock for #{target.identity} while the current thread is terminating"
end

.ensure_source_unchanged!(snapshot) ⇒ Object 



747
748
749
750
751
752
753
754
 
# File 'lib/cohere/transcribe/state/io.rb', line 747

def ensure_source_unchanged!(snapshot)
  return if SourceSnapshot.capture(snapshot.canonical_path) == snapshot

  raise TranscriptionRuntimeError, "Source changed while processing: #{snapshot.canonical_path}"
rescue SystemCallError, ArgumentError => e
  raise TranscriptionRuntimeError,
        "Cannot re-check source #{snapshot.canonical_path}: #{e.message}"
end

.envelope(payload) ⇒ Object 



543
544
545
546
547
548
549
 
# File 'lib/cohere/transcribe/state/io.rb', line 543

def envelope(payload)
  {
    "schema_version" => STATE_SCHEMA_VERSION,
    "payload_sha256" => Digest::SHA256.hexdigest(canonical_json(payload)),
    "payload" => payload
  }
end

.fingerprint(value) ⇒ Object 



132
133
134
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 132

def fingerprint(value)
  Digest::SHA256.hexdigest(canonical_json(value))
end

.finish_atomic_recovery!(subject:, completed:, primary_error:, recovery_errors:, retained_backups:) ⇒ Object

Raises:



374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
 
# File 'lib/cohere/transcribe/state/io.rb', line 374

def finish_atomic_recovery!(subject:, completed:, primary_error:, recovery_errors:,
                            retained_backups:)
  return if recovery_errors.empty?
  # Raising from an ensure during Thread#kill replaces termination with a
  # catchable exception, so cleanup remains best effort while aborting.
  return if Thread.current.status == "aborting"

  outcome = if completed
              "completed but cleanup was incomplete"
            elsif primary_error
              "failed and recovery was incomplete"
            else
              "did not complete and recovery was incomplete"
            end
  detail = recovery_errors.join("; ")
  retained = retained_backups.compact.map(&:to_s).uniq.sort
  retained_detail = if retained.empty?
                      ""
                    elsif retained.one?
                      "; preserved backup: #{retained.first}"
                    else
                      "; preserved backups: #{retained}"
                    end
  raise TranscriptionRuntimeError,
        "#{subject} #{outcome} (#{detail})#{retained_detail}",
        cause: primary_error
end

.finite_number(value, name) ⇒ Object 



198
199
200
201
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 198

def finite_number(value, name)
  validate!(value.is_a?(Numeric) && !value.is_a?(Complex) && value.finite?, "#{name} is not finite numeric data")
  value.to_f
end

.immutable_optional_string(value) ⇒ Object 



207
208
209
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 207

def immutable_optional_string(value)
  value&.dup&.freeze
end

.implementation_fingerprint(files) ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 116

def implementation_fingerprint(files)
  root = Pathname(__dir__).parent
  missing = files.reject { |relative| root.join(relative).file? }
  unless missing.empty?
    raise TranscriptionRuntimeError,
          "Implementation fingerprint references missing package artifacts: #{missing.join(", ")}"
  end

  fingerprint(
    "package_version" => VERSION,
    "artifacts_sha256" => files.to_h do |relative|
      [relative, Digest::SHA256.file(root.join(relative)).hexdigest]
    end
  )
end

.inspect_lock_path!(path) ⇒ Object 



498
499
500
501
502
503
504
505
 
# File 'lib/cohere/transcribe/state/locking.rb', line 498

def inspect_lock_path!(path)
  stat = path.lstat
  return if stat.file? && !stat.symlink?

  raise TranscriptionRuntimeError, "Output lock is not a regular file: #{path}"
rescue Errno::ENOENT
  nil
end

.json_value(value) ⇒ Object 



136
137
138
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 136

def json_value(value)
  value.is_a?(Pathname) ? value.to_s : value
end

.legacy_lock_compatibility_enabled?(version = VERSION) ⇒ Boolean

Returns:

  • (Boolean) 


260
261
262
 
# File 'lib/cohere/transcribe/state/locking.rb', line 260

def legacy_lock_compatibility_enabled?(version = VERSION)
  Gem::Version.new(version) < Gem::Version.new(LEGACY_LOCK_REMOVAL_VERSION)
end

.legacy_temporary_lock_path(identity) ⇒ Object 



273
274
275
276
 
# File 'lib/cohere/transcribe/state/locking.rb', line 273

def legacy_temporary_lock_path(identity)
  scope = Process.respond_to?(:uid) ? Process.uid.to_s : Digest::SHA256.hexdigest(Dir.home)[0, 16]
  Pathname(Dir.tmpdir).expand_path.join("#{LOCK_DIRECTORY_PREFIX}-#{scope}", lock_filename(identity))
end

.lock_acquisition_unwinding? ⇒ Boolean

Ruby exposes an in-progress Thread#kill unwind only through #status; pending_interrupt? is already false while ensure blocks are running.

Returns:

  • (Boolean) 


490
491
492
 
# File 'lib/cohere/transcribe/state/locking.rb', line 490

def lock_acquisition_unwinding?
  Thread.current.status == "aborting"
end

.lock_filename(identity) ⇒ Object 



278
279
280
 
# File 'lib/cohere/transcribe/state/locking.rb', line 278

def lock_filename(identity)
  "#{Digest::SHA256.hexdigest(identity.downcase)}#{LOCK_SUFFIX}"
end

.lock_path_specs_for_target(target) ⇒ Object 



243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/cohere/transcribe/state/locking.rb', line 243

def lock_path_specs_for_target(target)
  primary = Pathname(target.path).expand_path.cleanpath
  canonical = canonical_lock_path(target.identity)
  legacy = legacy_temporary_lock_path(target.identity)
  unless primary == canonical
    role = primary == legacy ? :legacy : :custom
    return [LockPath.new(path: primary.freeze, role: role)].freeze
  end

  # Remove this temporary-registry compatibility path in 0.2.0. Until
  # then, acquiring both paths keeps released 0.1.2 processes coordinated
  # with output-adjacent locks.
  paths = [LockPath.new(path: primary.freeze, role: :canonical)]
  paths << LockPath.new(path: legacy.freeze, role: :legacy) if legacy_lock_compatibility_enabled? && legacy != primary
  paths.freeze
end

.lock_paths_for_target(target) ⇒ Object 



239
240
241
 
# File 'lib/cohere/transcribe/state/locking.rb', line 239

def lock_paths_for_target(target)
  lock_path_specs_for_target(target).map(&:path).freeze
end

.lock_registry_directory(output_parent) ⇒ Object 



269
270
271
 
# File 'lib/cohere/transcribe/state/locking.rb', line 269

def lock_registry_directory(output_parent)
  Pathname(output_parent).expand_path.cleanpath.join(LOCK_DIRECTORY_NAME)
end

.lock_target_for_outputs(output_paths) ⇒ Object 



206
207
208
209
210
211
212
213
 
# File 'lib/cohere/transcribe/state/locking.rb', line 206

def lock_target_for_outputs(output_paths)
  parent, stem = output_parent_and_stem(output_paths)
  identity = parent.join(stem).expand_path.cleanpath.to_s
  OutputLockTarget.new(
    path: canonical_lock_path(identity).freeze,
    identity: identity.freeze
  )
end

.missing_lock_path_error?(error) ⇒ Boolean

Returns:

  • (Boolean) 


494
495
496
 
# File 'lib/cohere/transcribe/state/locking.rb', line 494

def missing_lock_path_error?(error)
  error.cause.is_a?(Errno::ENOENT)
end

.open_lock_file(path, shared: false) ⇒ Object 



282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
 
# File 'lib/cohere/transcribe/state/locking.rb', line 282

def open_lock_file(path, shared: false)
  path = Pathname(path)
  succeeded = false
  created = false
  validate_lock_directory!(path.dirname, shared: shared)
  inspect_lock_path!(path)
  flags = File::RDWR
  flags |= File::NOFOLLOW if defined?(File::NOFOLLOW)
  flags |= File::CLOEXEC if defined?(File::CLOEXEC)
  mode = shared ? shared_lock_file_mode(path.dirname.lstat) : 0o600
  begin
    descriptor, created = open_or_create_lock_descriptor(path, flags, mode)
    access = "r+"
  rescue Errno::EACCES, Errno::EROFS => e
    raise unless shared

    handle, access = open_shared_lock_after_write_denial(path, flags, mode, e)
  end
  unless handle
    handle = File.new(descriptor, access, autoclose: true)
    descriptor = nil
  end
  opened = verify_lock_identity!(
    path,
    handle,
    message: "Output lock changed while it was being opened or is not regular"
  )
  if shared
    owned_writable = access == "r+" && Process.respond_to?(:uid) && opened.uid == Process.uid
    apply_lock_mode_if_supported(handle, mode) if (created || owned_writable) && (opened.mode & 0o777) != mode
  else
    validate_owned_private_file!(opened, path)
  end
  succeeded = true
  handle
rescue Errno::ELOOP, Errno::EISDIR, Errno::ENXIO => e
  raise TranscriptionRuntimeError, "Output lock is not a regular file: #{path}", cause: e
ensure
  if !succeeded && handle
    handle.close
  elsif descriptor
    ::IO.new(descriptor).close
  end
end

.open_or_create_lock_descriptor(path, flags, mode) ⇒ Object

Raises:

  • (Errno::ENOENT) 


327
328
329
330
331
332
333
334
335
336
337
338
339
 
# File 'lib/cohere/transcribe/state/locking.rb', line 327

def open_or_create_lock_descriptor(path, flags, mode)
  2.times do
    return [::IO.sysopen(path.to_s, flags, mode), false]
  rescue Errno::ENOENT
    begin
      return [::IO.sysopen(path.to_s, flags | File::CREAT | File::EXCL, mode), true]
    rescue Errno::EEXIST
      next
    end
  end

  raise Errno::ENOENT, path.to_s
end

.open_shared_lock_after_write_denial(path, flags, mode, original_error) ⇒ Object 



439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
 
# File 'lib/cohere/transcribe/state/locking.rb', line 439

def open_shared_lock_after_write_denial(path, flags, mode, original_error)
  descriptor = ::IO.sysopen(path.to_s, readonly_lock_flags)
  readonly = File.new(descriptor, "r", autoclose: true)
  descriptor = nil
  opened = verify_lock_identity!(
    path,
    readonly,
    message: "Output lock changed while opening it read-only"
  )
  if original_error.is_a?(Errno::EACCES) &&
     Process.respond_to?(:uid) && opened.uid == Process.uid
    apply_lock_mode_if_supported(readonly, mode)
    begin
      descriptor = ::IO.sysopen(path.to_s, flags, mode)
      writable = File.new(descriptor, "r+", autoclose: true)
      descriptor = nil
      readonly.close
      readonly = nil
      handle = writable
      writable = nil
      return [handle, "r+"]
    rescue Errno::EACCES, Errno::EROFS
      nil
    end
  end

  handle = readonly
  readonly = nil
  [handle, "r"]
ensure
  readonly&.close unless readonly&.closed?
  writable&.close unless writable&.closed?
  ::IO.new(descriptor).close if descriptor
end

.output_parent_and_stem(output_paths) ⇒ Object

Raises:

  • (ArgumentError) 


514
515
516
517
518
519
520
521
522
523
524
525
526
527
 
# File 'lib/cohere/transcribe/state/io.rb', line 514

def output_parent_and_stem(output_paths)
  raise ArgumentError, "An output set must contain at least one path" if output_paths.empty?

  paths = output_paths.values.map { |path| Pathname(path) }
  parent = paths.first.dirname
  stem = paths.first.basename(paths.first.extname).to_s
  unless paths.drop(1).all? do |path|
    path.dirname == parent && path.basename(path.extname).to_s == stem
  end
    raise ArgumentError, "All formats in one output set must share a parent and stem"
  end

  [parent, stem]
end

.publication_mismatch_reason(payload, source_snapshot:, output_paths:, asr_contract_key:, render_contract_key:, directory_binding:, guard_bindings: nil) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 94

def publication_mismatch_reason(payload, source_snapshot:, output_paths:, asr_contract_key:,
                                render_contract_key:, directory_binding:, guard_bindings: nil)
  return "state is #{payload["kind"].inspect}, not published" unless payload["kind"] == "published"
  return "state marker ASR contract does not match" unless payload["asr_contract_key"] == asr_contract_key
  return "state marker render contract does not match" unless payload["render_contract_key"] == render_contract_key
  return "state marker source snapshot does not match" unless payload["source"] == source_snapshot.payload

  generation_id = payload["generation_id"]
  return "state marker generation ID is invalid" unless generation_id.is_a?(String) && !generation_id.empty?

  records = payload["outputs"]
  return "state marker output formats do not match" unless records.is_a?(Hash) && records.keys.sort == output_paths.keys.sort

  output_paths.each do |format, path_value|
    path = Pathname(path_value)
    record = records[format]
    return "state marker path for #{format} does not match" unless record.is_a?(Hash) && record["name"] == path.basename.to_s

    begin
      output_record = bound_output_record(
        path,
        directory_binding: directory_binding,
        guard_bindings: guard_bindings
      )
      return "#{format} output changed or is not regular" unless output_record

      size, sha256 = output_record
      return "#{format} output does not match its state marker" if record["size"] != size || record["sha256"] != sha256
    rescue Errno::ENOENT
      return "#{format} output is missing or not regular"
    end
  end
  nil
end

.published_manifest_content(source_snapshot:, output_paths:, contents:, asr_contract_key:, render_contract_key:, generation_id:) ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 16

def published_manifest_content(source_snapshot:, output_paths:, contents:, asr_contract_key:,
                               render_contract_key:, generation_id:)
  payload = published_payload(
    source_snapshot: source_snapshot,
    output_paths: output_paths,
    contents: contents,
    asr_contract_key: asr_contract_key,
    render_contract_key: render_contract_key,
    generation_id: generation_id
  )
  JSON.pretty_generate(envelope(payload)) << "\n"
end

.published_payload(source_snapshot:, output_paths:, contents:, asr_contract_key:, render_contract_key:, generation_id:) ⇒ Object

Raises:

  • (ArgumentError) 


29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 29

def published_payload(source_snapshot:, output_paths:, contents:, asr_contract_key:,
                      render_contract_key:, generation_id:)
  raise ArgumentError, "generation ID is invalid" unless generation_id.is_a?(String) && !generation_id.empty?

  {
    "kind" => "published",
    "generation_id" => generation_id,
    "asr_contract_key" => asr_contract_key,
    "render_contract_key" => render_contract_key,
    "source" => source_snapshot.payload,
    "updated_unix_seconds" => Time.now.to_f,
    "outputs" => output_paths.keys.sort.to_h do |format|
      content = contents.fetch(format)
      [
        format,
        {
          "name" => Pathname(output_paths.fetch(format)).basename.to_s,
          "size" => content.bytesize,
          "sha256" => Digest::SHA256.hexdigest(content)
        }
      ]
    end
  }
end

.readonly_lock_flags ⇒ Object 



474
475
476
477
478
479
 
# File 'lib/cohere/transcribe/state/locking.rb', line 474

def readonly_lock_flags
  readonly_flags = File::RDONLY
  readonly_flags |= File::NOFOLLOW if defined?(File::NOFOLLOW)
  readonly_flags |= File::CLOEXEC if defined?(File::CLOEXEC)
  readonly_flags
end

.refresh_legacy_lock(handle) ⇒ Object 



422
423
424
425
426
427
428
 
# File 'lib/cohere/transcribe/state/locking.rb', line 422

def refresh_legacy_lock(handle)
  handle.rewind
  handle.write("#{Process.pid}\n")
  handle.flush
  handle.truncate(handle.pos)
  nil
end

.render_contract_key(options) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 95

def render_contract_key(options)
  formats = options.publication&.formats || []
  fingerprint(
    "contract_schema_version" => CONTRACT_SCHEMA_VERSION,
    "configuration" => {
      "alignment" => options.alignment,
      "align_batch_size" => options.align_batch_size,
      "align_dtype" => options.align_dtype,
      "formats" => formats.sort,
      "max_chars" => options.max_chars,
      "max_cue_dur" => options.max_cue_dur,
      "max_gap" => options.max_gap
    },
    "models" => {
      "align_revision" => options.alignment == "word" ? Alignment::ModelProvider::REVISION : nil
    },
    "output_schema_version" => Output::Publication::OUTPUT_SCHEMA_VERSION,
    "implementation_sha256" => implementation_fingerprint(RENDER_IMPLEMENTATION_FILES)
  )
end

.restore_asr_checkpoint(path:, source_snapshot:, asr_contract_key:, directory_binding: nil, guard_bindings: nil) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 85

def restore_asr_checkpoint(path:, source_snapshot:, asr_contract_key:,
                           directory_binding: nil, guard_bindings: nil)
  payload, reason = decode_state(
    path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  return CheckpointRestore.new(checkpoint: nil, reason: reason.freeze) unless payload
  unless payload["kind"] == "asr_complete"
    return CheckpointRestore.new(
      checkpoint: nil,
      reason: "state is #{payload["kind"].inspect}, not an ASR checkpoint".freeze
    )
  end
  unless payload["asr_contract_key"] == asr_contract_key
    return CheckpointRestore.new(checkpoint: nil, reason: "ASR checkpoint contract does not match")
  end
  unless payload["source"] == source_snapshot.payload
    return CheckpointRestore.new(checkpoint: nil, reason: "state marker source snapshot does not match")
  end

  CheckpointRestore.new(checkpoint: validate_checkpoint(payload).freeze, reason: nil)
rescue TypeError, ArgumentError => e
  CheckpointRestore.new(checkpoint: nil, reason: "ASR checkpoint is invalid (#{e.message})".freeze)
end

.same_file?(first, second) ⇒ Boolean

Returns:

  • (Boolean) 


808
809
810
 
# File 'lib/cohere/transcribe/state/io.rb', line 808

def same_file?(first, second)
  first.dev == second.dev && first.ino == second.ino && first.file? && second.file?
end

.secure_digest_equal?(actual, expected) ⇒ Boolean

Returns:

  • (Boolean) 


812
813
814
815
816
817
818
 
# File 'lib/cohere/transcribe/state/io.rb', line 812

def secure_digest_equal?(actual, expected)
  return false unless actual.is_a?(String) && actual.bytesize == expected.bytesize

  difference = 0
  actual.bytes.zip(expected.bytes) { |left, right| difference |= left ^ right }
  difference.zero?
end

.securely_read_regular_file(path, directory_binding: nil, guard_bindings: nil) ⇒ Object 



756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
 
# File 'lib/cohere/transcribe/state/io.rb', line 756

def securely_read_regular_file(path, directory_binding: nil, guard_bindings: nil)
  if directory_binding
    return with_bound_parent(
      path,
      directory_binding: directory_binding,
      guard_bindings: guard_bindings
    ) do |bound, basename|
      bound.verify!
      handle = nil
      Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
        handle = bound.open_regular(basename)
      end
      opened = handle.stat
      raise TranscriptionRuntimeError, "State marker is too large: #{path}" if opened.size > MAX_STATE_BYTES

      contents = handle.read(MAX_STATE_BYTES + 1)
      raise TranscriptionRuntimeError, "State marker is too large: #{path}" if contents.bytesize > MAX_STATE_BYTES
      unless bound.same_regular_entry?(basename, opened)
        raise TranscriptionRuntimeError, "State marker changed while it was being read: #{path}"
      end

      bound.verify!
      contents
    ensure
      handle&.close
    end
  end

  before = path.lstat
  raise TranscriptionRuntimeError, "State marker is not a regular file: #{path}" unless before.file? && !before.symlink?
  raise TranscriptionRuntimeError, "State marker is too large: #{path}" if before.size > MAX_STATE_BYTES

  flags = File::RDONLY
  flags |= File::NOFOLLOW if defined?(File::NOFOLLOW)
  flags |= File::CLOEXEC if defined?(File::CLOEXEC)
  descriptor = ::IO.sysopen(path.to_s, flags)
  handle = ::IO.new(descriptor, "rb", autoclose: true)
  descriptor = nil
  opened = handle.stat
  current = path.lstat
  unless opened.file? && same_file?(opened, before) && same_file?(opened, current)
    raise TranscriptionRuntimeError, "State marker changed while it was being opened: #{path}"
  end

  handle.read(MAX_STATE_BYTES + 1).tap do |contents|
    raise TranscriptionRuntimeError, "State marker is too large: #{path}" if contents.bytesize > MAX_STATE_BYTES
  end
ensure
  handle&.close
  ::IO.new(descriptor).close if descriptor
end

.shared_lock_file_mode(directory_stat) ⇒ Object 



403
404
405
406
407
408
 
# File 'lib/cohere/transcribe/state/locking.rb', line 403

def shared_lock_file_mode(directory_stat)
  mode = 0o600
  mode |= 0o060 if directory_stat.mode.anybits?(0o020) && directory_stat.mode.anybits?(0o010)
  mode |= 0o006 if directory_stat.mode.anybits?(0o002) && directory_stat.mode.anybits?(0o001)
  mode
end

.state_path_for_outputs(output_paths) ⇒ Object 



529
530
531
532
 
# File 'lib/cohere/transcribe/state/io.rb', line 529

def state_path_for_outputs(output_paths)
  parent, stem = output_parent_and_stem(output_paths)
  parent.join(".#{stem}#{STATE_SUFFIX}")
end

.time_nanoseconds(time) ⇒ Object 



510
511
512
 
# File 'lib/cohere/transcribe/state/io.rb', line 510

def time_nanoseconds(time)
  (time.to_i * 1_000_000_000) + time.nsec
end

.validate!(condition, message) ⇒ Object

Raises:

  • (ArgumentError) 


203
204
205
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 203

def validate!(condition, message)
  raise ArgumentError, message unless condition
end

.validate_checkpoint(payload) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 111

def validate_checkpoint(payload)
  generation_id = payload["generation_id"]
  validate!(generation_id.is_a?(String) && !generation_id.empty?, "generation ID is invalid")
  checkpoint = payload["checkpoint"]
  validate!(checkpoint.is_a?(Hash), "checkpoint is not an object")

  duration = finite_number(checkpoint["duration"], "duration")
  validate!(duration >= 0, "duration is invalid")
  segment_times = validated_spans(checkpoint["segment_times"], duration, "segment_times")
  speech_spans = validated_spans(checkpoint["speech_spans"], duration, "speech_spans")
  texts = checkpoint["segment_texts"]
  validate!(texts.is_a?(Array) && texts.all?(String), "segment_texts is invalid")
  validate!(texts.length == segment_times.length, "segment text/time counts differ")
  texts = texts.map { |text| text.dup.freeze }.freeze
  segment_count = segment_times.length

  generated_tokens = validated_generated_tokens(checkpoint.fetch("generated_tokens", []), segment_count)
  repetition = validated_indices(checkpoint.fetch("repetition_stopped_segments", []), segment_count,
                                 "repetition_stopped_segments")
  truncation = validated_indices(checkpoint.fetch("truncation_retried_segments", []), segment_count,
                                 "truncation_retried_segments")
  token_limit = validated_indices(checkpoint.fetch("token_limit_segments", []), segment_count,
                                  "token_limit_segments")
  optional_names = %w[
    decode_backend decode_fallback_reason vad_engine_actual vad_provider vad_fallback_reason
  ]
  optional_names.each do |name|
    value = checkpoint[name]
    validate!(value.nil? || value.is_a?(String), "checkpoint provenance strings are invalid")
  end
  provider_options = checkpoint["vad_provider_options"]
  validate!(provider_options.nil? || provider_options.is_a?(Hash), "vad_provider_options is invalid")

  Checkpoint.new(
    generation_id: generation_id.dup.freeze,
    duration: duration,
    segment_times: segment_times,
    speech_spans: speech_spans,
    segment_texts: texts,
    generated_tokens_by_segment: generated_tokens,
    repetition_stopped_segments: repetition,
    truncation_retried_segments: truncation,
    token_limit_segments: token_limit,
    decode_backend: immutable_optional_string(checkpoint["decode_backend"]),
    decode_fallback_reason: immutable_optional_string(checkpoint["decode_fallback_reason"]),
    vad_engine_actual: immutable_optional_string(checkpoint["vad_engine_actual"]),
    vad_provider: immutable_optional_string(checkpoint["vad_provider"]),
    vad_provider_options: deep_freeze_json(provider_options),
    vad_fallback_reason: immutable_optional_string(checkpoint["vad_fallback_reason"])
  )
end

.validate_lock_directory!(path, shared: false) ⇒ Object 



341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
 
# File 'lib/cohere/transcribe/state/locking.rb', line 341

def validate_lock_directory!(path, shared: false)
  if shared
    create_shared_lock_directory(path)
  else
    begin
      FileUtils.mkdir_p(path, mode: 0o700)
    rescue SystemCallError => e
      raise TranscriptionRuntimeError, "Cannot prepare output lock directory #{path}: #{e.message}"
    end
  end
  stat = path.lstat
  raise TranscriptionRuntimeError, "Output lock directory is not a real directory: #{path}" unless stat.directory? && !stat.symlink?

  if shared
    validate_shared_lock_directory_mode!(path, stat)
    return
  end
  return if Gem.win_platform?

  if Process.respond_to?(:uid) && stat.uid != Process.uid
    raise TranscriptionRuntimeError,
          "Output lock directory is not owned by the current user: #{path}"
  end
  return if stat.mode.nobits?(0o077)

  raise TranscriptionRuntimeError,
        "Output lock directory permissions must be private (0700): #{path}"
rescue Errno::ENOENT, Errno::ENOTDIR => e
  raise TranscriptionRuntimeError, "Cannot prepare output lock directory #{path}: #{e.message}"
end

.validate_owned_private_file!(stat, path) ⇒ Object

Raises:



507
508
509
510
511
512
513
514
515
516
517
 
# File 'lib/cohere/transcribe/state/locking.rb', line 507

def validate_owned_private_file!(stat, path)
  return if Gem.win_platform?

  if Process.respond_to?(:uid) && stat.uid != Process.uid
    raise TranscriptionRuntimeError, "Output lock is not owned by the current user: #{path}"
  end
  return if stat.mode.nobits?(0o077)

  raise TranscriptionRuntimeError,
        "Output lock permissions must be private (0600): #{path}"
end

.validate_shared_lock_directory_mode!(path, stat) ⇒ Object

Raises:



393
394
395
396
397
398
399
400
401
 
# File 'lib/cohere/transcribe/state/locking.rb', line 393

def validate_shared_lock_directory_mode!(path, stat)
  return if Gem.win_platform? || !Process.respond_to?(:uid)
  return if stat.uid == Process.uid
  return if stat.mode.nobits?(0o022) || stat.mode.anybits?(0o1000)

  raise TranscriptionRuntimeError,
        "Output lock directory is writable by multiple users but has no sticky bit: #{path}. " \
        "Ask its owner or administrator to add the sticky bit, or remove it when no transcription job is running"
end

.validated_generated_tokens(value, segment_count) ⇒ Object 



178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 178

def validated_generated_tokens(value, segment_count)
  validate!(value.is_a?(Array), "generated_tokens is invalid")
  seen = {}
  value.map do |row|
    valid = row.is_a?(Array) && row.length == 2 && row.all?(Integer)
    valid &&= row[0].between?(0, segment_count - 1) && row[1] >= 0
    valid &&= !seen.key?(row[0])
    validate!(valid, "generated_tokens contains an invalid or duplicate row")
    seen[row[0]] = true
    [row[0], row[1]].freeze
  end.freeze
end

.validated_indices(value, segment_count, name) ⇒ Object 



191
192
193
194
195
196
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 191

def validated_indices(value, segment_count, name)
  validate!(value.is_a?(Array), "#{name} is invalid")
  valid = value.all? { |index| index.is_a?(Integer) && index.between?(0, segment_count - 1) }
  validate!(valid && value.uniq.length == value.length, "#{name} is invalid or contains duplicate indices")
  value.dup.freeze
end

.validated_spans(value, duration, name) ⇒ Object 



163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 163

def validated_spans(value, duration, name)
  validate!(value.is_a?(Array), "#{name} is not a list")
  previous_end = nil
  value.map do |row|
    validate!(row.is_a?(Array) && row.length == 2, "#{name} contains an invalid row")
    start_time = finite_number(row[0], name)
    end_time = finite_number(row[1], name)
    valid = start_time.between?(0, end_time) && end_time <= duration + 1e-6
    valid &&= previous_end.nil? || start_time >= previous_end
    validate!(valid, "#{name} contains an overlapping, out-of-order, or out-of-range row")
    previous_end = end_time
    [start_time, [end_time, duration].min].freeze
  end.freeze
end

.verify_lock_identity!(path, handle, message:) ⇒ Object 



519
520
521
522
523
524
525
526
527
 
# File 'lib/cohere/transcribe/state/locking.rb', line 519

def verify_lock_identity!(path, handle, message:)
  opened = handle.stat
  current = Pathname(path).lstat
  return opened if opened.file? && same_file?(opened, current)

  raise TranscriptionRuntimeError, "#{message}: #{path}", cause: nil
rescue SystemCallError => e
  raise TranscriptionRuntimeError, "#{message}: #{path}", cause: e
end

.verify_published_outputs(source_snapshot:, output_paths:, state_path:, asr_contract_key:, render_contract_key:, directory_binding:, guard_bindings: nil) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 54

def verify_published_outputs(source_snapshot:, output_paths:, state_path:, asr_contract_key:,
                             render_contract_key:, directory_binding:, guard_bindings: nil)
  payload, reason = decode_state(
    state_path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  return PublicationVerification.new(verified: false, generation_id: nil, reason: reason.freeze) unless payload

  mismatch = publication_mismatch_reason(
    payload,
    source_snapshot: source_snapshot,
    output_paths: output_paths,
    asr_contract_key: asr_contract_key,
    render_contract_key: render_contract_key,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  return PublicationVerification.new(verified: false, generation_id: nil, reason: mismatch.freeze) if mismatch

  verify_publication_bindings!(directory_binding, guard_bindings)
  PublicationVerification.new(
    verified: true,
    generation_id: payload.fetch("generation_id").dup.freeze,
    reason: nil
  )
rescue PublicationDirectoryChangedError => e
  PublicationVerification.new(
    verified: false,
    generation_id: nil,
    reason: "publication parent changed during verification (#{e.message})".freeze
  )
rescue EncodingError, SystemCallError, TypeError, ArgumentError => e
  PublicationVerification.new(
    verified: false,
    generation_id: nil,
    reason: "cannot verify output generation (#{e.class}: #{e.message})".freeze
  )
end

.with_bound_parent(path, directory_binding: nil, guard_bindings: nil, recovery_errors: nil) ⇒ Object 



413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
 
# File 'lib/cohere/transcribe/state/io.rb', line 413

def with_bound_parent(path, directory_binding: nil, guard_bindings: nil,
                      recovery_errors: nil)
  path = Pathname(path).expand_path
  directory_binding ||= DirectoryBinding.capture(path.dirname)
  unless [directory_binding.access_path, directory_binding.canonical_path].include?(path.dirname)
    raise TranscriptionRuntimeError,
          "Publication path is outside its planned parent: #{path}"
  end
  guards = Array(guard_bindings).compact
  guards << directory_binding
  bound = nil
  Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
    bound = BoundDirectory.open(directory_binding, guards: guards.uniq)
  end
  completed = false
  primary_error = nil
  begin
    result = yield bound, path.basename.to_s
    completed = true
    result
  rescue Exception => e # rubocop:disable Lint/RescueException -- preserve the active outcome through close
    primary_error = e
    raise
  end
ensure
  close_errors = recovery_errors || []
  close_atomic_resources([bound], close_errors, label: ->(resource) { resource.binding.canonical_path })
  unless recovery_errors
    finish_atomic_recovery!(
      subject: "Publication directory operation",
      completed: completed,
      primary_error: primary_error,
      recovery_errors: close_errors,
      retained_backups: []
    )
  end
end

.with_output_lock(target, blocking: false) ⇒ Object 



215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/cohere/transcribe/state/locking.rb', line 215

def with_output_lock(target, blocking: false)
  lock = nil
  primary_error = nil
  Thread.handle_interrupt(Object => :never) do
    Thread.handle_interrupt(Object => :on_blocking) do
      OutputSetLock.acquire(
        target,
        blocking: blocking,
        ownership_hook: ->(acquired_lock) { lock = acquired_lock }
      )
    end
    Thread.handle_interrupt(Object => :immediate) { yield lock }
  rescue Exception => e # rubocop:disable Lint/RescueException -- preserve protected-work failures
    primary_error = e
    raise
  ensure
    begin
      lock&.release
    rescue Exception # rubocop:disable Lint/RescueException -- preserve protected-work failures
      raise unless primary_error || lock_acquisition_unwinding?
    end
  end
end

.write_asr_checkpoint(path:, result:, source_snapshot:, asr_contract_key:, speech_spans:, vad_provider_options: nil, generation_id: nil, directory_binding: nil, guard_bindings: nil, lock: nil) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 33

def write_asr_checkpoint(path:, result:, source_snapshot:, asr_contract_key:,
                         speech_spans:, vad_provider_options: nil, generation_id: nil,
                         directory_binding: nil, guard_bindings: nil, lock: nil)
  generation_id = generation_id.to_s
  generation_id = SecureRandom.hex(16) if generation_id.empty?
  payload = asr_checkpoint_payload(
    result: result,
    source_snapshot: source_snapshot,
    asr_contract_key: asr_contract_key,
    speech_spans: speech_spans,
    vad_provider_options: vad_provider_options,
    generation_id: generation_id
  )
  write_state_atomic(
    path,
    payload,
    source_snapshot: source_snapshot,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings,
    commit_guard: lock&.method(:verify!)
  )
  generation_id.freeze
end

.write_state_atomic(path, payload, source_snapshot:, directory_binding: nil, guard_bindings: nil, operation_hook: nil, rename: nil, commit_guard: nil) ⇒ Object 



576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
 
# File 'lib/cohere/transcribe/state/io.rb', line 576

def write_state_atomic(path, payload, source_snapshot:, directory_binding: nil,
                       guard_bindings: nil, operation_hook: nil, rename: nil,
                       commit_guard: nil)
  path = Pathname(path).expand_path
  directory_binding ||= ensure_bound_directory(path.dirname)
  recovery_errors = []
  retained_backups = []
  completed = false
  primary_error = nil
  begin
    with_bound_parent(
      path,
      directory_binding: directory_binding,
      guard_bindings: guard_bindings,
      recovery_errors: recovery_errors
    ) do |bound, basename|
      operation_hook&.call(:directory_opened, path)
      temporary_name = nil
      temporary = nil
      backup_name = nil
      backup = nil
      source = nil
      committed = false
      begin
        mode = bound_state_mode(bound, basename)
        Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
          temporary_name, temporary = bound.create_temporary(basename, ".tmp")
        end
        temporary.chmod(mode)
        operation_hook&.call(:before_stage_write, path)
        temporary.write(JSON.pretty_generate(envelope(payload)))
        temporary.write("\n")
        temporary.flush
        temporary.fsync
        temporary.close
        temporary = nil
        bound.verify!

        operation_hook&.call(:before_backup_open, path)
        Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
          source = begin
            bound.open_regular(basename)
          rescue Errno::ENOENT
            nil
          end
        end
        if source
          Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
            backup_name, backup = bound.create_temporary(basename, ".bak")
          end
          backup_copy_error = nil
          begin
            backup.chmod(source.stat.mode & 0o7777)
            operation_hook&.call(:before_backup_copy, path)
            IO.copy_stream(source, backup)
            backup.flush
            backup.fsync
          rescue Exception => e # rubocop:disable Lint/RescueException -- retain the copy failure through close
            backup_copy_error = e
            raise
          ensure
            close_failures = close_atomic_resources(
              [source, backup],
              recovery_errors,
              label: ->(handle) { handle.respond_to?(:path) ? handle.path : path }
            )
            raise close_failures.first if close_failures.any? && backup_copy_error.nil?
          end
        end

        ensure_source_unchanged!(source_snapshot)
        bound.verify!
        operation_hook&.call(:before_rename, path)
        Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
          commit_guard&.call
          if rename
            committed = true
            rename.call(bound, temporary_name, basename, :commit)
          else
            bound.rename(temporary_name, basename)
            committed = true
          end
          operation_hook&.call(:after_rename, path)
          bound.verify!
          bound.fsync
          commit_guard&.call
          completed = true
        end
        path
      rescue Exception => e # rubocop:disable Lint/RescueException -- rollback includes interrupts
        primary_error = e
        raise
      ensure
        Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
          if !completed && committed
            begin
              bound.unlink(basename, missing_ok: true)
              if backup_name
                if rename
                  rename.call(bound, backup_name, basename, :rollback)
                else
                  bound.rename(backup_name, basename)
                end
                backup_name = nil
              end
            rescue StandardError => e
              retained_backups << bound.display_path(backup_name) if backup_name
              recovery_errors << "#{path}: #{e.message}"
            end
            begin
              bound.fsync
            rescue SystemCallError, TranscriptionRuntimeError => e
              recovery_errors << "directory sync: #{e.message}"
            end
          end

          close_atomic_resources(
            [source, backup, temporary],
            recovery_errors,
            label: ->(handle) { handle.respond_to?(:path) ? handle.path : path }
          )
          if temporary_name
            begin
              bound.unlink(temporary_name, missing_ok: true)
            rescue SystemCallError, TranscriptionRuntimeError => e
              temporary_path = bound.display_path(temporary_name)
              recovery_errors << "cleanup #{temporary_path}: #{e.message}"
            end
          end
          if backup_name && !retained_backups.include?(bound.display_path(backup_name))
            begin
              bound.unlink(backup_name, missing_ok: true)
              backup_name = nil
            rescue SystemCallError, TranscriptionRuntimeError => e
              backup_path = bound.display_path(backup_name)
              retained_backups << backup_path
              recovery_errors << "cleanup #{backup_path}: #{e.message}"
            end
          end
        end
      end
    end
  rescue Exception => e # rubocop:disable Lint/RescueException -- retain failures outside the bound block
    primary_error ||= e
    raise
  ensure
    finish_atomic_recovery!(
      subject: "State commit",
      completed: completed,
      primary_error: primary_error,
      recovery_errors: recovery_errors,
      retained_backups: retained_backups
    )
  end
end