An engine, a theory of failure, plugins, and something that actually fetches

Milestone 5The execution engine

Goal

Replace the one-line reduce with a real interpreter: a context that flows through the pipeline, a result object recording what happened to every step, and a middleware chain so cross-cutting concerns (logging, timing, dry runs, retries) are not hard-coded into the runner.

Concepts

Immutable context objects, the middleware pattern built by folding lambdas, returning results instead of raising, and keeping a library free of logging dependencies.

Design

Three decisions worth arguing about before any code.

1. What flows between steps? Milestone 2 passed the bare return value, which meant a step could not know the pipeline's name, whether this was a dry run, or what earlier steps had learned. So we introduce a Context: payload plus run metadata. It is immutable, and a step "changes" it by producing a new one.

2. Does a failure raise or return? Both, but the default is to return. A RunResult that says ok: false and carries every step's outcome is much more useful to a CLI, a dashboard or a test than an exception, because a failed run still has partial results worth showing. run! exists for callers who prefer the exception.

3. Where do logging, timing and dry runs live? Not in the runner. Each is a middleware: a callable wrapping each step, exactly like Rack. This is the difference between a runner you can extend and one you have to edit.

  Runner#run
     │
     ├── Validator (milestone 4)
     │
     └── for each step:
            Logging ─► DryRun ─► Timing ─► [Retry] ─► registry.fetch(step).call
            └──────────── the chain, built once ─────────────┘

  each step produces a StepResult; the run produces a RunResult

Implementation

The context

  # Context is what flows between steps. It carries the payload plus
  # everything the runner and the steps need to know about the run, and it
  # is immutable: a step produces a new context rather than editing one.
  Context = Data.define(:payload, :pipeline, :step_index, :dry_run, :vars, :log) do
    def self.start(pipeline:, payload: nil, dry_run: false, log: Log.null, vars: {})
      new(payload: payload, pipeline: pipeline, step_index: 0,
          dry_run: dry_run, vars: vars.freeze, log: log)
    end

    def current_step = pipeline.steps[step_index]
    def advance(value) = with(payload: value, step_index: step_index + 1)
    def set(key, value) = with(vars: vars.merge(key => value).freeze)
    def dry_run? = dry_run
  end

Explanation

• def self.start inside a Data.define block is a class method on the generated class. The block is a class body, so def self.x works exactly as it would anywhere else.
• advance and set both return new contexts via with. Nothing is mutated, so a middleware that keeps a reference to the context it saw cannot be surprised later.
• dry_run? exists because Data generates dry_run but Ruby convention wants a question mark on a boolean reader. Adding the alias costs one line and makes every call site read properly.

The middleware chain, and the fold that builds it

  module Middleware
    Logging = lambda do |step, context, nxt|
      context.log.info("-> #{step} at #{step.location}")
      nxt.call(step, context)
    end

    DryRun = lambda do |step, context, nxt|
      if context.dry_run?
        context.log.info("would run #{step}")
        context.payload
      else
        nxt.call(step, context)
      end
    end

    Timing = lambda do |step, context, nxt|
      started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      value = nxt.call(step, context)
      elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
      context.log.info(format("%s took %.1fms", step.name, elapsed * 1000))
      value
    end

    # Build folds the middleware list into one callable, innermost last.
    def self.build(list, &innermost)
      list.reverse.reduce(innermost) do |nxt, layer|
        ->(step, context) { layer.call(step, context, nxt) }
      end
    end
  end

Middleware.build is four lines and repays reading slowly, because the same fold appears in Rack, in Rails' middleware stack, in Plug, and in Express.

1. Start with the innermost behaviour (the block: actually run the step).
2. Walk the middleware list backwards.
3. For each layer, produce a new two-argument lambda that calls that layer, handing it the chain built so far as its nxt.
4. The final value is a single callable that, when invoked, threads through every layer in the original order.

The reversal is the part people get wrong. Folding forwards would run the list inside-out, so your logging middleware would report after the step instead of before it. If you ever need to convince yourself, build the chain with two layers that print on the way in and on the way out, and watch the nesting.

Each middleware is a plain lambda, not a class, because the contract is one method. A lambda is also strict about arity, so passing a two-argument one is an immediate ArgumentError rather than a confusing nil.

Note what DryRun does: it returns context.payload without calling nxt, which short-circuits everything inside it, including the actual step. A middleware that can decline to continue is what makes caching, authorisation and dry runs possible in the same mechanism.

The runner

    def run(pipeline, payload = nil, dry_run: false, vars: {})
      Validator.new(@registry).validate!(pipeline) if @validate

      context = Context.start(pipeline: pipeline, payload: payload,
                              dry_run: dry_run, log: @log, vars: vars)
      counter = AttemptCounter.new
      chain = build_chain(pipeline, counter)
      results = []

      pipeline.steps.each_with_index do |step, index|
        step_context = context.with(step_index: index)
        counter.reset
        started = now

        begin
          value = chain.call(step, step_context)
          results << StepResult.new(step: step, status: :ok, value: value, error: nil,
                                    attempts: counter.count, seconds: now - started)
          context = step_context.advance(value)
        rescue StandardError => e
          error = e.is_a?(Error) ? e : StepFailed.new(step, e)
          results << StepResult.new(step: step, status: :failed, value: nil, error: error,
                                    attempts: counter.count, seconds: now - started)
          handle_failure(pipeline, error, step, context)
          run_always(pipeline, context)
          return RunResult.new(pipeline: pipeline, ok: false, results: results.freeze,
                               payload: nil, error: error)
        end
      end

      run_always(pipeline, context)
      RunResult.new(pipeline: pipeline, ok: true, results: results.freeze,
                    payload: context.payload, error: nil)
    end

Explanation

• Process.clock_gettime(Process::CLOCK_MONOTONIC) rather than Time.now. A monotonic clock cannot go backwards when NTP adjusts the system time, which is the correct tool for measuring durations in any language.
• The chain is built once per run, not per step. Middleware composition is not free, and doing it in the loop would be a silly cost.
• e.is_a?(Error) ? e : StepFailed.new(step, e) — our own errors pass through unwrapped, foreign errors get wrapped with the step that caused them. We will revisit whether that is the right call in Milestone 8, where it produces a mildly surprising result in a test.

The logger we did not take from a gem

  # A deliberately tiny logger so the gem has no logging dependency and
  # tests can assert on lines. Anything responding to #info/#warn/#error
  # can be used instead, including Ruby's Logger.
  class Log
    def self.null = new(nil)
    def self.to_io(io) = new(io)

    def initialize(io) = @io = io

    def info(msg)  = write("INFO ", msg)
    def warn(msg)  = write("WARN ", msg)
    def error(msg) = write("ERROR", msg)

    private

    def write(level, msg)
      return if @io.nil?

      @io.puts("#{level} #{msg}")
    end
  end

Twelve lines instead of a dependency, and the important one is Log.null: an object that satisfies the interface and does nothing. The null object pattern is how you avoid if @log scattered through a codebase, and it is why the runner can call context.log.info unconditionally. A library that forces a logging framework on its users is a library people wrap to make it quiet.

Running it

--- a real run ---
INFO  -> fetch("papers", from: "arxiv") at demo5.rb:10
INFO  fetch took 0.0ms
INFO  -> dedupe() at demo5.rb:11
INFO  dedupe took 0.0ms
INFO  -> summarize(max_words: 2) at demo5.rb:12
INFO  summarize took 0.0ms
research: ok (3 steps)
  ✓ fetch("papers", from: "arxiv") 0.0ms
  ✓ dedupe() 0.0ms
  ✓ summarize(max_words: 2) 0.0ms
["papers: alpha", "papers: beta"]

--- the same pipeline as a dry run: nothing executes ---
INFO  -> fetch("papers", from: "arxiv") at demo5.rb:10
INFO  would run fetch("papers", from: "arxiv")
INFO  -> dedupe() at demo5.rb:11
INFO  would run dedupe()
...

The log lines carry demo5.rb:10, the user's own file and line, which is the Milestone 4 source locations paying a second dividend. When a pipeline misbehaves in production, the log points at the line that wrote the step.

class Runner
  def use(middleware)
    @middleware = @middleware + [middleware]
    self
  end
end

module Middleware
  # Memoize skips a step whose (name, options, input) it has seen before.
  class Memoize
    def initialize = @cache = {}

    def call(step, context, nxt)
      return nxt.call(step, context) if context.dry_run?

      key = cache_key(step, context.payload)
      if @cache.key?(key)
        context.log.info("cache hit for #{step.name}")
        return @cache[key]
      end

      @cache[key] = nxt.call(step, context)
    end

    def size = @cache.size

    private

    def cache_key(step, payload)
      [step.name, step.args, step.options, digest(payload)]
    end

    # Hash a large payload rather than keeping it alive in the key.
    def digest(payload)
      Digest::SHA256.hexdigest(Marshal.dump(payload))
    rescue TypeError
      # Not everything is marshalable (procs, IO objects, sockets).
      # Refusing to cache is always safe; guessing is not.
      :uncacheable
    end
  end
end

Experiment

Reverse the middleware order in DEFAULT_MIDDLEWARE so it reads [Timing, DryRun, Logging] and run a dry run. Timing now wraps the dry-run short circuit, so you get timings for steps that never ran, and the logging line never appears at all because DryRun returns before reaching it. Middleware order is semantics, not style, and this is the cheapest possible way to feel that.

Checkpoint

1. Why does Middleware.build reverse the list before folding?
2. What can a middleware do that a callback list cannot?
3. Why does run return a result rather than raising, and when would you want run!?
4. What is the null object pattern and where does it appear here?
5. Why is the context immutable, given that we reassign it in the loop anyway?

Milestone 6Failure as a design, not a rescue

Goal

retry_on with backoff, when_failed handlers that receive the real error, and always for cleanup. All declared in the DSL, all inspectable before running.

Concepts

Retry as middleware, policies as data, exception chaining and the limits of cause, and handlers that must not be able to break the run.

Design

The retry logic could live in the runner as an if. Making it middleware instead means it composes with everything else, can be turned off by not adding it, and can be replaced by a user with a smarter one. And the policy is separate from the mechanism:

  RetryPolicy = Data.define(:errors, :times, :backoff, :base_delay) do
    def self.build(errors, times: 3, backoff: :exponential, base_delay: 0.01)
      errors = Array(errors)
      errors = [StandardError] if errors.empty?
      errors.each do |klass|
        unless klass.is_a?(Class) && klass <= Exception
          raise ArgumentError, "retry_on expects exception classes, got #{klass.inspect}"
        end
      end
      new(errors: errors.freeze, times: times, backoff: backoff, base_delay: base_delay)
    end

    def delay_for(attempt)
      case backoff
      when :none        then 0
      when :linear      then base_delay * attempt
      when :exponential then base_delay * (2**(attempt - 1))
      else raise ArgumentError, "unknown backoff #{backoff.inspect}"
      end
    end

    def to_s = "retry #{errors.join(', ')} up to #{times}x (#{backoff})"
  end

Explanation

• klass <= Exception uses Module#<=, which answers "is this class the same as or a subclass of". It is the correct way to ask, and it catches the common mistake of writing retry_on :timeout or retry_on "IOError" at build time with a clear message.
• The policy is a value, so pipeline.retry_policy.to_s works without running anything: "retry IOError up to 4x (exponential)". Being able to ask a pipeline about its failure behaviour is exactly the kind of thing the AST exists for.
• base_delay is configurable mainly so tests can use 10 ms instead of a second. A retry policy you cannot speed up is a retry policy your test suite will not exercise.

The retry middleware

    def self.retrying(policy, counter)
      lambda do |step, context, nxt|
        attempt = 0
        begin
          attempt += 1
          counter.count = attempt
          nxt.call(step, context)
        rescue *policy.errors => e
          raise if attempt >= policy.times

          delay = policy.delay_for(attempt)
          context.log.warn("#{step.name} failed (#{e.class}), retry #{attempt}/#{policy.times - 1} in #{delay}s")
          sleep(delay)
          retry
        end
      end
    end

rescue *policy.errors splats an array of exception classes into the rescue clause, which is how you make the caught set dynamic. retry restarts the begin block, so no loop is needed.

class AttemptCounter
  attr_accessor :count
  def initialize = @count = 1
  def reset = @count = 1
end

Two bugs I hit writing this

flaky = Automation.define("flaky") do
  retry_on IOError, times: 4
  flaky                       # <- intended: the step named :flaky
end

notified: explode failed with step explode failed: the summariser is down (RuntimeError)

  class StepFailed < Error
    attr_reader :step, :original

    # `cause` is only set by Ruby when you raise inside a rescue, and we
    # often build this object without raising it, so we keep the original
    # ourselves.
    def initialize(step, original)
      @step = step
      @original = original
      super("step #{step.name} failed: #{original.message} (#{original.class})")
    end
  end

Handlers that cannot break the run

    def handle_failure(pipeline, error, step, context)
      handler = pipeline.handler(:when_failed)
      return unless handler

      handler.callable.call(error.respond_to?(:original) ? error.original : error, step)
    rescue StandardError => e
      context.log.error("when_failed handler itself raised #{e.class}: #{e.message}")
    end

A when_failed handler is user code, running at the worst possible moment, and it frequently does something that can fail (send an email, post to Slack). If it raises and we do not catch it, the user sees the notifier's error instead of the actual failure, which is the single most annoying bug class in error-handling code. Catch it, log it, and preserve the original failure. The same applies to always.

Running it

--- retries, with backoff ---
"retry IOError up to 4x (exponential)"
INFO  -> flaky() at demo5.rb:39
WARN  flaky failed (IOError), retry 1/3 in 0.01s
WARN  flaky failed (IOError), retry 2/3 in 0.02s
INFO  flaky took 30.4ms
flaky: ok (1 steps)
  ✓ flaky() 30.4ms after 3 attempts
"worked on attempt 3"

--- failure: handlers run, the result is returned not raised ---
  notified: explode failed with the summariser is down
  cleanup ran, payload=nil
guarded: FAILED (1 steps)
  ✗ explode() Automation::StepFailed: step explode failed: the summariser is down (RuntimeError)
false
:explode
Automation::StepFailed

Exponential backoff visible in the delays (0.01 then 0.02), the attempt count carried into the result, the handler receiving the original RuntimeError rather than our wrapper, always running after the failure, and the run returning rather than raising.

    def __add_step__(verb, args, options)
      retry_spec = options.delete(:retry)
      policy = retry_spec && ::Automation::RetryPolicy.build(
        retry_spec[:errors] || [], **retry_spec.except(:errors)
      )
      @steps << ::Automation::AST::StepNode.new(
        name: verb, args: args.freeze, options: options.freeze,
        retry_policy: policy, location: __where__
      )
      self
    end

          if pipeline.handler(:continue_on_error)
            context = step_context.advance(nil)
            next
          end

    # Timeout.timeout raises inside whatever the step happens to be doing,
    # at an arbitrary point. If that point is inside a file write, an open
    # socket handshake, or an `ensure` that is releasing a lock, you can be
    # left with corrupted state that no rescue can repair. Use it only for
    # work you can safely abandon, and prefer a library's own timeout
    # (Net::HTTP's open_timeout/read_timeout) whenever one exists.
    Timeouts = lambda do |step, context, nxt|
      seconds = step.options[:timeout]
      next nxt.call(step, context) unless seconds

      Timeout.timeout(seconds, Automation::StepTimeout) { nxt.call(step, context) }
    end

Experiment

In handle_failure, delete the rescue StandardError => e guard and give a when_failed handler a bug (raise NameError, "notifier exploded"). Run a pipeline that fails: instead of a logged "when_failed handler itself raised..." line and a normal RunResult with ok: false, the whole process now crashes with the notifier's own NameError, and the actual step failure that triggered the handler is nowhere in the output. That is the "single most annoying bug class" named above, reproduced on purpose. Put the rescue back and it disappears.

Common mistakes in Milestone 6

Checkpoint

1. Why is the retry policy a Data object rather than three keyword arguments to the runner?
2. What does rescue *policy.errors do?
3. When does Ruby set Exception#cause, and why did that not help us?
4. Why must a when_failed handler's own exception be caught?
5. Explain the zero-step pipeline bug in one sentence.
6. Why is Timeout.timeout a last resort?

Milestone 7Plugins, and preferring real methods to ghosts

Goal

Third parties add verbs by writing a class. Those verbs become real methods on the DSL builder rather than method_missing ghosts, which lets us catch typos at build time with a spelling suggestion and a line number.

Concepts

Class macros built with define_method, the inherited hook, registry metadata, generating methods per registry generation, Module#prepend for instrumentation, and did_you_mean.

Design

A block is the right way to register a two-line step. It is the wrong way to register one with five documented options, defaults, and a docstring. So plugins get a class:

  class Plugin
    class << self
      def option(name, required: false, default: nil, doc: nil)
        name = name.to_sym
        options_spec[name] = { required: required, default: default, doc: doc }

        # A real method, generated once, that reads this option.
        define_method(name) { @options.fetch(name, default) }
        name
      end

      def options_spec = @options_spec ||= {}

      # Subclasses start with their own spec, seeded from the parent's, so
      # inheritance works the way people expect.
      def inherited(subclass)
        super
        subclass.instance_variable_set(:@options_spec, options_spec.dup)
      end

      def register!(registry = Automation.registry)
        registry.register(step_name, self, spec: options_spec, doc: @doc)
      end

      # The registry calls this. It builds an instance per invocation, so a
      # step may hold state during one call without leaking between runs.
      def call(input, *args, **options)
        new(options).call(input, *args)
      end
    end
  end

Which makes a plugin read like this:

class Summarize < Automation::Plugin
  step_name :summarize
  doc "Shorten each item to a few words."

  option :max_words, default: 10, doc: "how many words to keep"
  option :suffix,    default: "...", doc: "appended to truncated items"

  def call(items)
    items.map do |item|
      words = item.split
      words.size > max_words ? words.first(max_words).join(" ") + suffix : item
    end
  end
end
Summarize.register!

option does two things at once, and that duality is the whole class-macro pattern: it records metadata the validator and the documentation generator can read, and it defines an instance method so max_words works inside call. This is how attr_accessor, Rails' validates, RSpec's let and ActiveRecord's associations all work. Once you have written one, the entire Ruby ecosystem becomes less mysterious.

p Automation.registry.spec_for(:summarize).keys   # => [:max_words, :suffix]
p Summarize.instance_methods(false).sort          # => [:call, :max_words, :suffix]

The readers really exist. They are not ghosts, they appear in instance_methods, and an editor or documentation tool can see them.

Explanation

• class << self opens the singleton class, so everything inside defines class methods. It is the idiomatic way to write several of them together.
• inherited is a hook Ruby calls when a class is subclassed. Copying the parent's spec there is what makes class Fetch < BaseFetch inherit options. Always call super in a hook: someone else's library may have its own inherited in the chain, and omitting super silently breaks them.
• Plugin.call constructs a fresh instance per invocation, so a step can use instance variables freely during one call without leaking between runs. Cheap, and it removes a whole class of bug.

The validator gets better information

      spec = @registry.spec_for(step.name)
      return problems_from_spec(step, spec) if spec

      params = @registry.fetch(step.name).parameters
      missing_keywords(step, params) + unknown_keywords(step, params)

Milestone 4's validator reflected on a callable's parameters, which works for lambdas and fails for class-based steps whose call takes **options. Now a declared spec is used when present and reflection is the fallback. Declared metadata beats inferred metadata whenever it exists, and offering both means neither kind of step author is penalised.

Real verb methods instead of method_missing

  # Real methods beat method_missing: they are faster, they show up in
  # instance_methods, and anything they do NOT catch is a typo we can
  # report immediately. We build one subclass per registry generation.
  def self.builder_class_for(registry)
    @builder_classes ||= {}
    cached = @builder_classes[registry.object_id]
    return cached[:class] if cached && cached[:generation] == registry.generation

    klass = Class.new(ASTBuilder)
    registry.known.each do |verb|
      klass.send(:define_method, verb) do |*args, **options, &_block|
        __add_step__(verb, args, options)
      end
    end

    @builder_classes[registry.object_id] = { class: klass, generation: registry.generation }
    klass
  end

Class.new(ASTBuilder) creates an anonymous subclass at run time; classes are objects, so this is an ordinary constructor call. Every registered verb gets a real method on it, and the class is cached until the registry changes, which is what the generation counter on the registry tracks.

Now method_missing only sees names that are not steps, which turns it from a catch-all into a precise error detector:

    def method_missing(verb, *args, **options, &block)
      return @outer.__send__(verb, *args, **options, &block) if @outer.respond_to?(verb, true)

      # Not a registered step (those are real methods now) and not
      # something the caller can answer: in strict mode this is a typo.
      if @strict
        ::Kernel.raise ::Automation::UnknownStep.new(
          verb, @registry.known, location: __where__(2), suggestion: __suggest__(verb)
        )
      end

      __add_step__(verb, args, options)
    end

    def __suggest__(verb)
      ::DidYouMean::SpellChecker.new(dictionary: @registry.known.map(&:to_s))
                                .correct(verb.to_s).first
    end

--- a typo is caught while building, with a suggestion ---
demo7.rb:49: unknown step :summarise; did you mean summarize? (known: fetch, summarize)

File, line, the mistake, and the fix, at build time, from a dynamic language. DidYouMean ships with Ruby (it is what produces "did you mean?" on ordinary NoMethodErrors) and its SpellChecker is a public class you can point at any dictionary.

strict: false remains available and is what the data-only path and the validator tests use: build anything, check later. Two modes, one for authoring and one for tooling.

Instrumenting a plugin you do not own

module CountCalls
  def self.counts = @counts ||= Hash.new(0)

  def call(input, *args, **options)
    CountCalls.counts[step_name] += 1   # self is the class here, not an instance
    super
  end
end
Summarize.singleton_class.prepend(CountCalls)

[true, true, true]
{:summarize=>3}

prepend inserts a module before the class in the ancestor chain, so CountCalls#call runs first and super continues to the original. Prepending to singleton_class targets the class method. This is how modern Ruby does monkey-patching: instead of reopening a class and aliasing the old method away (the alias_method_chain era), you prepend a module and call super, which composes cleanly with other people doing the same thing.

The registry documents itself

  fetch      from* since="1d"
  summarize  max_words=10 suffix="..."    Shorten each item to a few words.

Generated by walking registry.known and reading each entry's spec and doc. Twelve lines of code, and it is the beginning of automation --help, of a generated reference page, and of editor completion data. Metadata you declare once can be used by things you have not written yet, which is the entire argument for the plugin class over a block.

      def problems_from_spec(step, spec)
        type_errors = step.options.filter_map do |key, value|
          expected = spec.dig(key, :type)
          next if expected.nil? || value.is_a?(expected)

          "#{step.location}: step #{step.name} option #{key.inspect} " \
            "should be #{expected}, got #{value.class}"
        end
        # ... plus the missing/unknown checks
      end

  class Plugin
    def self.inherited(subclass)
      super
      subclass.instance_variable_set(:@options_spec, options_spec.dup)
      Automation.plugin_classes << subclass
    end
  end

  def self.load_plugins(dir, registry: self.registry)
    before = plugin_classes.size
    failures = {}

    Dir.glob(File.join(dir, "*.rb")).sort.each do |file|
      require File.expand_path(file)
    rescue ScriptError, StandardError => e
      failures[file] = e          # one bad plugin must not stop the rest
    end

    plugin_classes.drop(before).each { |klass| klass.register!(registry) }
    failures
  end

      def deprecated_option(old, use:)
        option(old)
        define_method(old) do
          unless self.class.warned_about?(old)
            warn "#{caller_locations(1, 1).first}: option #{old} is deprecated, use #{use}"
          end
          @options.fetch(old) { send(use) }
        end
      end

Experiment

In builder_class_for, remove the cached[:generation] == registry.generation check so a cached builder class is returned unconditionally. Build one pipeline (which populates the cache), register a brand-new plugin on the same registry, and try to use its verb in a second pipeline: you get NoMethodError: undefined method '<verb>' instead of a new step, because the stale class from before the registration is still being handed out. Put the generation check back and the second pipeline builds correctly. This is the caching half of the object_id lesson from Milestone 9, one milestone early.

Common mistakes in Milestone 7

Checkpoint

1. What two things does the option class macro do, and why both?
2. Why must inherited call super?
3. How does the builder know when its generated methods are stale?
4. Why does method_missing now produce better errors than it did in Milestone 3?
5. What does prepend do that reopening the class and aliasing does not?
6. Why does the validator prefer a declared spec over reflection?

Milestone 8Real work: HTTP, a store, and tests that mean something

Goal

The four shipped steps do real work. fetch makes an HTTP request, save_to writes to a persistent store, summarize calls a pluggable summariser, and every one of them is tested against real behaviour rather than a mock of our own code.

Concepts

Configuration as an explicit frozen object, adapters as a seam, mapping transport failures onto one error class, and testing with a real server instead of a stub.

Design

Config is read once and frozen. A step that reads ENV itself cannot be tested twice with different settings, and in a library it is worse: your users cannot configure you without setting global state.

  Config = Data.define(:http_timeout, :user_agent, :max_redirects, :store_path, :summarizer) do
    def self.from_env(env = ENV, **overrides)
      new(
        http_timeout: Float(overrides[:http_timeout] || env.fetch("AUTOMATION_HTTP_TIMEOUT", 5)),
        user_agent: overrides[:user_agent] || env.fetch("AUTOMATION_USER_AGENT", "automation/#{VERSION}"),
        max_redirects: Integer(overrides[:max_redirects] || env.fetch("AUTOMATION_MAX_REDIRECTS", 3)),
        store_path: overrides[:store_path] || env.fetch("AUTOMATION_STORE", "knowledge_base.pstore"),
        summarizer: overrides[:summarizer] || Summarizers::Truncate
      ).freeze
    end
  end

env = ENV as a parameter is the small move that makes this testable: a test passes a Hash. Float(...) and Integer(...) are the strict conversion methods, which raise on garbage rather than returning 0 like to_i does. "abc".to_i is 0, and a timeout of zero seconds discovered in production is a bad afternoon.

The HTTP adapter, and why it collapses errors

    def request(uri, config)
      Net::HTTP.start(uri.host, uri.port,
                      use_ssl: uri.scheme == "https",
                      open_timeout: config.http_timeout,
                      read_timeout: config.http_timeout) do |http|
        http.request(Net::HTTP::Get.new(uri, "User-Agent" => config.user_agent))
      end
    rescue Errno::ECONNREFUSED, Net::OpenTimeout, Net::ReadTimeout, SocketError => e
      # Map every transport failure onto one class, so retry_on has
      # something simple to match.
      raise HttpError.new(uri.to_s, "transport", e.message)
    end

Explanation

Net::HTTP can raise at least a dozen different exceptions from unrelated hierarchies: Errno::ECONNREFUSED, SocketError, Net::OpenTimeout, OpenSSL::SSL::SSLError and more. Asking your users to write retry_on Errno::ECONNREFUSED, SocketError, Net::OpenTimeout, ... is asking them to maintain a list that will be wrong. An adapter's job is to present one coherent failure model, so we map the lot onto HttpError and users write retry_on Automation::HttpError.

Also note open_timeout and read_timeout passed to the library rather than wrapping the call in Timeout.timeout. The library knows where it is safe to give up; a generic timeout does not.

The store

  # Store is the knowledge base. PStore ships with Ruby, is transactional,
  # and needs no native extension, which makes it a good default. For
  # anything with concurrent writers or queries, swap in SQLite behind this
  # same three-method interface.
  class Store
    def initialize(path)
      @pstore = PStore.new(path, true) # true: thread-safe
    end

    def put(collection, records)
      @pstore.transaction do
        existing = @pstore[collection] || []
        @pstore[collection] = existing + Array(records)
      end
      Array(records).size
    end

    def all(collection)
      @pstore.transaction(true) { @pstore[collection] || [] } # true: read-only
    end
  end

PStore is a standard-library key-value store with transactions, backed by Marshal. It costs nothing to depend on, it is genuinely transactional, and the three-method interface means replacing it with SQLite later touches one file. That is the point of an adapter: choose the boring dependency, behind a seam.

Its real limits, which belong in your README rather than in a surprise: the whole collection is read and written on every transaction, so it does not scale past a few megabytes, and there are no queries.

The summariser contract

  # A summarizer is anything responding to #call(text, max_words:). Keeping
  # it that small means a local model, an API client, or a stub in a test
  # are all interchangeable.
  module Summarizers
    Truncate = lambda do |text, max_words:|
      words = text.to_s.split
      words.size <= max_words ? text.to_s : "#{words.first(max_words).join(' ')}…"
    end

    FirstSentence = lambda do |text, max_words:|
      sentence = text.to_s.split(/(?<=[.!?])\s/).first.to_s
      Truncate.call(sentence.empty? ? text : sentence, max_words: max_words)
    end
  end

Two lambdas and a documented signature, instead of an abstract base class. Anyone can pass their own, including one that calls a language model. The regex uses a lookbehind (?<=[.!?]) to split after sentence-ending punctuation while keeping it, which is a technique Course 3 will use rather more aggressively.

Testing against a real server

# FakeServer is a real HTTP server on a real socket, which is usually a
# better test double than stubbing Net::HTTP: it exercises the client code
# rather than replacing it.
class FakeServer
  attr_reader :port, :requests

  def initialize(status: "200 OK", body: "[]", content_type: "application/json")
    @server = TCPServer.new("127.0.0.1", 0)
    @port = @server.addr[1]
    @requests = []
    @thread = Thread.new { serve(status, body, content_type) }
  end

  def url(path = "/") = "http://127.0.0.1:#{port}#{path}"
end

TCPServer.new("127.0.0.1", 0) asks the kernel for any free port, and @server.addr[1] reports which one it chose. That is the trick that makes socket-based tests safe to run in parallel and on any machine: never hard-code a port.

Why a real server rather than stubbing Net::HTTP: a stub tests that you called the method you think you called. A real socket tests the URL you built, the headers you sent, the redirect you followed, the timeout you set and the error you mapped. It costs about thirty lines and catches an entire category of bug that mocks are structurally unable to see.

  def test_the_user_agent_is_sent
    url = @server.url   # @server would resolve on the builder, not on us
    run_it(pipeline { fetch from: url })
    assert_match(%r{User-Agent: automation/}, @server.requests.first)
  end

  def test_an_http_error_becomes_a_step_failure_with_the_status
    server = FakeServer.new(status: "503 Service Unavailable", body: "down for maintenance")
    result = run_it(pipeline { fetch from: server.url })

    refute result.ok?
    # Our own errors are not wrapped in StepFailed; the step they came from
    # is still recoverable from the result.
    assert_kind_of Automation::HttpError, result.error
    assert_equal "503", result.error.status
    assert_equal :fetch, result.failed_step.name
  ensure
    server.stop
  end

  def test_retry_on_transport_errors_eventually_gives_up
    dead = "http://127.0.0.1:1/unreachable"
    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)

    result = run_it(pipeline do
      retry_on Automation::HttpError, times: 3, backoff: :linear, base_delay: 0.01
      fetch from: dead
    end)

    refute result.ok?
    assert_equal 3, result.results.first.attempts
    assert_operator Process.clock_gettime(Process::CLOCK_MONOTONIC) - started, :<, 2
  end

  def test_dry_run_touches_no_network
    server = FakeServer.new(body: "[]")
    run_it(pipeline { fetch from: server.url }, dry_run: true)

    assert_empty server.requests
  ensure
    server.stop
  end

test_dry_run_touches_no_network is my favourite test in the project. The claim "dry run does not execute anything" is easy to assert weakly (the payload is unchanged) and hard to assert convincingly. Checking that the server received zero requests is proof rather than evidence, and it would survive a refactor that moved the dry-run check somewhere wrong.

The retry test asserts a time bound as well as the attempt count, because a retry policy with the wrong units (seconds instead of milliseconds) passes every correctness assertion and makes your suite take three minutes.

run_it(pipeline { fetch from: @server.url })   # @server is nil here

End to end

--- real run ---
research: ok (4 steps)
  ✓ fetch(from: "http://127.0.0.1:36755/papers.json") 2.4ms
  ✓ filter(field: :topic, matching: "AI") 0.0ms
  ✓ summarize(field: :abstract, max_words: 8) 0.0ms
  ✓ save_to(collection: "papers") 0.2ms
[{:title=>"Scaling laws for neural language models",
  :topic=>"AI",
  :abstract=>"We study empirical scaling laws for language model performance on the cross-entropy loss.",
  :summary=>"We study empirical scaling laws for language model…"}]

stored: 1 record(s) in the knowledge base

Real HTTP over a real socket, a real filter, a pluggable summariser, and a record on disk, driven by this:

  research = Automation.define("research") do
    retry_on Automation::HttpError, times: 3, backoff: :exponential, base_delay: 0.05
    fetch from: url
    filter field: :topic, matching: "AI"
    summarize field: :abstract, max_words: 8
    save_to collection: "papers"
    when_failed { |error, step| warn "  ! #{step.name}: #{error.message}" }
  end

That is the Part 0 sketch, working, eight milestones in.

# test/store_contract.rb
module StoreContract
  def test_put_then_all_returns_the_records
    @store.put("papers", [{ id: 1 }, { id: 2 }])
    assert_equal [{ id: 1 }, { id: 2 }], @store.all("papers")
  end

  def test_put_appends_rather_than_replacing
    @store.put("papers", { id: 1 })
    @store.put("papers", { id: 2 })
    assert_equal 2, @store.count("papers")
  end

  def test_unknown_collection_is_empty_not_an_error
    assert_equal [], @store.all("nothing_here")
  end

  def test_records_survive_a_new_instance
    @store.put("papers", { id: 1 })
    assert_equal 1, reopen.count("papers")
  end
end

class PStoreTest < Minitest::Test
  include StoreContract

  def setup
    @dir = Dir.mktmpdir
    @store = Automation::Store.new(File.join(@dir, "kb.pstore"))
  end

  def reopen = Automation::Store.new(File.join(@dir, "kb.pstore"))
  def teardown = FileUtils.remove_entry(@dir)
end

class JSONStoreTest < Minitest::Test
  include StoreContract

  def setup
    @dir = Dir.mktmpdir
    @store = Automation::JSONStore.new(File.join(@dir, "kb.json"))
  end

  def reopen = Automation::JSONStore.new(File.join(@dir, "kb.json"))
  def teardown = FileUtils.remove_entry(@dir)
end

Experiment

Set AUTOMATION_HTTP_TIMEOUT=abc in the environment and call Config.from_env: Float() raises ArgumentError: invalid value for Float(): "abc" immediately, at configuration time. Now temporarily replace that Float(...) call with a bare .to_i and repeat: it returns 0 with no error at all, and the next fetch step gets a zero-second timeout instead of a clear configuration failure. Put Float() back and the loud, early error returns.

Common mistakes in Milestone 8

Checkpoint

1. Why does Config.from_env take env as a parameter?
2. Why map every transport failure onto HttpError?
3. What does TCPServer.new("127.0.0.1", 0) do, and why 0?
4. Why is "the fake server received zero requests" a better dry-run assertion than "the payload is unchanged"?
5. Which four kinds of name behave differently inside an instance_eval block?
6. What would you change to replace PStore with SQLite, and what would you not change?

Repository state after Milestone 8

automation/
├── lib/automation/
│   ├── errors.rb        Error, UnknownStep (with suggestions), StepFailed, InvalidPipeline
│   ├── registry.rb      Entry metadata, generation counter
│   ├── plugin.rb        class macros: step_name, option, doc, register!
│   ├── steps.rb         Fetch, Filter, Summarize, SaveTo
│   ├── ast.rb           StepNode, HandlerNode, PipelineNode
│   ├── define.rb        ASTBuilder, generated verbs, strict mode, from_h
│   ├── validator.rb     declared specs, reflection fallback, file:line
│   ├── context.rb       Context, StepResult, Log
│   ├── middleware.rb    Logging, DryRun, Timing, retrying, build
│   ├── retry_policy.rb  RetryPolicy, AttemptCounter
│   ├── runner.rb        RunResult, Runner, run!
│   ├── config.rb        Config.from_env
│   ├── http.rb          HTTP.get/get_json, HttpError
│   ├── store.rb         PStore-backed knowledge base
│   └── summarizers.rb   Truncate, FirstSentence
└── test/
    ├── test_step.rb, test_registry.rb, test_define.rb,
    ├── test_validator.rb    build-time strictness and validation rules
    └── test_steps.rb        FakeServer, end to end, retries, dry run, store

$ ruby -Ilib -Itest test/all.rb
28 runs, 74 assertions, 0 failures, 0 errors, 0 skips
$ git commit -am "milestone 8: real adapters, config, and tests against a real socket"