diff --git a/TODO.complete/33-streaming-parser.md b/TODO.complete/33-streaming-parser.md index dbeea3e..97a9f41 100644 --- a/TODO.complete/33-streaming-parser.md +++ b/TODO.complete/33-streaming-parser.md @@ -1,11 +1,25 @@ # 33 — Streaming parser for large documents (P3) -**Status:** PENDING +**Status:** COMPLETED -## Gaps -- Neither `lib/dcc/streaming.rb` nor `lib/dcc/streaming/reader.rb` exists. -- Nothing in `lib/` uses `Nokogiri::XML::Reader`. -- `spec/dcc/streaming/reader_spec.rb` was removed — see phase 38. +## Outcome +Built on `Moxml::SAX` rather than the `Nokogiri::XML::Reader` named in the +Goal below. `CONTRIBUTING.adoc` bans Nokogiri outside +`lib/dcc/validate/xsd.rb`, so the reader could not use it. The lazy, +constant-memory streaming this phase asked for is unaffected. + +Items and quantities are enumerable. Results are out of scope for this phase: +there is no streaming entry point for them, and no path to one from a streamed +item — `Item` has no result-shaped attribute, and `results` hangs off +`MeasurementResult`. + +Memory is verified by `spec/support/streaming_memory_probe.rb`, which runs in +a child process and asserts retained heap stays flat as the stream advances, +instead of the 50 MB / 200 MB fixture in the Verification section below. That +bounds growth at any document size rather than at one size. + +`spec/dcc/streaming/reader_spec.rb`, removed in phase 38, is replaced by +`spec/dcc/streaming_spec.rb`. ## Goal Lazy enumeration of items / results / quantities for multi-MB DCC documents using Nokogiri::XML::Reader. diff --git a/lib/dcc.rb b/lib/dcc.rb index ed8a1c5..69e898b 100644 --- a/lib/dcc.rb +++ b/lib/dcc.rb @@ -41,6 +41,7 @@ module Dcc autoload :QuantityFormat, "dcc/quantity_format" autoload :QuantityMath, "dcc/quantity_math" autoload :Signature, "dcc/signature" + autoload :Streaming, "dcc/streaming" autoload :Transform, "dcc/transform" autoload :Validate, "dcc/validate" autoload :Server, "dcc/server" @@ -104,9 +105,18 @@ def parser_for(version) # @return [Integer] 2 or 3. def detect_version(input) str = read_input(input) - match = str.match(/schemaVersion\s*=\s*["'](\d+)\./) - major = match && match[1] ? match[1].to_i : 3 - major == 2 ? 2 : 3 + match = str.match(/schemaVersion\s*=\s*["'](\d+\.)/) + major_version_from(match && match[1]) + end + + # Map a `schemaVersion` attribute value to its major DCC version. + # Only a leading run of digits followed by a dot counts; anything else + # (missing, malformed, or an unsupported major) resolves to 3. + # @param schema_version [String, nil] e.g. "2.3.0". + # @return [Integer] 2 or 3. + def major_version_from(schema_version) + match = schema_version.to_s.match(/\A(\d+)\./) + match && match[1].to_i == 2 ? 2 : 3 end # Read an input that may be a String or an IO-like object. diff --git a/lib/dcc/streaming.rb b/lib/dcc/streaming.rb new file mode 100644 index 0000000..d5a4e58 --- /dev/null +++ b/lib/dcc/streaming.rb @@ -0,0 +1,79 @@ +# frozen_string_literal: true + +# `Dcc::Streaming` walks a DCC document one subtree at a time, so a batch +# pipeline can process a multi-MB certificate without holding the whole +# object tree in memory. Each matched subtree is reconstructed on its own, +# parsed into the version-appropriate model, yielded, and dropped. +# +# @example Walk every calibrated item +# File.open("certificate.xml") do |io| +# Dcc::Streaming.each_item(io) { |item| puts item.id } +# end +# +# @example Take the first two quantities without reading the whole document +# Dcc::Streaming.each_quantity(io).first(2) +# +# Version detection differs from `Dcc.parse`. Streaming reads `schemaVersion` +# off a real DCC element, so a certificate wrapped in an envelope is detected +# from its own root even when the envelope carries a `schemaVersion` of its +# own. `Dcc.detect_version` scans the document lexically instead, so it also +# matches a `schemaVersion=` inside a comment, inside text or CDATA, on a +# prefixed attribute such as `f:schemaVersion`, or on a foreign wrapper +# element. The two can disagree on those documents, and streaming is the one +# reading the schema's actual attribute. Pass `version:` to settle it. +module Dcc + module Streaming + autoload :Reader, "dcc/streaming/reader" + + class << self + # Yield each `dcc:item` in document order. + # + # @param io [IO, StringIO] readable XML stream. + # @param version [Integer, nil] major DCC version (2 or 3). When nil it + # is detected from the first `schemaVersion` attribute seen outside a + # matched subtree before the first item is yielded, and falls back to + # 3. See the note on version detection above. + # @param context [Symbol, String, nil] substitution context. Defaults + # to the version's configured context. + # @yieldparam [Dcc::V2::Item, Dcc::V3::Item] + # @return [Enumerator] when no block is given. + # @raise [Dcc::ParseError] if the stream is not well-formed XML. + def each_item(io, version: nil, + context: ::Dcc::UNSPECIFIED_CONTEXT, &block) + return enum_for(:each_item, io, version:, context:) unless block + + stream(io, :item, version, context, &block) + end + + # Yield each outermost `dcc:quantity` in document order. + # + # A quantity nested inside another quantity's `influenceConditions` is + # not yielded on its own — it is reachable from the outer object. + # + # @param io [IO, StringIO] readable XML stream. + # @param version [Integer, nil] major DCC version (2 or 3). Detected the + # same way `each_item` detects it when nil. + # @param context [Symbol, String, nil] substitution context. + # @yieldparam [Dcc::V2::Quantity, Dcc::V3::Quantity] + # @return [Enumerator] when no block is given. + # @raise [Dcc::ParseError] if the stream is not well-formed XML. + def each_quantity(io, version: nil, + context: ::Dcc::UNSPECIFIED_CONTEXT, &block) + return enum_for(:each_quantity, io, version:, context:) unless block + + stream(io, :quantity, version, context, &block) + end + + private + + # `version` is validated here rather than on first match, so an + # unsupported value raises the same way regardless of what the + # document happens to contain. + def stream(io, model_id, version, context, &) + ::Dcc.parser_for(version) if version + ::Dcc.load_all! + Reader.call(io, model_id:, version:, context:, &) + end + end + end +end diff --git a/lib/dcc/streaming/reader.rb b/lib/dcc/streaming/reader.rb new file mode 100644 index 0000000..a36595f --- /dev/null +++ b/lib/dcc/streaming/reader.rb @@ -0,0 +1,267 @@ +# frozen_string_literal: true + +require "moxml" + +module Dcc + module Streaming + # SAX handler that reconstructs each matched subtree as a standalone XML + # fragment, parses it into the version-appropriate model, yields it, and + # drops it. Only one subtree is buffered at a time, so peak memory tracks + # the largest single subtree rather than the document. + # + # The fragment reproduces the XML infoset, not the original bytes. + # Entity references arrive expanded, CDATA becomes text, processing + # instruction spacing is normalised, and every namespace in scope is + # re-declared on the fragment root so prefixes still resolve once the + # subtree is detached. + class Reader < ::Moxml::SAX::Handler + # Stream `io`, yielding one model per matched subtree. + # + # @param io [IO, StringIO] readable XML stream. + # @param model_id [Symbol] registered model id, `:item` or `:quantity`. + # @param version [Integer, nil] major DCC version. When nil it is read + # from the first `schemaVersion` attribute seen outside a matched + # subtree before the first model is yielded, and falls back to 3. + # `Dcc.detect_version` instead scans the whole document lexically, + # because it is not restricted to a single pass. + # @param context [Symbol, String, nil] substitution context. + # @return [void] + # @raise [Dcc::ParseError] if the stream is not well-formed XML. A + # failing IO also surfaces this way, with the original IO message + # replaced by the parser's — libxml2 does not pass it through. + def self.call(io, model_id:, version:, context:, &) + handler = new(model_id:, version:, context:, &) + adapter = ::Lutaml::Model::Config.xml_adapter_type + ::Moxml.new(adapter).sax_parse(io, handler) + end + + # A handler carries one document's worth of state, so `.call` builds a + # fresh one per stream and construction stays closed. + private_class_method :new + + def initialize(model_id:, version:, context:, &block) + super() + @model_id = model_id + @major = version + @context = context + @block = block + @scopes = [{}] + @subtree = nil + @depth = 0 + end + + # Declarations are merged and snapshotted before the element's own + # prefix is resolved, so an element that rebinds its prefix matches on + # its new binding rather than the inherited one. + def on_start_element(name, attributes = {}, namespaces = {}) + # Almost no element declares a namespace, so merging every time would + # allocate a copy of the parent scope per element. Scopes are only + # ever read, so the unchanged ones can share one hash. + @scopes.push( + namespaces.empty? ? @scopes.last : @scopes.last.merge(namespaces), + ) + return begin_subtree(name, attributes) unless @subtree + + @depth += 1 + @subtree.start_element(name, attributes, namespaces) + end + + def on_end_element(name) + finish_element(name) if @subtree + @scopes.pop + end + + def on_characters(text) + @subtree&.characters(text) + end + + def on_cdata(text) + @subtree&.characters(text) + end + + def on_comment(text) + @subtree&.comment(text) + end + + def on_processing_instruction(target, data) + @subtree&.processing_instruction(target, data) + end + + # Translated here rather than around the whole parse, so an exception + # raised by the consumer block travels untouched instead of being + # reported as malformed input. + def on_error(error) + raise ::Dcc::ParseError, error.message + end + + private + + # Only elements outside a matched subtree get a say in the version, so a + # `schemaVersion` on a descendant — or inside an opaque `dcc:xml` blob — + # cannot choose the parser for the whole stream. + def begin_subtree(name, attributes) + @major ||= schema_version_major(name, attributes) + return unless match?(name) + + @depth = 0 + @subtree = Subtree.new(name, attributes, @scopes.last) + end + + def finish_element(name) + @subtree.end_element(name) + if @depth.zero? + emit(@subtree.to_xml) + @subtree = nil + else + @depth -= 1 + end + end + + def emit(xml) + @block.call(model_class.from_xml(xml, register: context_id)) + end + + def match?(name) + split(name).last == @model_id.to_s && dcc_element?(name) + end + + def dcc_element?(name) + dcc_namespace?(@scopes.last[split(name).first]) + end + + # `Dcc::Namespace::Dcc` also accepts the `.xsd` alias that older PTB + # documents bind, so streaming matches whatever `Dcc.parse` accepts. + def dcc_namespace?(uri) + namespace = ::Dcc::Namespace::Dcc + uri == namespace.uri || namespace.uri_aliases.include?(uri) + end + + def split(name) + prefix, local = name.split(":", 2) + local ? [prefix, local] : [nil, prefix] + end + + # Resolved through the registry rather than referenced directly, so a + # configured custom model substitutes the root class too. + def model_class + @model_class ||= + ::Lutaml::Model::GlobalContext.resolve_type(@model_id, context_id) + end + + def context_id + @context_id ||= ::Dcc::ContextOptions.normalize_context_option( + context: @context, + register: nil, + default_context: configuration.default_context_id, + warning_source: "Dcc::Streaming", + ) + end + + def configuration + ::Dcc.parser_for(major)::Configuration + end + + # `schemaVersion` sits on the DCC root, which is not always the document + # root — a certificate can arrive wrapped in an envelope. Only a DCC + # element settles the version, so a wrapper carrying a `schemaVersion` + # of its own does not retype the certificate nested inside it. + def schema_version_major(name, attributes) + version = attributes["schemaVersion"] + return unless version && dcc_element?(name) + + ::Dcc.major_version_from(version) + end + + # A document that declares `schemaVersion` nowhere falls back the way an + # unusable one does, so `Dcc` keeps deciding that rather than a literal + # repeated here. The answer is written back rather than defaulted on + # each read: the model class and the context are memoised off this at + # the first yield, so a `schemaVersion` appearing later must not move + # the version out from under them. + def major + @major ||= ::Dcc.major_version_from(nil) + end + + # Accumulates the raw XML of a single matched subtree. + # + # SAX hands over decoded characters, so control whitespace has to go + # back as character references. Written literally it would be + # normalised a second time on reparse — a carriage return in text + # would arrive as a newline, and a tab or newline in an attribute + # would collapse to a space. + # + # C14N mandates exactly that set of character references, so moxml's + # encoder is reused rather than restating the table here. + class Subtree + def initialize(name, attributes, scope) + @buffer = +"" + write_start(name, attributes, scope) + end + + def start_element(name, attributes, namespaces) + write_start(name, attributes, namespaces) + end + + def end_element(name) + @buffer << "" + end + + def characters(text) + @buffer << ::Moxml::C14n.escape_text(text) + end + + def comment(text) + @buffer << "" + end + + def processing_instruction(target, data) + @buffer << "" + end + + def to_xml + @buffer + end + + private + + def write_start(name, attributes, namespaces) + @buffer << "<#{name}" + namespaces.each do |prefix, href| + write_attribute(prefix ? "xmlns:#{prefix}" : "xmlns", href) + end + attributes.each do |key, value| + write_attribute(key, decode_ampersands(value)) + end + @buffer << ">" + end + + # SAX decodes an attribute value except for `&`, which libxml2 hands + # back still written as the decimal reference `&` whatever form + # the source used. Escaping that would emit `&#38;`, and the value + # would come back one round trip more corrupt each time. A bare `&` + # never reaches us, which is what makes putting it back exact rather + # than a guess. + # + # Namespace URIs and character data do not come through here: moxml + # hands those over fully decoded, and a URI carrying a literal `&` + # would be corrupted by this substitution. Re-measure both channels if + # the moxml dependency moves. + # Most attributes carry no `&`, and `gsub` allocates a copy even + # when it substitutes nothing. Measured over 300k calls: the guard is + # about twice as fast when there is no match, and about 13% slower on + # the rare value that does match. + def decode_ampersands(value) + return value unless value.include?("&") + + value.gsub("&", "&") + end + + def write_attribute(name, value) + escaped = ::Moxml::C14n.escape_attribute(value) + @buffer << %( #{name}="#{escaped}") + end + end + private_constant :Subtree + end + end +end diff --git a/spec/dcc/streaming_spec.rb b/spec/dcc/streaming_spec.rb new file mode 100644 index 0000000..7361723 --- /dev/null +++ b/spec/dcc/streaming_spec.rb @@ -0,0 +1,425 @@ +# frozen_string_literal: true + +require "spec_helper" +require "json" +require "open3" +require "rbconfig" +require "stringio" + +RSpec.describe Dcc::Streaming do + def dcc_ns + 'xmlns:dcc="https://ptb.de/dcc"' + end + + def doc(body, version: "3.3.0") + si = 'xmlns:si="https://ptb.de/si"' + %(#{body}) + end + + def rooted(declarations, body) + %(#{body}) + end + + def stream_items(xml, **opts) + found = [] + described_class.each_item(StringIO.new(xml), **opts) { |i| found << i } + found + end + + def stream_quantities(xml, **opts) + found = [] + described_class.each_quantity(StringIO.new(xml), **opts) { |q| found << q } + found + end + + def item(id: "a", body: "M") + %(#{body}) + end + + def real(value: "1", unit: '\\metre') + "#{value}" \ + "#{unit}" + end + + def blob_quantity(payload) + doc("#{real}" \ + "" \ + 't' \ + "#{payload}" \ + "" \ + "") + end + + def blob_raw(payload) + quantity = stream_quantities(blob_quantity(payload)).first + condition = quantity.influence_conditions.influence_condition.first + condition.data.first.xml.first.raw.to_s + end + + describe ".each_item" do + let(:ptb) { load_fixture("dcc_examples", "example.xml") } + + it "yields every item in the PTB reference document" do + expect(stream_items(ptb).map(&:id)).to eq(%w[dcc10g dcc100g]) + end + + it "auto-detects v2 from the fixture's schemaVersion" do + expect(stream_items(ptb).first).to be_a(Dcc::V2::Item) + end + + it "auto-detects v3 from a 3.x schemaVersion" do + expect(stream_items(doc(item)).first).to be_a(Dcc::V3::Item) + end + + it "honours an explicit version override" do + streamed = stream_items(ptb, version: 3) + expect(streamed.first).to be_a(Dcc::V3::Item) + end + + it "parses the item body into the typed model" do + expect(stream_items(doc(item)).first.model).to eq("M") + end + + it "yields nothing for a valid document with no items" do + expect(stream_items(doc(""))).to be_empty + end + + it "returns an Enumerator when no block is given" do + streamed = described_class.each_item(StringIO.new(doc(item))) + expect(streamed).to be_a(Enumerator) + end + + # The document is left unterminated, so reaching the end raises. Only a + # reader that stops at the second yield gets through this. + it "stops reading once the caller stops asking" do + body = Array.new(50) { |i| item(id: "i#{i}") }.join + truncated = %(#{body}) + streamed = described_class.each_item(StringIO.new(truncated)) + expect(streamed.first(2).map(&:id)).to eq(%w[i0 i1]) + end + + it "raises when that same document is consumed to the end" do + body = Array.new(50) { |i| item(id: "i#{i}") }.join + truncated = %(#{body}) + expect { stream_items(truncated) }.to raise_error(Dcc::ParseError) + end + end + + describe ".each_quantity" do + let(:nested) do + doc("#{real}" \ + "" \ + "t" \ + "#{real(value: '20')}" \ + "" \ + "" \ + "") + end + + it "populates the D-SI quantity through the fallback context" do + quantity = stream_quantities(doc("#{real}" \ + "")).first + expect(quantity.real.map(&:unit)).to eq(['\\metre']) + end + + it "yields only the outermost of a nested pair" do + expect(stream_quantities(nested).map(&:id)).to eq(["outer"]) + end + + it "keeps the nested quantity reachable from the outer one" do + outer = stream_quantities(nested).first + inner = outer.influence_conditions.influence_condition.first + expect(inner.data.first.quantity.first.id).to eq("inner") + end + end + + describe "opaque XML payloads" do + it "preserves comments inside dcc:xml" do + payload = 'v' + expect(blob_raw(payload)).to include("") + end + + it "preserves processing instructions inside dcc:xml" do + payload = 'v' + expect(blob_raw(payload)).to include("") + end + + it "preserves a processing instruction with no data" do + payload = 'v' + expect(blob_raw(payload)).to include("v' + expect(blob_raw(payload)).to include("a b") + end + + it "escapes markup characters in text" do + payload = 'a < b & c' + expect(blob_raw(payload)).to include("a < b & c") + end + + # SAX hands back an attribute value with every `&` still written as + # `&`, and hands back namespace URIs, comments and PI data untouched. + # The reader has to decode one channel and leave the others alone, so + # these pin each channel separately. Decoding one level lower, in the + # shared attribute writer, passes the rows above and fails these. + it "keeps an ampersand in a prefixed attribute on a descendant" do + payload = 'v' + expect(blob_raw(payload)).to include('f:q="a&b"') + end + + it "keeps an ampersand in a plain attribute on a descendant" do + payload = 'v' + expect(blob_raw(payload)).to include('q="a&b"') + end + + it "keeps an ampersand in a namespace URI" do + payload = 'v' + expect(blob_raw(payload)).to include('xmlns:f="http://f/?x=1&y"') + end + + # The row that separates the correct fix from the plausible wrong one: a + # namespace URI arrives decoded, so a literal `&` in one is real text + # and must survive. The plain-ampersand row above cannot catch this, + # because it has no `&` in it to corrupt. + it "leaves a literal ampersand reference in a namespace URI alone" do + payload = 'v' + expect(blob_raw(payload)) + .to include('xmlns:f="http://f/?x=1&#38;y"') + end + + it "leaves a literal ampersand reference in a comment alone" do + payload = 'v' + expect(blob_raw(payload)).to include("") + end + + it "leaves a literal ampersand reference in a PI alone" do + payload = 'v' + expect(blob_raw(payload)).to include("") + end + end + + describe "namespace resolution" do + it "matches an element that rebinds its own prefix" do + hit = %() + xml = rooted('xmlns:x="urn:root"', %(#{hit})) + expect(stream_items(xml).map(&:id)).to eq(["hit"]) + end + + it "restores the outer binding for a following sibling" do + shadowed = %() + xml = rooted('xmlns:x="https://ptb.de/dcc"', + %(#{shadowed})) + expect(stream_items(xml).map(&:id)).to eq(%w[a b]) + end + + it "matches the .xsd namespace alias older documents bind" do + body = %(M) + xml = rooted('xmlns:dcc="https://ptb.de/dcc.xsd"', body) + expect(stream_items(xml).first.model).to eq("M") + end + + it "matches items declared in a default namespace" do + body = %(M) + xml = rooted('xmlns="https://ptb.de/dcc"', body) + expect(stream_items(xml).first.model).to eq("M") + end + + it "handles an undeclared default namespace on a descendant" do + payload = 'v' + expect(blob_raw(payload)).to include('xmlns=""') + end + end + + describe "custom model substitution" do + before { stub_const("MyItem", Class.new(Dcc::V3::Item)) } + + after { Dcc::V3::Configuration.clear_custom_models } + + it "substitutes the root class, not just nested models" do + Dcc::V3::Configuration.custom_models = { Dcc::V3::Item => MyItem } + expect(stream_items(doc(item)).first).to be_a(MyItem) + end + end + + describe "version detection" do + it "defaults to v3 when schemaVersion is absent" do + xml = %(#{item}) + expect(stream_items(xml).first).to be_a(Dcc::V3::Item) + end + + it "defaults to v3 when schemaVersion is malformed" do + expect(stream_items(doc(item, version: "abc")).first) + .to be_a(Dcc::V3::Item) + end + + it "rejects an unsupported explicit version even with no matches" do + expect { stream_items(doc(""), version: 5) } + .to raise_error(Dcc::UnknownVersionError) + end + + it "defaults to v3 for an unsupported major version" do + expect(stream_items(doc(item, version: "9.0.0")).first) + .to be_a(Dcc::V3::Item) + end + + # `schemaVersion` sits on the DCC root, which is not always the document + # root. Reading it off the outermost element instead pins every enveloped + # certificate to v3. + it "reads the version off a certificate nested in an envelope" do + body = %(#{doc(item, version: '2.4.0')}) + xml = %(#{body}) + expect(stream_items(xml).first).to be_a(Dcc::V2::Item) + end + + it "defaults to v3 when the item is itself the document root" do + xml = %(M) + expect(stream_items(xml).first).to be_a(Dcc::V3::Item) + end + + # Anything inside a matched subtree is payload, not document structure — + # an opaque `dcc:xml` blob could carry any attribute at all. + it "ignores a schemaVersion on a descendant of a matched item" do + model = %(M) + xml = %(#{item(body: model)}) + expect(stream_items(xml).first).to be_a(Dcc::V3::Item) + end + + # The first document with no `schemaVersion` above it falls back to v3, + # and that has to stick: the model class and the context are memoised off + # the version at the first yield. + it "keeps the version settled once the first item is yielded", + :aggregate_failures do + later = %(#{item(id: 'b')}) + streamed = stream_items(%(#{item}#{later})) + expect(streamed.map(&:id)).to eq(%w[a b]) + expect(streamed.map(&:class)).to eq([Dcc::V3::Item, Dcc::V3::Item]) + end + + # An envelope can carry a `schemaVersion` of its own. The certificate's + # version has to win, not whichever attribute the parser happens to reach + # first. + it "ignores a schemaVersion on a non-DCC wrapper element", + :aggregate_failures do + meta = %() + xml = %(#{meta}#{doc(item)}) + streamed = stream_items(xml) + expect(streamed.map(&:id)).to eq(["a"]) + expect(streamed.first).to be_a(Dcc::V3::Item) + end + end + + describe "equivalence with direct fragment parsing" do + def direct(body) + Dcc::V3::Item.from_xml(%( ["a&b", "a&b"], + "a decimal ampersand reference" => ["a&b", "a&b"], + "a hex ampersand reference" => ["a&b", "a&b"], + "two ampersands in one value" => ["?a=1&b=2&c=3", + "?a=1&b=2&c=3"], + "a re-escaped decimal ampersand" => ["a&#38;b", "a&b"], + "a re-escaped named ampersand" => ["a&amp;b", "a&b"], + "a re-escaped hex ampersand" => ["a&#x26;b", "a&b"], + "a less-than" => ["a<b", "a ["a>b", "a>b"], + "a double quote" => ["a"b", %(a"b)], + "an apostrophe" => ["a'b", "a'b"], + "a tab" => ["a b", "a\tb"], + "a newline" => ["a b", "a\nb"], + "a carriage return" => ["a b", "a\rb"], + "an astral character" => ["a\u{1F600}b", "a\u{1F600}b"], + "a non-breaking space" => ["a b", "a b"], + } + + character_classes.each do |description, (source, literal)| + it "keeps #{description} in an attribute", :aggregate_failures do + a, b = both(%(id="#{source}">M)) + expect(b.id).to eq(a.id) + expect(b.id).to eq(literal) + end + + it "keeps #{description} in a text node", :aggregate_failures do + a, b = both(%(id="x">#{source})) + expect(b.model).to eq(a.model) + expect(b.model).to eq(literal) + end + end + end + + describe "error handling" do + it "raises ParseError on malformed XML" do + expect { stream_items("oops") } + .to raise_error(Dcc::ParseError) + end + + it "raises ParseError on a truncated document" do + expect { stream_items(%()) } + .to raise_error(Dcc::ParseError) + end + + it "raises ParseError on empty input" do + expect { stream_items("") }.to raise_error(Dcc::ParseError) + end + + it "raises ParseError on whitespace-only input" do + expect { stream_items(" \n ") }.to raise_error(Dcc::ParseError) + end + + it "lets an exception from the consumer block through untouched" do + io = StringIO.new(doc(item)) + consumer = proc { raise Moxml::ParseError, "from the consumer" } + expect { described_class.each_item(io, &consumer) } + .to raise_error(Moxml::ParseError, /from the consumer/) + end + + it "raises ParseError when the IO itself fails" do + io = Class.new(StringIO) { def read(*) = raise(IOError, "boom") } + expect { described_class.each_item(io.new("")) { |_| nil } } + .to raise_error(Dcc::ParseError) + end + end + + describe "memory behaviour" do + let(:probe) do + lib = File.expand_path("../../lib", __dir__) + out, err, status = Open3.capture3(RbConfig.ruby, "-I#{lib}", + memory_probe_path) + raise "probe failed: #{err}" unless status.success? + + JSON.parse(out) + end + + # The counts are asserted non-zero first: a flat line at zero would + # otherwise satisfy the equality and leave the gate blind. + it "does not accumulate models, DOM nodes or fragment strings", + :aggregate_failures do + expect(probe["heap_growth_kb"]).to be < probe["budget_kb"] + expect(probe["items_first"]).to be_positive + expect(probe["elements_first"]).to be_positive + expect(probe["items_last"]).to eq(probe["items_first"]) + expect(probe["elements_last"]).to eq(probe["elements_first"]) + end + end + + # Runs in a child process so the suite's own loaded state and RSpec's + # per-example retention cannot pollute the measurement. + def memory_probe_path + File.expand_path("../support/streaming_memory_probe.rb", __dir__) + end +end diff --git a/spec/support/streaming_memory_probe.rb b/spec/support/streaming_memory_probe.rb new file mode 100644 index 0000000..47e2677 --- /dev/null +++ b/spec/support/streaming_memory_probe.rb @@ -0,0 +1,67 @@ +# frozen_string_literal: true + +# Memory probe for `Dcc::Streaming`. Run as a child process by +# `spec/dcc/streaming_spec.rb` so the suite's loaded state and RSpec's +# per-example retention cannot pollute the measurement. Prints one JSON +# object on stdout. +# +# It streams a document of uniformly sized items and samples the retained +# heap at two checkpoints. A reader that streams shows no growth between +# them; one that hoards models, DOM nodes or fragment strings shows growth +# proportional to the bytes it has seen. +# +# Both samples are taken inside the block, while the yielded item is still +# live, so each carries exactly one item's worth of objects. That is +# deliberate: the same bias sits in both checkpoints and cancels out of the +# comparison, and the resulting non-zero counts double as evidence the +# probe is measuring something at all. + +require "objspace" +require "stringio" +require "json" +require "dcc" + +PAD = ("z" * 4000).freeze +FIRST = 20 +LAST = 200 + +def document + body = (1..LAST).map do |i| + %(#{PAD}#{i}) + end.join + root = %(xmlns:dcc="https://ptb.de/dcc" schemaVersion="3.3.0") + %(#{body}) +end + +def sample + GC.start + GC.start + [ObjectSpace.memsize_of_all, + ObjectSpace.each_object(Dcc::V3::Item).count, + ObjectSpace.each_object(Moxml::Element).count] +end + +marks = {} +seen = 0 +Dcc::Streaming.each_item(StringIO.new(document)) do |_item| + seen += 1 + marks[seen] = sample if [FIRST, LAST].include?(seen) +end + +first = marks[FIRST] +last = marks[LAST] +streamed_kb = ((LAST - FIRST) * PAD.bytesize) / 1024.0 + +# Not an example group — this is a standalone program whose entire contract +# is printing the measurement to stdout for the parent process to parse. +# rubocop:disable RSpec/Output +puts({ + yielded: seen, + heap_growth_kb: ((last[0] - first[0]) / 1024.0).round, + budget_kb: (streamed_kb * 0.1).round, + items_first: first[1], + items_last: last[1], + elements_first: first[2], + elements_last: last[2], +}.to_json) +# rubocop:enable RSpec/Output