Switching transactions to callback-based events

Events are now queued as they are created, and
the queues are managed through simple interfaces,
rather than collecting events over time and
responding to them inline.

This drastically simplifies event management,
and will make moving it to a separate system
essentially trivial.

Signed-off-by: Luke Kanies <luke@madstop.com>

7 files changed, 593 insertions, 367 deletions

diff --git a/lib/puppet/transaction.rb b/lib/puppet/transaction.rb
index 725b86dc5..6671fb1f0 100644
--- a/lib/puppet/transaction.rb
+++ b/lib/puppet/transaction.rb
@@ -4,8 +4,7 @@
 require 'puppet'
 require 'puppet/util/tagging'
 
-module Puppet
-class Transaction
+class Puppet::Transaction
     require 'puppet/transaction/change'
     require 'puppet/transaction/event'
 
@@ -15,6 +14,9 @@ class Transaction
     # The list of events generated in this transaction.
     attr_reader :events
 
+    # Mostly only used for tests
+    attr_reader :resourcemetrics, :changes
+
     include Puppet::Util
     include Puppet::Util::Tagging
 
@@ -58,9 +60,7 @@ class Transaction
         begin
             changes = resource.evaluate
         rescue => detail
-            if Puppet[:trace]
-                puts detail.backtrace
-            end
+            puts detail.backtrace if Puppet[:trace]
 
             resource.err "Failed to retrieve current state of resource: %s" % detail
 
@@ -68,18 +68,16 @@ class Transaction
             @failures[resource] += 1
 
             # And then return
-            return []
+            return
         end
 
         changes = [changes] unless changes.is_a?(Array)
 
-        if changes.length > 0
-            @resourcemetrics[:out_of_sync] += 1
-        end
+        @resourcemetrics[:out_of_sync] += 1 if changes.length > 0
 
-        return [] if changes.empty? or ! allow_processing?(resource, changes)
+        return if changes.empty? or ! allow_processing?(resource, changes)
 
-        resourceevents = apply_changes(resource, changes)
+        apply_changes(resource, changes)
 
         # If there were changes and the resource isn't in noop mode...
         unless changes.empty? or resource.noop
@@ -90,63 +88,22 @@ class Transaction
             if resource.respond_to?(:flush)
                 resource.flush
             end
-
-            # And set a trigger for refreshing this resource if it's a
-            # self-refresher
-            if resource.self_refresh? and ! resource.deleting?
-                # Create an edge with this resource as both the source and
-                # target.  The triggering method treats these specially for
-                # logging.
-                events = resourceevents.collect { |e| e.name }
-                set_trigger(Puppet::Relationship.new(resource, resource, :callback => :refresh, :event => events))
-            end
         end
-
-        resourceevents
     end
 
     # Apply each change in turn.
     def apply_changes(resource, changes)
-        changes.collect { |change|
-            @changes << change
-            @count += 1
-            events = nil
-            begin
-                # use an array, so that changes can return more than one
-                # event if they want
-                events = [change.forward].flatten.reject { |e| e.nil? }
-            rescue => detail
-                if Puppet[:trace]
-                    puts detail.backtrace
-                end
-                change.property.err "change from %s to %s failed: %s" %
-                    [change.property.is_to_s(change.is), change.property.should_to_s(change.should), detail]
-                @failures[resource] += 1
-                next
-                # FIXME this should support using onerror to determine
-                # behaviour; or more likely, the client calling us
-                # should do so
-            end
-
-            # Mark that our change happened, so it can be reversed
-            # if we ever get to that point
-            unless events.nil? or (events.is_a?(Array) and (events.empty?) or events.include?(:noop))
-                change.changed = true
-                @resourcemetrics[:applied] += 1
-            end
-
-            events
-        }.flatten.reject { |e| e.nil? }
+        changes.each { |change| apply_change(resource, change) }
     end
 
     # Find all of the changed resources.
     def changed?
-        @changes.find_all { |change| change.changed }.collect { |change|
+        @changes.find_all { |change| change.changed }.collect do |change|
             unless change.property.resource
                 raise "No resource for %s" % change.inspect
             end
             change.property.resource
-        }.uniq
+        end.uniq
     end
 
     # Do any necessary cleanup.  If we don't get rid of the graphs, the
@@ -192,42 +149,18 @@ class Transaction
 
     # Evaluate a single resource.
     def eval_resource(resource)
-        events = []
-
-        if resource.is_a?(Puppet::Type::Component)
-            raise Puppet::DevError, "Got a component to evaluate"
-        end
-
         if skip?(resource)
             @resourcemetrics[:skipped] += 1
-        else
-            events += eval_children_and_apply_resource(resource)
+            return
         end
 
-        # Check to see if there are any events for this resource
-        if triggedevents = trigger(resource)
-            events += triggedevents
-        end
+        eval_children_and_apply_resource(resource)
 
-        # Collect the targets of any subscriptions to those events.  We pass
-        # the parent resource in so it will override the source in the events,
-        # since eval_generated children can't have direct relationships.
-        relationship_graph.matching_edges(events, resource).each do |orig_edge|
-            # We have to dup the label here, else we modify the original edge label,
-            # which affects whether a given event will match on the next run, which is,
-            # of course, bad.
-            edge = orig_edge.class.new(orig_edge.source, orig_edge.target, orig_edge.label)
-            edge.event = events.collect { |e| e.name }
-            set_trigger(edge)
-        end
-
-        # And return the events for collection
-        events
+        # Check to see if there are any events queued for this resource
+        process_events(resource)
     end
 
     def eval_children_and_apply_resource(resource)
-        events = []
-
         @resourcemetrics[:scheduled] += 1
 
         changecount = @changes.length
@@ -239,18 +172,18 @@ class Transaction
         if ! children.empty? and resource.depthfirst?
             children.each do |child|
                 # The child will never be skipped when the parent isn't
-                events += eval_resource(child, false)
+                eval_resource(child, false)
             end
         end
 
         # Perform the actual changes
         seconds = thinmark do
-            events += apply(resource)
+            apply(resource)
         end
 
         if ! children.empty? and ! resource.depthfirst?
             children.each do |child|
-                events += eval_resource(child)
+                eval_resource(child)
             end
         end
 
@@ -265,15 +198,14 @@ class Transaction
 
         # Keep track of how long we spend in each type of resource
         @timemetrics[resource.class.name] += seconds
-
-        events
     end
 
     # This method does all the actual work of running a transaction.  It
     # collects all of the changes, executes them, and responds to any
     # necessary events.
     def evaluate
-        @count = 0
+        # Start logging.
+        Puppet::Util::Log.newdestination(@report)
 
         prepare()
 
@@ -295,11 +227,7 @@ class Transaction
             ret
         }.flatten.reject { |e| e.nil? }
 
-        Puppet.debug "Finishing transaction %s with %s changes" %
-            [self.object_id, @count]
-
-        @events = allevents
-        allevents
+        Puppet.debug "Finishing transaction #{object_id} with #{@changes.length} changes"
     end
 
     # Determine whether a given resource has failed.
@@ -417,26 +345,21 @@ class Transaction
         # Metrics for distributing times across the different types.
         @timemetrics = Hash.new(0)
 
-        # The number of resources that were triggered in this run
-        @triggered = Hash.new { |hash, key|
-            hash[key] = Hash.new(0)
-        }
-
-        # Targets of being triggered.
-        @targets = Hash.new do |hash, key|
-            hash[key] = []
-        end
+        @event_queues = {}
 
         # The changes we're performing
         @changes = []
 
+        # The complete list of events generated.
+        @events = []
+
         # The resources that have failed and the number of failures each.  This
         # is used for skipping resources because of failed dependencies.
         @failures = Hash.new do |h, key|
             h[key] = 0
         end
 
-        @count = 0
+        @report = Report.new
     end
 
     # Prefetch any providers that support it.  We don't support prefetching
@@ -477,15 +400,67 @@ class Transaction
         @sorted_resources = relationship_graph.topsort
     end
 
+    # Respond to any queued events for this resource.
+    def process_events(resource)
+        restarted = false
+        queued_events(resource) do |callback, events|
+            r = process_callback(resource, callback, events)
+            restarted ||= r
+        end
+
+        if restarted
+            queue_events(Puppet::Transaction::Event.new(:restarted, resource))
+
+            @resourcemetrics[:restarted] += 1
+        end
+    end
+
+    # Queue events for other resources to respond to.  All of these events have
+    # to be from the same resource.
+    def queue_events(*events)
+        events.flatten!
+
+        @events += events
+
+        resource = events[0].resource
+
+        # Collect the targets of any subscriptions to those events.  We pass
+        # the parent resource in so it will override the source in the events,
+        # since eval_generated children can't have direct relationships.
+        relationship_graph.matching_edges(events, resource).each do |edge|
+            next unless method = edge.callback
+            next unless edge.target.respond_to?(method)
+
+            queue_events_for_resource(resource, edge.target, method, events)
+        end
+
+        if resource.self_refresh? and ! resource.deleting?
+            queue_events_for_resource(resource, resource, :refresh, events)
+        end
+    end
+
+    def queue_events_for_resource(source, target, callback, events)
+        source.info "Scheduling #{callback} of #{target}"
+
+        @event_queues[target] ||= {}
+        @event_queues[target][callback] ||= []
+        @event_queues[target][callback] += events
+    end
+
+    def queued_events(resource)
+        return unless callbacks = @event_queues[resource]
+        callbacks.each do |callback, events|
+            yield callback, events
+        end
+    end
+
     def relationship_graph
         catalog.relationship_graph
     end
 
     # Roll all completed changes back.
     def rollback
-        @targets.clear
-        @triggered.clear
-        allevents = @changes.reverse.collect { |change|
+        @changes.reverse.collect do |change|
             # skip changes that were never actually run
             unless change.changed
                 Puppet.debug "%s was not changed" % change.to_s
@@ -507,19 +482,12 @@ class Transaction
                 # but a chmod failed?  how would i handle that error? dern
             end
 
-            # FIXME This won't work right now.
-            relationship_graph.matching_edges(events).each do |edge|
-                @targets[edge.target] << edge
-            end
+            # And queue the events
+            queue_events(events)
 
             # Now check to see if there are any events for this child.
-            # Kind of hackish, since going backwards goes a change at a
-            # time, not a child at a time.
-            trigger(change.property.resource)
-
-            # And return the events for collection
-            events
-        }.flatten.reject { |e| e.nil? }
+            process_events(change.property.resource)
+        end
     end
 
     # Is the resource currently scheduled?
@@ -527,18 +495,6 @@ class Transaction
         self.ignoreschedules or resource.scheduled?
     end
 
-    # Set an edge to be triggered when we evaluate its target.
-    def set_trigger(edge)
-        return unless method = edge.callback
-        return unless edge.target.respond_to?(method)
-        if edge.target.respond_to?(:ref)
-            unless edge.source == edge.target
-                edge.source.info "Scheduling %s of %s" % [edge.callback, edge.target.ref]
-            end
-        end
-        @targets[edge.target] << edge
-    end
-
     # Should this resource be skipped?
     def skip?(resource)
         skip = false
@@ -581,87 +537,50 @@ class Transaction
         self.ignore_tags? or tags.empty? or resource.tagged?(*tags)
     end
 
-    # Are there any edges that target this resource?
-    def targeted?(resource)
-        # The default value is a new array so we have to test the length of it.
-        @targets.include?(resource) and @targets[resource].length > 0
-    end
-
-    # Trigger any subscriptions to a child.  This does an upwardly recursive
-    # search -- it triggers the passed resource, but also the resource's parent
-    # and so on up the tree.
-    def trigger(resource)
-        return nil unless targeted?(resource)
-        callbacks = Hash.new { |hash, key| hash[key] = [] }
-
-        trigged = []
-        @targets[resource].each do |edge|
-            # Collect all of the subs for each callback
-            callbacks[edge.callback] << edge
-        end
-
-        callbacks.each do |callback, subs|
-            noop = true
-            subs.each do |edge|
-                if edge.event.nil? or ! edge.event.include?(:noop)
-                    noop = false
-                end
-            end
-
-            if noop
-                resource.notice "Would have triggered %s from %s dependencies" %
-                    [callback, subs.length]
-
-                # And then add an event for it.
-                return [Puppet::Transaction::Event.new(:noop, resource)]
-            end
-
-            if subs.length == 1 and subs[0].source == resource
-                message = "Refreshing self"
-            else
-                message = "Triggering '%s' from %s dependencies" %
-                    [callback, subs.length]
-            end
-            resource.notice message
+    private
 
-            # At this point, just log failures, don't try to react
-            # to them in any way.
-            begin
-                resource.send(callback)
-                @resourcemetrics[:restarted] += 1
-            rescue => detail
-                resource.err "Failed to call %s on %s: %s" %
-                    [callback, resource, detail]
+    def apply_change(resource, change)
+        @changes << change
 
-                @resourcemetrics[:failed_restarts] += 1
+        # use an array, so that changes can return more than one
+        # event if they want
+        events = [change.forward].flatten.reject { |e| e.nil? }
 
-                if Puppet[:trace]
-                    puts detail.backtrace
-                end
-            end
+        # Mark that our change happened, so it can be reversed
+        # if we ever get to that point
+        change.changed = true
+        @resourcemetrics[:applied] += 1
+        queue_events(events)
+    rescue => detail
+        puts detail.backtrace if Puppet[:trace]
+        is = change.property.is_to_s(change.is)
+        should = change.property.should_to_s(change.should)
+        change.property.err "change from #{is} to #{should} failed: #{detail}"
+        @failures[resource] += 1
+    end
 
-            # And then add an event for it.
-            trigged << Puppet::Transaction::Event.new(:triggered, resource)
+    def process_callback(resource, callback, events)
+        process_noop_events(resource, callback, events) and return false if events.detect { |e| e.name == :noop }
+        resource.send(callback)
 
-            triggered(resource, callback)
-        end
+        resource.notice "Triggered '#{callback}' from #{events.length} events"
+        return true
+    rescue => detail
+        resource.err "Failed to call #{callback}: #{detail}"
 
-        if trigged.empty?
-            return nil
-        else
-            return trigged
-        end
+        @resourcemetrics[:failed_restarts] += 1
+        puts detail.backtrace if Puppet[:trace]
+        return false
     end
 
-    def triggered(resource, method)
-        @triggered[resource][method] += 1
-    end
+    def process_noop_events(resource, callback, events)
+        resource.notice "Would have triggered '#{callback}' from #{events.length} events"
 
-    def triggered?(resource, method)
-        @triggered[resource][method]
+        # And then add an event for it.
+        queue_events(Puppet::Transaction::Event.new(:noop, resource))
+        true # so the 'and if' works
     end
 end
-end
 
 require 'puppet/transaction/report'
 
diff --git a/lib/puppet/transaction/event.rb b/lib/puppet/transaction/event.rb
index 54c092e65..5d422f93d 100644
--- a/lib/puppet/transaction/event.rb
+++ b/lib/puppet/transaction/event.rb
@@ -1,21 +1,8 @@
 require 'puppet'
-require 'puppet/util/methodhelper'
-require 'puppet/util/errors'
-
-# events are transient packets of information; they result in one or more (or none)
-# subscriptions getting triggered, and then they get cleared
-# eventually, these will be passed on to some central event system
-class Puppet::Transaction::Event
-    include Puppet::Util::MethodHelper
-    include Puppet::Util::Errors
-
-    attr_reader :name, :source
-
-    def initialize(name, source)
-        @name, @source = name, source
-    end
 
+# A simple struct for storing what happens on the system.
+Puppet::Transaction::Event = Struct.new(:name, :resource, :property, :result, :log, :previous_value, :desired_value) do
     def to_s
-        source.to_s + " -> " + name.to_s
+        log
     end
 end
diff --git a/spec/integration/transaction.rb b/spec/integration/transaction.rb
index 7ab852bd2..1b8b11d48 100755
--- a/spec/integration/transaction.rb
+++ b/spec/integration/transaction.rb
@@ -2,9 +2,12 @@
 
 require File.dirname(__FILE__) + '/../spec_helper'
 
+require 'puppet_spec/files'
 require 'puppet/transaction'
 
 describe Puppet::Transaction do
+    include PuppetSpec::Files
+
     it "should not apply generated resources if the parent resource fails" do
         catalog = Puppet::Resource::Catalog.new
         resource = Puppet::Type.type(:file).new :path => "/foo/bar", :backup => false
@@ -63,4 +66,119 @@ describe Puppet::Transaction do
         transaction.evaluate
     end
 
+    it "should refresh resources that subscribe to changed resources" do
+        name = tmpfile("something")
+        file = Puppet::Type.type(:file).new(
+            :name => name,
+            :ensure => "file"
+        )
+        exec = Puppet::Type.type(:exec).new(
+            :name => "echo true",
+            :path => "/usr/bin:/bin",
+            :refreshonly => true,
+            :subscribe => Puppet::Resource::Reference.new(file.class.name, file.name)
+        )
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource file, exec
+
+        exec.expects(:refresh)
+
+        catalog.apply
+    end
+
+    it "should not refresh resources that only require changed resources" do
+        name = tmpfile("something")
+        file = Puppet::Type.type(:file).new(
+            :name => name,
+            :ensure => "file"
+        )
+        exec = Puppet::Type.type(:exec).new(
+            :name => "echo true",
+            :path => "/usr/bin:/bin",
+            :refreshonly => true,
+            :require => Puppet::Resource::Reference.new(file.class.name, file.name)
+        )
+
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource file
+        catalog.add_resource exec
+
+        exec.expects(:refresh).never
+
+        trans = catalog.apply
+
+        trans.events.length.should == 1
+    end
+
+    it "should cascade events such that multiple refreshes result" do
+        files = []
+
+        4.times { |i|
+            files << Puppet::Type.type(:file).new(
+                :name => tmpfile("something"),
+                :ensure => "file"
+            )
+        }
+
+        fname = tmpfile("something")
+        exec = Puppet::Type.type(:exec).new(
+            :name => "touch %s" % fname,
+            :path => "/usr/bin:/bin",
+            :refreshonly => true
+        )
+
+        exec[:subscribe] = files.collect { |f|
+            Puppet::Resource::Reference.new(:file, f.name)
+        }
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource(exec, *files)
+
+        catalog.apply
+        FileTest.should be_exist(fname)
+    end
+
+    # Make sure refreshing happens mid-transaction, rather than at the end.
+    it "should refresh resources as they're encountered rather than all at the end" do
+        file = tmpfile("something")
+
+        exec1 = Puppet::Type.type(:exec).new(
+            :title => "one",
+            :name => "echo one >> %s" % file,
+            :path => "/usr/bin:/bin"
+        )
+
+        exec2 = Puppet::Type.type(:exec).new(
+            :title => "two",
+            :name => "echo two >> %s" % file,
+            :path => "/usr/bin:/bin",
+            :refreshonly => true,
+            :subscribe => exec1
+        )
+
+        exec3 = Puppet::Type.type(:exec).new(
+            :title => "three",
+            :name => "echo three >> %s" % file,
+            :path => "/usr/bin:/bin",
+            :require => exec2
+        )
+        execs = [exec1, exec2, exec3]
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource(exec1,exec2,exec3)
+
+        trans = Puppet::Transaction.new(catalog)
+        execs.each { |e| catalog.should be_vertex(e) }
+        trans.prepare
+        execs.each { |e| catalog.should be_vertex(e) }
+        reverse = trans.relationship_graph.reversal
+        execs.each { |e| reverse.should be_vertex(e) }
+
+        catalog.apply
+
+        FileTest.should be_exist(file)
+        File.read(file).should == "one\ntwo\nthree\n"
+    end
 end
diff --git a/spec/unit/transaction.rb b/spec/unit/transaction.rb
index 260ed3703..a797d9438 100755
--- a/spec/unit/transaction.rb
+++ b/spec/unit/transaction.rb
@@ -5,17 +5,334 @@ require File.dirname(__FILE__) + '/../spec_helper'
 require 'puppet/transaction'
 
 describe Puppet::Transaction do
-    it "should match resources by name, not title, when prefetching" do
-        @catalog = Puppet::Resource::Catalog.new
-        @transaction = Puppet::Transaction.new(@catalog)
+    describe "when evaluating a resource" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+            @transaction.stubs(:eval_children_and_apply_resource)
+            @transaction.stubs(:skip?).returns false
+
+            @resource = stub("resource")
+        end
+
+        it "should check whether the resource should be skipped" do
+            @transaction.expects(:skip?).with(@resource).returns false
+
+            @transaction.eval_resource(@resource)
+        end
+
+        it "should eval and apply children" do
+            @transaction.expects(:eval_children_and_apply_resource).with(@resource)
+
+            @transaction.eval_resource(@resource)
+        end
+
+        it "should process events" do
+            @transaction.expects(:process_events).with(@resource)
+
+            @transaction.eval_resource(@resource)
+        end
+
+        describe "and the resource should be skipped" do
+            before do
+                @transaction.expects(:skip?).with(@resource).returns true
+            end
+
+            it "should increment the 'skipped' count" do
+                @transaction.eval_resource(@resource)
+                @transaction.resourcemetrics[:skipped].should == 1
+            end
+        end
+    end
+
+    describe "when applying changes" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+            @transaction.stubs(:queue_events)
+
+            @resource = stub 'resource'
+            @property = stub 'property', :is_to_s => "is", :should_to_s => "should"
+            @change = stub 'change', :property => @property, :changed= => nil, :forward => nil, :is => "is", :should => "should"
+        end
+
+        it "should apply each change" do
+            c1 = stub 'c1', :property => @property, :changed= => nil
+            c1.expects(:forward)
+            c2 = stub 'c2', :property => @property, :changed= => nil
+            c2.expects(:forward)
+
+            @transaction.apply_changes(@resource, [c1, c2])
+        end
+
+        it "should queue the events from each change" do
+            c1 = stub 'c1', :forward => "e1", :property => @property, :changed= => nil
+            c2 = stub 'c2', :forward => "e2", :property => @property, :changed= => nil
+
+            @transaction.expects(:queue_events).with(["e1"])
+            @transaction.expects(:queue_events).with(["e2"])
+
+            @transaction.apply_changes(@resource, [c1, c2])
+        end
+
+        it "should mark the change as 'changed'" do
+            @change.expects(:changed=).with true
+            @transaction.apply_changes(@resource, [@change])
+        end
+
+        it "should store the change in the transaction's change list" do
+            @transaction.apply_changes(@resource, [@change])
+            @transaction.changes.should include(@change)
+        end
+
+        it "should increment the number of applied resources" do
+            @transaction.apply_changes(@resource, [@change])
+            @transaction.resourcemetrics[:applied].should == 1
+        end
+
+        describe "and a change fails" do
+            before do
+                @change.expects(:forward).raises "a failure"
+                @change.property.stubs(:err)
+            end
+
+            it "should rescue the exception" do
+                lambda { @transaction.apply_changes(@resource, [@change]) }.should_not raise_error
+            end
+
+            it "should log the failure with the change's property" do
+                @change.property.expects(:err)
+                @transaction.apply_changes(@resource, [@change])
+            end
+
+            it "should increment the failures" do
+                @transaction.apply_changes(@resource, [@change])
+                @transaction.should be_any_failed
+            end
+        end
+    end
+
+    describe "when queueing events" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+
+            @graph = stub 'graph', :matching_edges => []
+            @transaction.stubs(:relationship_graph).returns @graph
+
+            @resource = stub("resource", :self_refresh? => false, :deleting => false)
+            @event = Puppet::Transaction::Event.new(:foo, @resource)
+        end
+
+        it "should store each event in its event list" do
+            @transaction.queue_events(@event)
+
+            @transaction.events.should include(@event)
+        end
+
+        it "should use the resource from the first edge as the source of all of the events" do
+            event1 = Puppet::Transaction::Event.new(:foo, @resource)
+            event2 = Puppet::Transaction::Event.new(:foo, stub('resource2', :self_refresh? => false, :deleting? => false))
+
+            @graph.expects(:matching_edges).with { |events, resource| resource == event1.resource }.returns []
+
+            @transaction.queue_events(event1, event2)
+        end
+
+        it "should queue events for the target and callback of any matching edges" do
+            edge1 = stub("edge1", :callback => :c1, :source => stub("s1"), :target => stub("t1", :c1 => nil))
+            edge2 = stub("edge2", :callback => :c2, :source => stub("s2"), :target => stub("t2", :c2 => nil))
+
+            @graph.expects(:matching_edges).with { |events, resource| events == [@event] }.returns [edge1, edge2]
+
+            @transaction.expects(:queue_events_for_resource).with(@resource, edge1.target, edge1.callback, [@event])
+            @transaction.expects(:queue_events_for_resource).with(@resource, edge2.target, edge2.callback, [@event])
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should queue events for the changed resource if the resource is self-refreshing and not being deleted" do
+            @graph.stubs(:matching_edges).returns []
+
+            @resource.expects(:self_refresh?).returns true
+            @resource.expects(:deleting?).returns false
+            @transaction.expects(:queue_events_for_resource).with(@resource, @resource, :refresh, [@event])
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should not queue events for the changed resource if the resource is not self-refreshing" do
+            @graph.stubs(:matching_edges).returns []
+
+            @resource.expects(:self_refresh?).returns false
+            @resource.stubs(:deleting?).returns false
+            @transaction.expects(:queue_events_for_resource).never
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should not queue events for the changed resource if the resource is being deleted" do
+            @graph.stubs(:matching_edges).returns []
+
+            @resource.expects(:self_refresh?).returns true
+            @resource.expects(:deleting?).returns true
+            @transaction.expects(:queue_events_for_resource).never
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should ignore edges that don't have a callback" do
+            edge1 = stub("edge1", :callback => :nil, :source => stub("s1"), :target => stub("t1", :c1 => nil))
+
+            @graph.expects(:matching_edges).returns [edge1]
+
+            @transaction.expects(:queue_events_for_resource).never
 
-        # Have both a title and name
-        resource = Puppet::Type.type(:sshkey).create :title => "foo", :name => "bar", :type => :dsa, :key => "eh"
-        @catalog.add_resource resource
+            @transaction.queue_events(@event)
+        end
+
+        it "should ignore targets that don't respond to the callback" do
+            edge1 = stub("edge1", :callback => :c1, :source => stub("s1"), :target => stub("t1"))
+
+            @graph.expects(:matching_edges).returns [edge1]
+
+            @transaction.expects(:queue_events_for_resource).never
+
+            @transaction.queue_events(@event)
+        end
+    end
+
+    describe "when queueing events for a resource" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+        end
 
-        resource.provider.class.expects(:prefetch).with("bar" => resource)
+        it "should do nothing if no events are queued" do
+            @transaction.queued_events(stub("target")) { |callback, events| raise "should never reach this" }
+        end
+
+        it "should yield the callback and events for each callback" do
+            target = stub("target")
+
+            2.times do |i|
+                @transaction.queue_events_for_resource(stub("source", :info => nil), target, "callback#{i}", ["event#{i}"])
+            end
+
+            @transaction.queued_events(target) { |callback, events| }
+        end
+
+        it "should use the source to log that it's scheduling a refresh of the target" do
+            target = stub("target")
+            source = stub 'source'
+            source.expects(:info)
+
+            @transaction.queue_events_for_resource(source, target, "callback", ["event"])
 
-        @transaction.prefetch
+            @transaction.queued_events(target) { |callback, events| }
+        end
+    end
+
+    describe "when processing events for a given resource" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+            @transaction.stubs(:queue_events)
+
+            @resource = stub 'resource', :notice => nil
+            @event = Puppet::Transaction::Event.new(:event, @resource)
+        end
+
+        it "should call the required callback once for each set of associated events" do
+            @transaction.expects(:queued_events).with(@resource).multiple_yields([:callback1, [@event]], [:callback2, [@event]])
+
+            @resource.expects(:callback1)
+            @resource.expects(:callback2)
+
+            @transaction.process_events(@resource)
+        end
+
+        it "should update the 'restarted' metric" do
+            @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+
+            @resource.stubs(:callback1)
+
+            @transaction.process_events(@resource)
+
+            @transaction.resourcemetrics[:restarted].should == 1
+        end
+
+        it "should queue a 'restarted' event with the resource as the source" do
+            @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+
+            @resource.stubs(:callback1)
+
+            @transaction.expects(:queue_events).with do |event|
+                event.name == :restarted and event.resource == @resource
+            end
+
+            @transaction.process_events(@resource)
+        end
+
+        it "should log that it restarted" do
+            @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+
+            @resource.stubs(:callback1)
+
+            @resource.expects(:notice).with { |msg| msg.include?("Triggered 'callback1'") }
+
+            @transaction.process_events(@resource)
+        end
+
+        describe "and the events include a noop event" do
+            before do
+                @event.stubs(:name).returns :noop
+                @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+            end
+
+            it "should log" do
+                @resource.expects(:notice).with { |msg| msg.include?("Would have triggered 'callback1'") }
+
+                @transaction.process_events(@resource)
+            end
+
+            it "should not call the callback" do
+                @resource.expects(:callback1).never
+
+                @transaction.process_events(@resource)
+            end
+
+            it "should queue a new noop event" do
+                @transaction.expects(:queue_events).with { |event| event.name == :noop and event.resource == @resource }
+
+                @transaction.process_events(@resource)
+            end
+        end
+
+        describe "and the callback fails" do
+            before do
+                @resource.expects(:callback1).raises "a failure"
+                @resource.stubs(:err)
+
+                @transaction.expects(:queued_events).yields(:callback1, [@event])
+            end
+
+            it "should log but not fail" do
+                @resource.expects(:err)
+
+                lambda { @transaction.process_events(@resource) }.should_not raise_error
+            end
+
+            it "should update the 'failed_restarts' metric" do
+                @transaction.process_events(@resource)
+                @transaction.resourcemetrics[:failed_restarts].should == 1
+            end
+
+            it "should not queue a 'restarted' event" do
+                @transaction.expects(:queue_events).never
+                @transaction.process_events(@resource)
+            end
+
+            it "should not increase the restarted resource count" do
+                @transaction.process_events(@resource)
+                @transaction.resourcemetrics[:restarted].should == 0
+            end
+        end
     end
 
     describe "when initializing" do
@@ -136,6 +453,21 @@ describe Puppet::Transaction do
             @transaction.add_metrics_to_report(@report)
         end
     end
+
+    describe "when prefetching" do
+        it "should match resources by name, not title" do
+            @catalog = Puppet::Resource::Catalog.new
+            @transaction = Puppet::Transaction.new(@catalog)
+
+            # Have both a title and name
+            resource = Puppet::Type.type(:sshkey).create :title => "foo", :name => "bar", :type => :dsa, :key => "eh"
+            @catalog.add_resource resource
+
+            resource.provider.class.expects(:prefetch).with("bar" => resource)
+
+            @transaction.prefetch
+        end
+    end
 end
 
 describe Puppet::Transaction, " when determining tags" do
diff --git a/spec/unit/transaction/event.rb b/spec/unit/transaction/event.rb
index 9fd71aae1..b7bd179f6 100755
--- a/spec/unit/transaction/event.rb
+++ b/spec/unit/transaction/event.rb
@@ -5,21 +5,17 @@ require File.dirname(__FILE__) + '/../../spec_helper'
 require 'puppet/transaction/event'
 
 describe Puppet::Transaction::Event do
-    Event = Puppet::Transaction::Event
-
-    it "should require a name and a source" do
-        lambda { Event.new }.should raise_error(ArgumentError)
-    end
-
-    it "should have a name getter" do
-        Event.new(:foo, "bar").name.should == :foo
-    end
-
-    it "should have a source accessor" do
-        Event.new(:foo, "bar").source.should == "bar"
+    [:log, :previous_value, :desired_value, :property, :resource, :name, :result].each do |attr|
+        it "should support #{attr}" do
+            event = Puppet::Transaction::Event.new
+            event.send(attr.to_s + "=", "foo")
+            event.send(attr).should == "foo"
+        end
     end
 
-    it "should be able to produce a string containing the event name and the source" do
-        Event.new(:event, :source).to_s.should == "source -> event"
+    it "should produce the log when converted to a string" do
+        event = Puppet::Transaction::Event.new
+        event.expects(:log).returns "my log"
+        event.to_s.should == "my log"
     end
 end
diff --git a/test/lib/puppettest/support/utils.rb b/test/lib/puppettest/support/utils.rb
index 5dd531200..405a0c54b 100644
--- a/test/lib/puppettest/support/utils.rb
+++ b/test/lib/puppettest/support/utils.rb
@@ -79,11 +79,9 @@ module PuppetTest::Support::Utils
 
         method = type
 
-        newevents = nil
-        assert_nothing_raised("Transaction %s %s failed" % [type, msg]) {
-            newevents = trans.send(method).reject { |e| e.nil? }.collect { |e|
-                e.name
-            }
+        trans.send(method)
+        newevents = trans.events.reject { |e| e.nil? }.collect { |e|
+            e.name
         }
 
         assert_equal(events, newevents, "Incorrect %s %s events" % [type, msg])
diff --git a/test/other/events.rb b/test/other/events.rb
deleted file mode 100755
index 052afc0c9..000000000
--- a/test/other/events.rb
+++ /dev/null
@@ -1,124 +0,0 @@
-#!/usr/bin/env ruby
-
-require File.dirname(__FILE__) + '/../lib/puppettest'
-
-require 'puppet'
-require 'puppettest'
-
-
-class TestEvents < Test::Unit::TestCase
-    include PuppetTest
-
-    def test_simplesubscribe
-        name = tempfile()
-        file = Puppet::Type.type(:file).new(
-            :name => name,
-            :ensure => "file"
-        )
-        exec = Puppet::Type.type(:exec).new(
-            :name => "echo true",
-            :path => "/usr/bin:/bin",
-            :refreshonly => true,
-            :subscribe => Puppet::Resource::Reference.new(file.class.name, file.name)
-        )
-
-        comp = mk_catalog("eventtesting", file, exec)
-
-        trans = assert_events([:file_created, :triggered], comp)
-
-        assert_equal(1, trans.triggered?(exec, :refresh))
-    end
-
-    def test_simplerequire
-        name = tempfile()
-        file = Puppet::Type.type(:file).new(
-            :name => name,
-            :ensure => "file"
-        )
-        exec = Puppet::Type.type(:exec).new(
-            :name => "echo true",
-            :path => "/usr/bin:/bin",
-            :refreshonly => true,
-            :require => Puppet::Resource::Reference.new(file.class.name, file.name)
-        )
-
-
-        config = mk_catalog
-        config.add_resource file
-        config.add_resource exec
-        trans = config.apply
-
-        assert_equal(1, trans.events.length)
-
-        assert_equal(0, trans.triggered?(exec, :refresh))
-    end
-
-    def test_multiplerefreshes
-        files = []
-
-        4.times { |i|
-            files << Puppet::Type.type(:file).new(
-                :name => tempfile(),
-                :ensure => "file"
-            )
-        }
-
-        fname = tempfile()
-        exec = Puppet::Type.type(:exec).new(
-            :name => "touch %s" % fname,
-            :path => "/usr/bin:/bin",
-            :refreshonly => true
-        )
-
-        exec[:subscribe] = files.collect { |f|
-            Puppet::Resource::Reference.new(:file, f.name)
-        }
-
-        comp = mk_catalog(exec, *files)
-
-        assert_apply(comp)
-        assert(FileTest.exists?(fname), "Exec file did not get created")
-    end
-
-    # Make sure refreshing happens mid-transaction, rather than at the end.
-    def test_refreshordering
-        file = tempfile()
-
-        exec1 = Puppet::Type.type(:exec).new(
-            :title => "one",
-            :name => "echo one >> %s" % file,
-            :path => "/usr/bin:/bin"
-        )
-
-        exec2 = Puppet::Type.type(:exec).new(
-            :title => "two",
-            :name => "echo two >> %s" % file,
-            :path => "/usr/bin:/bin",
-            :refreshonly => true,
-            :subscribe => exec1
-        )
-
-        exec3 = Puppet::Type.type(:exec).new(
-            :title => "three",
-            :name => "echo three >> %s" % file,
-            :path => "/usr/bin:/bin",
-            :require => exec2
-        )
-        execs = [exec1, exec2, exec3]
-
-        config = mk_catalog(exec1,exec2,exec3)
-
-        trans = Puppet::Transaction.new(config)
-        execs.each do |e| assert(config.vertex?(e), "%s is not in graph" % e.title) end
-        trans.prepare
-        execs.each do |e| assert(config.vertex?(e), "%s is not in relgraph" % e.title) end
-        reverse = trans.relationship_graph.reversal
-        execs.each do |e| assert(reverse.vertex?(e), "%s is not in reversed graph" % e.title) end
-
-        config.apply
-
-        assert(FileTest.exists?(file), "File does not exist")
-
-        assert_equal("one\ntwo\nthree\n", File.read(file))
-    end
-end