diff --git a/lib/puppet/simple_graph.rb b/lib/puppet/simple_graph.rb
index 27e068e30..31858f136 100644
--- a/lib/puppet/simple_graph.rb
+++ b/lib/puppet/simple_graph.rb
@@ -1,594 +1,602 @@
 require 'puppet/external/dot'
 require 'puppet/relationship'
 require 'set'
 
 # A hopefully-faster graph class to replace the use of GRATR.
 class Puppet::SimpleGraph
   #
   # All public methods of this class must maintain (assume ^ ensure) the following invariants, where "=~=" means
   # equiv. up to order:
   #
   #   @in_to.keys =~= @out_to.keys =~= all vertices
   #   @in_to.values.collect { |x| x.values }.flatten =~= @out_from.values.collect { |x| x.values }.flatten =~= all edges
   #   @in_to[v1][v2] =~= @out_from[v2][v1] =~= all edges from v1 to v2
   #   @in_to   [v].keys =~= vertices with edges leading to   v
   #   @out_from[v].keys =~= vertices with edges leading from v
   #   no operation may shed reference loops (for gc)
   #   recursive operation must scale with the depth of the spanning trees, or better (e.g. no recursion over the set
   #       of all vertices, etc.)
   #
   # This class is intended to be used with DAGs.  However, if the
   # graph has a cycle, it will not cause non-termination of any of the
   # algorithms.  The topsort method detects and reports cycles.
   #
   def initialize
     @in_to = {}
     @out_from = {}
     @upstream_from = {}
     @downstream_from = {}
   end
 
   # Clear our graph.
   def clear
     @in_to.clear
     @out_from.clear
     @upstream_from.clear
     @downstream_from.clear
   end
 
   # Which resources depend upon the given resource.
   def dependencies(resource)
     vertex?(resource) ? upstream_from_vertex(resource).keys : []
   end
 
   def dependents(resource)
     vertex?(resource) ? downstream_from_vertex(resource).keys : []
   end
 
   # Whether our graph is directed.  Always true.  Used to produce dot files.
   def directed?
     true
   end
 
   # Determine all of the leaf nodes below a given vertex.
   def leaves(vertex, direction = :out)
     tree_from_vertex(vertex, direction).keys.find_all { |c| adjacent(c, :direction => direction).empty? }
   end
 
   # Collect all of the edges that the passed events match.  Returns
   # an array of edges.
   def matching_edges(event, base = nil)
     source = base || event.resource
 
     unless vertex?(source)
       Puppet.warning "Got an event from invalid vertex #{source.ref}"
       return []
     end
     # Get all of the edges that this vertex should forward events
     # to, which is the same thing as saying all edges directly below
     # This vertex in the graph.
     @out_from[source].values.flatten.find_all { |edge| edge.match?(event.name) }
   end
 
   # Return a reversed version of this graph.
   def reversal
     result = self.class.new
     vertices.each { |vertex| result.add_vertex(vertex) }
     edges.each do |edge|
       result.add_edge edge.class.new(edge.target, edge.source, edge.label)
     end
     result
   end
 
   # Return the size of the graph.
   def size
     vertices.size
   end
 
   def to_a
     vertices
   end
 
   # This is a simple implementation of Tarjan's algorithm to find strongly
   # connected components in the graph; this is a fairly ugly implementation,
   # because I can't just decorate the vertices themselves.
   #
   # This method has an unhealthy relationship with the find_cycles_in_graph
   # method below, which contains the knowledge of how the state object is
   # maintained.
   def tarjan(root, s)
     # initialize the recursion stack we use to work around the nasty lack of a
     # decent Ruby stack.
     recur = [{ :node => root }]
 
     while not recur.empty? do
       frame = recur.last
       vertex = frame[:node]
 
       case frame[:step]
       when nil then
         s[:index][vertex]   = s[:number]
         s[:lowlink][vertex] = s[:number]
         s[:number]          = s[:number] + 1
 
         s[:stack].push(vertex)
         s[:seen][vertex] = true
 
         frame[:children] = adjacent(vertex)
         frame[:step]     = :children
 
       when :children then
         if frame[:children].length > 0 then
           child = frame[:children].shift
           if ! s[:index][child] then
             # Never seen, need to recurse.
             frame[:step] = :after_recursion
             frame[:child] = child
             recur.push({ :node => child })
           elsif s[:seen][child] then
             s[:lowlink][vertex] = [s[:lowlink][vertex], s[:index][child]].min
           end
         else
           if s[:lowlink][vertex] == s[:index][vertex] then
             this_scc = []
             begin
               top = s[:stack].pop
               s[:seen][top] = false
               this_scc << top
             end until top == vertex
             # NOTE: if we don't reverse we get the components in the opposite
             # order to what a human being would expect; reverse should be an
             # O(1) operation, without even copying, because we know the length
             # of the source, but I worry that an implementation will get this
             # wrong.  Still, the worst case is O(n) for n vertices as we can't
             # possibly put a vertex into two SCCs.
             #
             # Also, my feeling is that most implementations are going to do
             # better with a reverse operation than a string of 'unshift'
             # insertions at the head of the array; if they were going to mess
             # up the performance of one, it would be unshift.
             s[:scc] << this_scc.reverse
           end
           recur.pop               # done with this node, finally.
         end
 
       when :after_recursion then
         s[:lowlink][vertex] = [s[:lowlink][vertex], s[:lowlink][frame[:child]]].min
         frame[:step] = :children
 
       else
         fail "#{frame[:step]} is an unknown step"
       end
     end
   end
 
   # Find all cycles in the graph by detecting all the strongly connected
   # components, then eliminating everything with a size of one as
   # uninteresting - which it is, because it can't be a cycle. :)
   #
   # This has an unhealthy relationship with the 'tarjan' method above, which
   # it uses to implement the detection of strongly connected components.
   def find_cycles_in_graph
     state = {
       :number => 0, :index => {}, :lowlink => {}, :scc => [],
       :stack => [], :seen => {}
     }
 
     # we usually have a disconnected graph, must walk all possible roots
     vertices.each do |vertex|
       if ! state[:index][vertex] then
         tarjan vertex, state
       end
     end
 
     state[:scc].select { |c| c.length > 1 }
   end
 
   # Perform a BFS on the sub graph representing the cycle, with a view to
   # generating a sufficient set of paths to report the cycle meaningfully, and
   # ideally usefully, for the end user.
   #
   # BFS is preferred because it will generally report the shortest paths
   # through the graph first, which are more likely to be interesting to the
   # user.  I think; it would be interesting to verify that. --daniel 2011-01-23
   def paths_in_cycle(cycle, max_paths = 1)
     raise ArgumentError, "negative or zero max_paths" if max_paths < 1
 
     # Calculate our filtered outbound vertex lists...
     adj = {}
     cycle.each do |vertex|
       adj[vertex] = adjacent(vertex).select{|s| cycle.member? s}
     end
 
     found = []
 
     # frame struct is vertex, [path]
     stack = [[cycle.first, []]]
     while frame = stack.shift do
       if frame[1].member?(frame[0]) then
         found << frame[1] + [frame[0]]
         break if found.length >= max_paths
       else
         adj[frame[0]].each do |to|
           stack.push [to, frame[1] + [frame[0]]]
         end
       end
     end
 
     return found
   end
 
   def report_cycles_in_graph
     cycles = find_cycles_in_graph
     n = cycles.length           # where is "pluralize"? --daniel 2011-01-22
     s = n == 1 ? '' : 's'
 
     message = "Found #{n} dependency cycle#{s}:\n"
     cycles.each do |cycle|
       paths = paths_in_cycle(cycle)
       message += paths.map{ |path| '(' + path.join(" => ") + ')'}.join("\n") + "\n"
     end
 
     if Puppet[:graph] then
       filename = write_cycles_to_graph(cycles)
       message += "Cycle graph written to #{filename}."
     else
       message += "Try the '--graph' option and opening the "
       message += "resulting '.dot' file in OmniGraffle or GraphViz"
     end
 
     raise Puppet::Error, message
   end
 
   def write_cycles_to_graph(cycles)
     # This does not use the DOT graph library, just writes the content
     # directly.  Given the complexity of this, there didn't seem much point
     # using a heavy library to generate exactly the same content. --daniel 2011-01-27
     Puppet.settings.use(:graphing)
 
     graph = ["digraph Resource_Cycles {"]
     graph << '  label = "Resource Cycles"'
 
     cycles.each do |cycle|
       paths_in_cycle(cycle, 10).each do |path|
         graph << path.map { |v| '"' + v.to_s.gsub(/"/, '\\"') + '"' }.join(" -> ")
       end
     end
 
     graph << '}'
 
     filename = File.join(Puppet[:graphdir], "cycles.dot")
     File.open(filename, "w") { |f| f.puts graph }
     return filename
   end
 
   # Provide a topological sort.
   def topsort
     degree = {}
     zeros = []
     result = []
 
     # Collect each of our vertices, with the number of in-edges each has.
     vertices.each do |v|
       edges = @in_to[v]
       zeros << v if edges.empty?
       degree[v] = edges.length
     end
 
     # Iterate over each 0-degree vertex, decrementing the degree of
     # each of its out-edges.
     while v = zeros.pop
       result << v
       @out_from[v].each { |v2,es|
         zeros << v2 if (degree[v2] -= 1) == 0
       }
     end
 
     # If we have any vertices left with non-zero in-degrees, then we've found a cycle.
     if cycles = degree.values.reject { |ns| ns == 0  } and cycles.length > 0
       report_cycles_in_graph
     end
 
     result
   end
 
   # Add a new vertex to the graph.
   def add_vertex(vertex)
     @in_to[vertex]    ||= {}
     @out_from[vertex] ||= {}
   end
 
   # Remove a vertex from the graph.
   def remove_vertex!(v)
     return unless vertex?(v)
     @upstream_from.clear
     @downstream_from.clear
     (@in_to[v].values+@out_from[v].values).flatten.each { |e| remove_edge!(e) }
     @in_to.delete(v)
     @out_from.delete(v)
   end
 
   # Test whether a given vertex is in the graph.
   def vertex?(v)
     @in_to.include?(v)
   end
 
   # Return a list of all vertices.
   def vertices
     @in_to.keys
   end
 
   # Add a new edge.  The graph user has to create the edge instance,
   # since they have to specify what kind of edge it is.
   def add_edge(e,*a)
     return add_relationship(e,*a) unless a.empty?
     @upstream_from.clear
     @downstream_from.clear
     add_vertex(e.source)
     add_vertex(e.target)
     @in_to[   e.target][e.source] ||= []; @in_to[   e.target][e.source] |= [e]
     @out_from[e.source][e.target] ||= []; @out_from[e.source][e.target] |= [e]
   end
 
   def add_relationship(source, target, label = nil)
     add_edge Puppet::Relationship.new(source, target, label)
   end
 
   # Find all matching edges.
   def edges_between(source, target)
     (@out_from[source] || {})[target] || []
   end
 
   # Is there an edge between the two vertices?
   def edge?(source, target)
     vertex?(source) and vertex?(target) and @out_from[source][target]
   end
 
   def edges
     @in_to.values.collect { |x| x.values }.flatten
   end
 
   def each_edge
     @in_to.each { |t,ns| ns.each { |s,es| es.each { |e| yield e }}}
   end
 
   # Remove an edge from our graph.
   def remove_edge!(e)
     if edge?(e.source,e.target)
       @upstream_from.clear
       @downstream_from.clear
       @in_to   [e.target].delete e.source if (@in_to   [e.target][e.source] -= [e]).empty?
       @out_from[e.source].delete e.target if (@out_from[e.source][e.target] -= [e]).empty?
     end
   end
 
   # Find adjacent edges.
   def adjacent(v, options = {})
     return [] unless ns = (options[:direction] == :in) ? @in_to[v] : @out_from[v]
     (options[:type] == :edges) ? ns.values.flatten : ns.keys
   end
 
   # Take container information from another graph and use it
   # to replace any container vertices with their respective leaves.
   # This creates direct relationships where there were previously
   # indirect relationships through the containers.
   def splice!(other, type)
     # We have to get the container list via a topological sort on the
     # configuration graph, because otherwise containers that contain
     # other containers will add those containers back into the
     # graph.  We could get a similar affect by only setting relationships
     # to container leaves, but that would result in many more
     # relationships.
     stage_class = Puppet::Type.type(:stage)
     whit_class  = Puppet::Type.type(:whit)
     containers = other.topsort.find_all { |v| (v.is_a?(type) or v.is_a?(stage_class)) and vertex?(v) }
     containers.each do |container|
       # Get the list of children from the other graph.
       children = other.adjacent(container, :direction => :out)
 
       # MQR TODO: Luke suggests that it should be possible to refactor the system so that
       #           container nodes are retained, thus obviating the need for the whit.
       children = [whit_class.new(:name => container.name, :catalog => other)] if children.empty?
 
       # First create new edges for each of the :in edges
       [:in, :out].each do |dir|
         edges = adjacent(container, :direction => dir, :type => :edges)
         edges.each do |edge|
           children.each do |child|
             if dir == :in
               s = edge.source
               t = child
             else
               s = child
               t = edge.target
             end
 
             add_edge(s, t, edge.label)
           end
 
           # Now get rid of the edge, so remove_vertex! works correctly.
           remove_edge!(edge)
         end
       end
       remove_vertex!(container)
     end
   end
 
   # Just walk the tree and pass each edge.
   def walk(source, direction)
     # Use an iterative, breadth-first traversal of the graph. One could do
     # this recursively, but Ruby's slow function calls and even slower
     # recursion make the shorter, recursive algorithm cost-prohibitive.
     stack = [source]
     seen = Set.new
     until stack.empty?
       node = stack.shift
       next if seen.member? node
       connected = adjacent(node, :direction => direction)
       connected.each do |target|
         yield node, target
       end
       stack.concat(connected)
       seen << node
     end
   end
 
   # A different way of walking a tree, and a much faster way than the
   # one that comes with GRATR.
   def tree_from_vertex(start, direction = :out)
     predecessor={}
     walk(start, direction) do |parent, child|
       predecessor[child] = parent
     end
     predecessor
   end
 
   def downstream_from_vertex(v)
     return @downstream_from[v] if @downstream_from[v]
     result = @downstream_from[v] = {}
     @out_from[v].keys.each do |node|
       result[node] = 1
       result.update(downstream_from_vertex(node))
     end
     result
   end
 
+  def direct_dependents_of(v)
+    @out_from[v].keys 
+  end
+
   def upstream_from_vertex(v)
     return @upstream_from[v] if @upstream_from[v]
     result = @upstream_from[v] = {}
     @in_to[v].keys.each do |node|
       result[node] = 1
       result.update(upstream_from_vertex(node))
     end
     result
   end
 
+  def direct_dependencies_of(v)
+    @in_to[v].keys 
+  end
+
   # LAK:FIXME This is just a paste of the GRATR code with slight modifications.
 
   # Return a DOT::DOTDigraph for directed graphs or a DOT::DOTSubgraph for an
   # undirected Graph.  _params_ can contain any graph property specified in
   # rdot.rb. If an edge or vertex label is a kind of Hash then the keys
   # which match +dot+ properties will be used as well.
   def to_dot_graph (params = {})
     params['name'] ||= self.class.name.gsub(/:/,'_')
     fontsize   = params['fontsize'] ? params['fontsize'] : '8'
     graph      = (directed? ? DOT::DOTDigraph : DOT::DOTSubgraph).new(params)
     edge_klass = directed? ? DOT::DOTDirectedEdge : DOT::DOTEdge
     vertices.each do |v|
       name = v.to_s
       params = {'name'     => '"'+name+'"',
         'fontsize' => fontsize,
         'label'    => name}
       v_label = v.to_s
       params.merge!(v_label) if v_label and v_label.kind_of? Hash
       graph << DOT::DOTNode.new(params)
     end
     edges.each do |e|
       params = {'from'     => '"'+ e.source.to_s + '"',
         'to'       => '"'+ e.target.to_s + '"',
         'fontsize' => fontsize }
       e_label = e.to_s
       params.merge!(e_label) if e_label and e_label.kind_of? Hash
       graph << edge_klass.new(params)
     end
     graph
   end
 
   # Output the dot format as a string
   def to_dot (params={}) to_dot_graph(params).to_s; end
 
   # Call +dotty+ for the graph which is written to the file 'graph.dot'
   # in the # current directory.
   def dotty (params = {}, dotfile = 'graph.dot')
     File.open(dotfile, 'w') {|f| f << to_dot(params) }
     system('dotty', dotfile)
   end
 
   # Produce the graph files if requested.
   def write_graph(name)
     return unless Puppet[:graph]
 
     Puppet.settings.use(:graphing)
 
     file = File.join(Puppet[:graphdir], "#{name}.dot")
     File.open(file, "w") { |f|
       f.puts to_dot("name" => name.to_s.capitalize)
     }
   end
 
   # This flag may be set to true to use the new YAML serialzation
   # format (where @vertices is a simple list of vertices rather than a
   # list of VertexWrapper objects).  Deserialization supports both
   # formats regardless of the setting of this flag.
   class << self
     attr_accessor :use_new_yaml_format
   end
   self.use_new_yaml_format = false
 
   # Stub class to allow graphs to be represented in YAML using the old
   # (version 2.6) format.
   class VertexWrapper
     attr_reader :vertex, :adjacencies
     def initialize(vertex, adjacencies)
       @vertex = vertex
       @adjacencies = adjacencies
     end
 
     def inspect
       { :@adjacencies => @adjacencies, :@vertex => @vertex.to_s }.inspect
     end
   end
 
   # instance_variable_get is used by Object.to_zaml to get instance
   # variables.  Override it so that we can simulate the presence of
   # instance variables @edges and @vertices for serialization.
   def instance_variable_get(v)
     case v.to_s
     when '@edges' then
       edges
     when '@vertices' then
       if self.class.use_new_yaml_format
         vertices
       else
         result = {}
         vertices.each do |vertex|
           adjacencies = {}
           [:in, :out].each do |direction|
             adjacencies[direction] = {}
             adjacent(vertex, :direction => direction, :type => :edges).each do |edge|
               other_vertex = direction == :in ? edge.source : edge.target
               (adjacencies[direction][other_vertex] ||= Set.new).add(edge)
             end
           end
           result[vertex] = Puppet::SimpleGraph::VertexWrapper.new(vertex, adjacencies)
         end
         result
       end
     else
       super(v)
     end
   end
 
   def to_yaml_properties
     other_vars = instance_variables.
       map {|v| v.to_s}.
       reject { |v| %w{@in_to @out_from @upstream_from @downstream_from}.include?(v) }
 
     (other_vars + %w{@vertices @edges}).sort.uniq
   end
 
   def yaml_initialize(tag, var)
     initialize()
     vertices = var.delete('vertices')
     edges = var.delete('edges')
     if vertices.is_a?(Hash)
       # Support old (2.6) format
       vertices = vertices.keys
     end
     vertices.each { |v| add_vertex(v) }
     edges.each { |e| add_edge(e) }
     var.each do |varname, value|
       instance_variable_set("@#{varname}", value)
     end
   end
 end
diff --git a/lib/puppet/transaction.rb b/lib/puppet/transaction.rb
index 928a1b488..79fdb044c 100644
--- a/lib/puppet/transaction.rb
+++ b/lib/puppet/transaction.rb
@@ -1,368 +1,378 @@
 # the class that actually walks our resource/property tree, collects the changes,
 # and performs them
 
 require 'puppet'
 require 'puppet/util/tagging'
 require 'puppet/application'
+require 'sha1'
 
 class Puppet::Transaction
   require 'puppet/transaction/event'
   require 'puppet/transaction/event_manager'
   require 'puppet/transaction/resource_harness'
   require 'puppet/resource/status'
 
   attr_accessor :component, :catalog, :ignoreschedules
   attr_accessor :configurator
 
   # The report, once generated.
   attr_accessor :report
 
   # Routes and stores any events and subscriptions.
   attr_reader :event_manager
 
   # Handles most of the actual interacting with resources
   attr_reader :resource_harness
 
   include Puppet::Util
   include Puppet::Util::Tagging
 
   # Wraps application run state check to flag need to interrupt processing
   def stop_processing?
     Puppet::Application.stop_requested?
   end
 
   # Add some additional times for reporting
   def add_times(hash)
     hash.each do |name, num|
       report.add_times(name, num)
     end
   end
 
   # Are there any failed resources in this transaction?
   def any_failed?
     report.resource_statuses.values.detect { |status| status.failed? }
   end
 
   # Apply all changes for a resource
   def apply(resource, ancestor = nil)
     status = resource_harness.evaluate(resource)
     add_resource_status(status)
     event_manager.queue_events(ancestor || resource, status.events)
   rescue => detail
     resource.err "Could not evaluate: #{detail}"
   end
 
   # Find all of the changed resources.
   def changed?
     report.resource_statuses.values.find_all { |status| status.changed }.collect { |status| catalog.resource(status.resource) }
   end
 
   # Find all of the applied resources (including failed attempts).
   def applied_resources
     report.resource_statuses.values.collect { |status| catalog.resource(status.resource) }
   end
 
   # Copy an important relationships from the parent to the newly-generated
   # child resource.
   def make_parent_child_relationship(parent, child)
     relationship_graph.add_vertex(child)
     edge = parent.depthfirst? ? [child, parent] : [parent, child]
     if relationship_graph.edge?(*edge.reverse)
       parent.debug "Skipping automatic relationship to #{child}"
     else
       relationship_graph.add_edge(*edge)
     end
   end
 
   # Evaluate a single resource.
   def eval_resource(resource, ancestor = nil)
     if skip?(resource)
       resource_status(resource).skipped = true
     else
       eval_children_and_apply_resource(resource, ancestor)
     end
 
     # Check to see if there are any events queued for this resource
     event_manager.process_events(resource)
   end
 
   def eval_children_and_apply_resource(resource, ancestor = nil)
     resource_status(resource).scheduled = true
     apply(resource, ancestor)
   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
     # Start logging.
     Puppet::Util::Log.newdestination(@report)
 
     prepare
 
     Puppet.info "Applying configuration version '#{catalog.version}'" if catalog.version
 
     begin
-      visit_resources do |resource|
+      relationship_graph.traverse do |resource|
         next if stop_processing?
         if resource.is_a?(Puppet::Type::Component)
           Puppet.warning "Somehow left a component in the relationship graph"
           next
         end
         ret = nil
         seconds = thinmark do
           ret = eval_resource(resource)
         end
 
         resource.info "Evaluated in %0.2f seconds" % seconds if Puppet[:evaltrace] and @catalog.host_config?
         ret
       end
     ensure
       # And then close the transaction log.
       Puppet::Util::Log.close(@report)
     end
 
     Puppet.debug "Finishing transaction #{object_id}"
   end
 
   def events
     event_manager.events
   end
 
   def failed?(resource)
     s = resource_status(resource) and s.failed?
   end
 
   # Does this resource have any failed dependencies?
   def failed_dependencies?(resource)
     # First make sure there are no failed dependencies.  To do this,
     # we check for failures in any of the vertexes above us.  It's not
     # enough to check the immediate dependencies, which is why we use
     # a tree from the reversed graph.
     found_failed = false
     relationship_graph.dependencies(resource).each do |dep|
       next unless failed?(dep)
       resource.notice "Dependency #{dep} has failures: #{resource_status(dep).failed}"
       found_failed = true
     end
 
     found_failed
   end
 
   def eval_generate(resource)
     return [] unless resource.respond_to?(:eval_generate)
     begin
       made = resource.eval_generate
     rescue => detail
       puts detail.backtrace if Puppet[:trace]
       resource.err "Failed to generate additional resources using 'eval_generate: #{detail}"
     end
     parents = [resource]
     [made].flatten.compact.uniq.each do |res|
       begin
         res.tag(*resource.tags)
         @catalog.add_resource(res)
         res.finish
         make_parent_child_relationship(parents.reverse.find { |r| r.name == res.name[0,r.name.length]}, res)
         parents << res
       rescue Puppet::Resource::Catalog::DuplicateResourceError
         res.info "Duplicate generated resource; skipping"
       end
     end
   end
 
   # A general method for recursively generating new resources from a
   # resource.
   def generate_additional_resources(resource)
     return [] unless resource.respond_to?(:generate)
     begin
       made = resource.generate
     rescue => detail
       puts detail.backtrace if Puppet[:trace]
       resource.err "Failed to generate additional resources using 'generate': #{detail}"
     end
     return [] unless made
     made = [made] unless made.is_a?(Array)
     made.uniq.find_all do |res|
       begin
         res.tag(*resource.tags)
         @catalog.add_resource(res)
         res.finish
         make_parent_child_relationship(resource, res)
         generate_additional_resources(res)
         true
       rescue Puppet::Resource::Catalog::DuplicateResourceError
         res.info "Duplicate generated resource; skipping"
         false
       end
     end
   end
 
   # Collect any dynamically generated resources.  This method is called
   # before the transaction starts.
   def generate
     list = @catalog.vertices
     newlist = []
     while ! list.empty?
       list.each do |resource|
         newlist += generate_additional_resources(resource)
       end
       list = newlist
       newlist = []
     end
   end
 
   # Should we ignore tags?
   def ignore_tags?
     ! (@catalog.host_config? or Puppet[:name] == "puppet")
   end
 
   # this should only be called by a Puppet::Type::Component resource now
   # and it should only receive an array
   def initialize(catalog)
     @catalog = catalog
 
     @report = Puppet::Transaction::Report.new("apply")
 
     @event_manager = Puppet::Transaction::EventManager.new(self)
 
     @resource_harness = Puppet::Transaction::ResourceHarness.new(self)
   end
 
   # Prefetch any providers that support it.  We don't support prefetching
   # types, just providers.
   def prefetch
     prefetchers = {}
     @catalog.vertices.each do |resource|
       if provider = resource.provider and provider.class.respond_to?(:prefetch)
         prefetchers[provider.class] ||= {}
         prefetchers[provider.class][resource.name] = resource
       end
     end
 
     # Now call prefetch, passing in the resources so that the provider instances can be replaced.
     prefetchers.each do |provider, resources|
       Puppet.debug "Prefetching #{provider.name} resources for #{provider.resource_type.name}"
       begin
         provider.prefetch(resources)
       rescue => detail
         puts detail.backtrace if Puppet[:trace]
         Puppet.err "Could not prefetch #{provider.resource_type.name} provider '#{provider.name}': #{detail}"
       end
     end
   end
 
   # Prepare to evaluate the resources in a transaction.
   def prepare
     # Now add any dynamically generated resources
     generate
 
     # Then prefetch.  It's important that we generate and then prefetch,
     # so that any generated resources also get prefetched.
     prefetch
   end
 
-  def visit_resources(&block)
-    eval_generated = {}
-    while r = relationship_graph.topsort.find { |r| !applied_resources.include?(r) }
-      #p relationship_graph.topsort.collect { |r0| r0.title[/(-0.*)/,1] }
-      if !eval_generated[r]
-        #p [:generate,r.title]
-        eval_generate(r)
-        eval_generated[r] = true
-      else
-        #p [:apply,r.title]
-        yield r
-      end
-    end
-  end
 
   # We want to monitor changes in the relationship graph of our
   # catalog but this is complicated by the fact that the catalog
   # both is_a graph and has_a graph, by the fact that changes to 
   # the structure of the object can have adverse serialization
   # effects, by threading issues, by order-of-initialization issues,
   # etc.  
   #
   # Since the proper lifetime/scope of the monitoring is a transaction
   # and the transaction is already commiting a mild law-of-demeter 
   # transgression, we cut the Gordian knot here by simply wrapping the
   # transaction's view of the resource graph to capture and maintain
   # the information we need.  Nothing outside the transaction needs
   # this information, and nothing outside the transaction can see it
   # except via the Transaction#relationship_graph
 
   class Relationship_graph_wrapper
+    attr_reader :real_graph,:transaction,:ready,:generated,:done
     def initialize(real_graph,transaction)
       @real_graph = real_graph
       @transaction = transaction
+      @ready = {}
+      @generated = {}
+      @done = {}
+      vertices.each { |v| check_if_now_ready(v) }
     end
     def method_missing(*args,&block)
-      @real_graph.send(*args,&block)
+      real_graph.send(*args,&block)
     end
     def add_vertex(v)
-      @real_graph.add_vertex(v)
+      real_graph.add_vertex(v)
     end
     def add_edge(f,t)
-      @real_graph.add_edge(f,t)
+      ready.delete(t)
+      real_graph.add_edge(f,t)
+    end
+    def check_if_now_ready(r)
+      ready[r] = true if direct_dependencies_of(r).all? { |r2| done[r2] }
+    end  
+    def traverse(&block)
+      unguessable_deterministic_key = Hash.new { |h,k| h[k] = Digest::SHA1.hexdigest("NaCl, MgSO4 (salts) and then #{k.title}") }
+      while r = ready.keys.sort_by { |r0| unguessable_deterministic_key[r0] }.first
+        if !generated[r]
+          transaction.eval_generate(r)
+          generated[r] = true
+        else
+          ready.delete(r)
+          yield r
+          done[r] = true
+          direct_dependents_of(r).each { |v| check_if_now_ready(v) }
+        end
+      end
     end
   end
 
   def relationship_graph
     @relationship_graph ||= Relationship_graph_wrapper.new(catalog.relationship_graph,self)
   end
 
   def add_resource_status(status)
     report.add_resource_status status
   end
 
   def resource_status(resource)
     report.resource_statuses[resource.to_s] || add_resource_status(Puppet::Resource::Status.new(resource))
   end
 
   # Is the resource currently scheduled?
   def scheduled?(resource)
     self.ignoreschedules or resource_harness.scheduled?(resource_status(resource), resource)
   end
 
   # Should this resource be skipped?
   def skip?(resource)
     if missing_tags?(resource)
       resource.debug "Not tagged with #{tags.join(", ")}"
     elsif ! scheduled?(resource)
       resource.debug "Not scheduled"
     elsif failed_dependencies?(resource)
       resource.warning "Skipping because of failed dependencies"
     elsif resource.virtual?
       resource.debug "Skipping because virtual"
     else
       return false
     end
     true
   end
 
   # The tags we should be checking.
   def tags
     self.tags = Puppet[:tags] unless defined?(@tags)
 
     super
   end
 
   def handle_qualified_tags( qualified )
     # The default behavior of Puppet::Util::Tagging is
     # to split qualified tags into parts. That would cause
     # qualified tags to match too broadly here.
     return
   end
 
   # Is this resource tagged appropriately?
   def missing_tags?(resource)
     return false if ignore_tags?
     return false if tags.empty?
 
     not resource.tagged?(*tags)
   end
 end
 
 require 'puppet/transaction/report'
 
diff --git a/spec/unit/transaction_spec.rb b/spec/unit/transaction_spec.rb
index 57bf8005e..076b626bb 100755
--- a/spec/unit/transaction_spec.rb
+++ b/spec/unit/transaction_spec.rb
@@ -1,449 +1,451 @@
 #!/usr/bin/env ruby
 
 require File.expand_path(File.dirname(__FILE__) + '/../spec_helper')
 
 require 'puppet/transaction'
 
 def without_warnings
   flag = $VERBOSE
   $VERBOSE = nil
   yield
   $VERBOSE = flag
 end
 
 describe Puppet::Transaction do
   before do
     @basepath = Puppet.features.posix? ? "/what/ever" : "C:/tmp"
     @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
   end
 
   it "should delegate its event list to the event manager" do
     @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
     @transaction.event_manager.expects(:events).returns %w{my events}
     @transaction.events.should == %w{my events}
   end
 
   it "should delegate adding times to its report" do
     @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
     @transaction.report.expects(:add_times).with(:foo, 10)
     @transaction.report.expects(:add_times).with(:bar, 20)
 
     @transaction.add_times :foo => 10, :bar => 20
   end
 
   it "should be able to accept resource status instances" do
     resource = Puppet::Type.type(:notify).new :title => "foobar"
     status = Puppet::Resource::Status.new(resource)
     @transaction.add_resource_status(status)
     @transaction.resource_status(resource).should equal(status)
   end
 
   it "should be able to look resource status up by resource reference" do
     resource = Puppet::Type.type(:notify).new :title => "foobar"
     status = Puppet::Resource::Status.new(resource)
     @transaction.add_resource_status(status)
     @transaction.resource_status(resource.to_s).should equal(status)
   end
 
   # This will basically only ever be used during testing.
   it "should automatically create resource statuses if asked for a non-existent status" do
     resource = Puppet::Type.type(:notify).new :title => "foobar"
     @transaction.resource_status(resource).should be_instance_of(Puppet::Resource::Status)
   end
 
   it "should add provided resource statuses to its report" do
     resource = Puppet::Type.type(:notify).new :title => "foobar"
     status = Puppet::Resource::Status.new(resource)
     @transaction.add_resource_status(status)
     @transaction.report.resource_statuses[resource.to_s].should equal(status)
   end
 
   it "should consider a resource to be failed if a status instance exists for that resource and indicates it is failed" do
     resource = Puppet::Type.type(:notify).new :name => "yayness"
     status = Puppet::Resource::Status.new(resource)
     status.failed = "some message"
     @transaction.add_resource_status(status)
     @transaction.should be_failed(resource)
   end
 
   it "should not consider a resource to be failed if a status instance exists for that resource but indicates it is not failed" do
     resource = Puppet::Type.type(:notify).new :name => "yayness"
     status = Puppet::Resource::Status.new(resource)
     @transaction.add_resource_status(status)
     @transaction.should_not be_failed(resource)
   end
 
   it "should consider there to be failed resources if any statuses are marked failed" do
     resource = Puppet::Type.type(:notify).new :name => "yayness"
     status = Puppet::Resource::Status.new(resource)
     status.failed = "some message"
     @transaction.add_resource_status(status)
     @transaction.should be_any_failed
   end
 
   it "should not consider there to be failed resources if no statuses are marked failed" do
     resource = Puppet::Type.type(:notify).new :name => "yayness"
     status = Puppet::Resource::Status.new(resource)
     @transaction.add_resource_status(status)
     @transaction.should_not be_any_failed
   end
 
   it "should be possible to replace the report object" do
     report = Puppet::Transaction::Report.new("apply")
     @transaction.report = report
 
     @transaction.report.should == report
   end
 
   it "should consider a resource to have failed dependencies if any of its dependencies are failed"
 
   describe "when initializing" do
     it "should create an event manager" do
       @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
       @transaction.event_manager.should be_instance_of(Puppet::Transaction::EventManager)
       @transaction.event_manager.transaction.should equal(@transaction)
     end
 
     it "should create a resource harness" do
       @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
       @transaction.resource_harness.should be_instance_of(Puppet::Transaction::ResourceHarness)
       @transaction.resource_harness.transaction.should equal(@transaction)
     end
   end
 
   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 = Puppet::Type.type(:file).new :path => @basepath
     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, nil)
 
       @transaction.eval_resource(@resource)
     end
 
     it "should process events" do
       @transaction.event_manager.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 mark the resource's status as skipped" do
         @transaction.eval_resource(@resource)
         @transaction.resource_status(@resource).should be_skipped
       end
     end
   end
 
   describe "when applying a resource" do
     before do
       @resource = Puppet::Type.type(:file).new :path => @basepath
       @status = Puppet::Resource::Status.new(@resource)
 
       @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
       @transaction.event_manager.stubs(:queue_events)
       @transaction.resource_harness.stubs(:evaluate).returns(@status)
     end
 
     it "should use its resource harness to apply the resource" do
       @transaction.resource_harness.expects(:evaluate).with(@resource)
       @transaction.apply(@resource)
     end
 
     it "should add the resulting resource status to its status list" do
       @transaction.apply(@resource)
       @transaction.resource_status(@resource).should be_instance_of(Puppet::Resource::Status)
     end
 
     it "should queue any events added to the resource status" do
       @status.expects(:events).returns %w{a b}
       @transaction.event_manager.expects(:queue_events).with(@resource, ["a", "b"])
       @transaction.apply(@resource)
     end
 
     it "should log and skip any resources that cannot be applied" do
       @transaction.resource_harness.expects(:evaluate).raises ArgumentError
       @resource.expects(:err)
       @transaction.apply(@resource)
       @transaction.report.resource_statuses[@resource.to_s].should be_nil
     end
   end
 
   describe "when generating resources" do
     it "should call 'generate' on all created resources" do
       first = Puppet::Type.type(:notify).new(:name => "first")
       second = Puppet::Type.type(:notify).new(:name => "second")
       third = Puppet::Type.type(:notify).new(:name => "third")
 
       @catalog = Puppet::Resource::Catalog.new
       @transaction = Puppet::Transaction.new(@catalog)
 
       first.expects(:generate).returns [second]
       second.expects(:generate).returns [third]
       third.expects(:generate)
 
       @transaction.generate_additional_resources(first)
     end
 
     it "should finish all resources" do
       generator = stub 'generator', :depthfirst? => true, :tags => []
       resource = stub 'resource', :tag => nil
 
       @catalog = Puppet::Resource::Catalog.new
       @transaction = Puppet::Transaction.new(@catalog)
 
       generator.expects(:generate).returns [resource]
 
       @catalog.expects(:add_resource).yields(resource)
 
       resource.expects(:finish)
 
       @transaction.generate_additional_resources(generator)
     end
 
     it "should skip generated resources that conflict with existing resources" do
       generator = mock 'generator', :tags => []
       resource = stub 'resource', :tag => nil
 
       @catalog = Puppet::Resource::Catalog.new
       @transaction = Puppet::Transaction.new(@catalog)
 
       generator.expects(:generate).returns [resource]
 
       @catalog.expects(:add_resource).raises(Puppet::Resource::Catalog::DuplicateResourceError.new("foo"))
 
       resource.expects(:finish).never
       resource.expects(:info) # log that it's skipped
 
       @transaction.generate_additional_resources(generator).should be_empty
     end
 
     it "should copy all tags to the newly generated resources" do
       child = stub 'child'
       generator = stub 'resource', :tags => ["one", "two"]
 
       @catalog = Puppet::Resource::Catalog.new
       @transaction = Puppet::Transaction.new(@catalog)
 
       generator.stubs(:generate).returns [child]
       @catalog.stubs(:add_resource)
 
       child.expects(:tag).with("one", "two")
       child.expects(:finish)
       generator.expects(:depthfirst?)
 
       @transaction.generate_additional_resources(generator)
     end
   end
 
   describe "when skipping a resource" do
     before :each do
       @resource = Puppet::Type.type(:notify).new :name => "foo"
       @catalog = Puppet::Resource::Catalog.new
       @resource.catalog = @catalog
       @transaction = Puppet::Transaction.new(@catalog)
     end
 
     it "should skip resource with missing tags" do
       @transaction.stubs(:missing_tags?).returns(true)
       @transaction.should be_skip(@resource)
     end
 
     it "should skip unscheduled resources" do
       @transaction.stubs(:scheduled?).returns(false)
       @transaction.should be_skip(@resource)
     end
 
     it "should skip resources with failed dependencies" do
       @transaction.stubs(:failed_dependencies?).returns(true)
       @transaction.should be_skip(@resource)
     end
 
     it "should skip virtual resource" do
       @resource.stubs(:virtual?).returns true
       @transaction.should be_skip(@resource)
     end
   end
 
   describe "when determining if tags are missing" do
     before :each do
       @resource = Puppet::Type.type(:notify).new :name => "foo"
       @catalog = Puppet::Resource::Catalog.new
       @resource.catalog = @catalog
       @transaction = Puppet::Transaction.new(@catalog)
 
       @transaction.stubs(:ignore_tags?).returns false
     end
 
     it "should not be missing tags if tags are being ignored" do
       @transaction.expects(:ignore_tags?).returns true
 
       @resource.expects(:tagged?).never
 
       @transaction.should_not be_missing_tags(@resource)
     end
 
     it "should not be missing tags if the transaction tags are empty" do
       @transaction.tags = []
       @resource.expects(:tagged?).never
       @transaction.should_not be_missing_tags(@resource)
     end
 
     it "should otherwise let the resource determine if it is missing tags" do
       tags = ['one', 'two']
       @transaction.tags = tags
       @resource.expects(:tagged?).with(*tags).returns(false)
       @transaction.should be_missing_tags(@resource)
     end
   end
 
   describe "when determining if a resource should be scheduled" do
     before :each do
       @resource = Puppet::Type.type(:notify).new :name => "foo"
       @catalog = Puppet::Resource::Catalog.new
       @resource.catalog = @catalog
       @transaction = Puppet::Transaction.new(@catalog)
     end
 
     it "should always schedule resources if 'ignoreschedules' is set" do
       @transaction.ignoreschedules = true
       @transaction.resource_harness.expects(:scheduled?).never
 
       @transaction.should be_scheduled(@resource)
     end
 
     it "should let the resource harness determine whether the resource should be scheduled" do
       @transaction.resource_harness.expects(:scheduled?).with(@transaction.resource_status(@resource), @resource).returns "feh"
 
       @transaction.scheduled?(@resource).should == "feh"
     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
 
   it "should return all resources for which the resource status indicates the resource has changed when determinig changed resources" do
     @catalog = Puppet::Resource::Catalog.new
     @transaction = Puppet::Transaction.new(@catalog)
     names = []
     2.times do |i|
       name = File.join(@basepath, "file#{i}")
       resource = Puppet::Type.type(:file).new :path => name
       names << resource.to_s
       @catalog.add_resource resource
       @transaction.add_resource_status Puppet::Resource::Status.new(resource)
     end
 
     @transaction.resource_status(names[0]).changed = true
 
     @transaction.changed?.should == [@catalog.resource(names[0])]
   end
 
   describe 'when checking application run state' do
     before do
       without_warnings { Puppet::Application = Class.new(Puppet::Application) }
       @catalog = Puppet::Resource::Catalog.new
       @transaction = Puppet::Transaction.new(@catalog)
     end
 
     after do
       without_warnings { Puppet::Application = Puppet::Application.superclass }
     end
 
     it 'should return true for :stop_processing? if Puppet::Application.stop_requested? is true' do
       Puppet::Application.stubs(:stop_requested?).returns(true)
       @transaction.stop_processing?.should be_true
     end
 
     it 'should return false for :stop_processing? if Puppet::Application.stop_requested? is false' do
       Puppet::Application.stubs(:stop_requested?).returns(false)
       @transaction.stop_processing?.should be_false
     end
 
     describe 'within an evaluate call' do
       before do
         @resource = stub 'resource', :ref => 'some_ref'
+        @relationship_graph = stub 'relationship_graph'
         @catalog.add_resource @resource
         @transaction.stubs(:prepare)
-        @transaction.stubs(:visit_resources).yields(@resource)
+        @transaction.stubs(:relationship_graph).returns @relationship_graph
+        @relationship_graph.stubs(:traverse).yields @resource
       end
 
       it 'should stop processing if :stop_processing? is true' do
         @transaction.expects(:stop_processing?).returns(true)
         @transaction.expects(:eval_resource).never
         @transaction.evaluate
       end
 
       it 'should continue processing if :stop_processing? is false' do
         @transaction.expects(:stop_processing?).returns(false)
         @transaction.expects(:eval_resource).returns(nil)
         @transaction.evaluate
       end
     end
   end
 end
 
 describe Puppet::Transaction, " when determining tags" do
   before do
     @config = Puppet::Resource::Catalog.new
     @transaction = Puppet::Transaction.new(@config)
   end
 
   it "should default to the tags specified in the :tags setting" do
     Puppet.expects(:[]).with(:tags).returns("one")
     @transaction.tags.should == %w{one}
   end
 
   it "should split tags based on ','" do
     Puppet.expects(:[]).with(:tags).returns("one,two")
     @transaction.tags.should == %w{one two}
   end
 
   it "should use any tags set after creation" do
     Puppet.expects(:[]).with(:tags).never
     @transaction.tags = %w{one two}
     @transaction.tags.should == %w{one two}
   end
 
   it "should always convert assigned tags to an array" do
     @transaction.tags = "one::two"
     @transaction.tags.should == %w{one::two}
   end
 
   it "should accept a comma-delimited string" do
     @transaction.tags = "one, two"
     @transaction.tags.should == %w{one two}
   end
 
   it "should accept an empty string" do
     @transaction.tags = ""
     @transaction.tags.should == []
   end
 end