Index: /trunk/doc/architecture.md
===================================================================
--- /trunk/doc/architecture.md	(revision 226)
+++ /trunk/doc/architecture.md	(revision 227)
@@ -32,6 +32,2 @@
 and downstream message handlers are called from this goroutine, thus they can
 safely access both upstream and downstream state.
-
-In addition to these goroutines, each downstream connection also has one
-goroutine per network to handle new upstream messages coming from the ring
-buffer.
Index: /trunk/downstream.go
===================================================================
--- /trunk/downstream.go	(revision 226)
+++ /trunk/downstream.go	(revision 227)
@@ -232,4 +232,27 @@
 
 	dc.conn.SendMessage(msg)
+}
+
+func (dc *downstreamConn) sendFromUpstream(msg *irc.Message, uc *upstreamConn) {
+	dc.lock.Lock()
+	_, ours := dc.ourMessages[msg]
+	delete(dc.ourMessages, msg)
+	dc.lock.Unlock()
+	if ours && !dc.getCap("echo-message") {
+		// The message comes from our connection, don't echo it
+		// back
+		return
+	}
+
+	msg = msg.Copy()
+	switch msg.Command {
+	case "PRIVMSG", "NOTICE":
+		msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
+		msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
+	default:
+		panic(fmt.Sprintf("unexpected %q message", msg.Command))
+	}
+
+	dc.SendMessage(msg)
 }
 
@@ -664,73 +687,41 @@
 
 	dc.forEachNetwork(func(net *network) {
-		dc.runNetwork(net, sendHistory)
+		var seqPtr *uint64
+		if sendHistory {
+			seq, ok := net.history[dc.clientName]
+			if ok {
+				seqPtr = &seq
+			}
+		}
+
+		consumer, _ := net.ring.NewConsumer(seqPtr)
+
+		if _, ok := dc.ringConsumers[net]; ok {
+			panic("network has been added twice")
+		}
+		dc.ringConsumers[net] = consumer
+
+		// TODO: this means all history is lost when trying to send it while the
+		// upstream is disconnected. We need to store history differently so that
+		// we don't need access to upstreamConn to forward it to a downstream
+		// client.
+		uc := net.upstream()
+		if uc == nil {
+			dc.logger.Printf("ignoring messages for upstream %q: upstream is disconnected", net.Addr)
+			return
+		}
+
+		for {
+			msg := consumer.Peek()
+			if msg == nil {
+				break
+			}
+
+			dc.sendFromUpstream(msg, uc)
+			consumer.Consume()
+		}
 	})
 
 	return nil
-}
-
-// runNetwork starts listening for messages coming from the network's ring
-// buffer.
-//
-// It panics if the network is not suitable for the downstream connection.
-func (dc *downstreamConn) runNetwork(net *network, loadHistory bool) {
-	if dc.network != nil && net != dc.network {
-		panic("network not suitable for downstream connection")
-	}
-
-	var seqPtr *uint64
-	if loadHistory {
-		seq, ok := net.history[dc.clientName]
-		if ok {
-			seqPtr = &seq
-		}
-	}
-
-	consumer, ch := net.ring.NewConsumer(seqPtr)
-
-	if _, ok := dc.ringConsumers[net]; ok {
-		panic("network has been added twice")
-	}
-	dc.ringConsumers[net] = consumer
-
-	go func() {
-		for range ch {
-			uc := net.upstream()
-			if uc == nil {
-				dc.logger.Printf("ignoring messages for upstream %q: upstream is disconnected", net.Addr)
-				continue
-			}
-
-			for {
-				msg := consumer.Peek()
-				if msg == nil {
-					break
-				}
-
-				dc.lock.Lock()
-				_, ours := dc.ourMessages[msg]
-				delete(dc.ourMessages, msg)
-				dc.lock.Unlock()
-				if ours && !dc.getCap("echo-message") {
-					// The message comes from our connection, don't echo it
-					// back
-					consumer.Consume()
-					continue
-				}
-
-				msg = msg.Copy()
-				switch msg.Command {
-				case "PRIVMSG", "NOTICE":
-					msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
-					msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
-				default:
-					panic(fmt.Sprintf("unexpected %q message", msg.Command))
-				}
-
-				dc.SendMessage(msg)
-				consumer.Consume()
-			}
-		}
-	}()
 }
 
Index: /trunk/upstream.go
===================================================================
--- /trunk/upstream.go	(revision 226)
+++ /trunk/upstream.go	(revision 227)
@@ -1366,4 +1366,8 @@
 func (uc *upstreamConn) produce(msg *irc.Message) {
 	uc.network.ring.Produce(msg)
+
+	uc.forEachDownstream(func(dc *downstreamConn) {
+		dc.sendFromUpstream(dc.ringConsumers[uc.network].Consume(), uc)
+	})
 }
 
Index: /trunk/user.go
===================================================================
--- /trunk/user.go	(revision 226)
+++ /trunk/user.go	(revision 227)
@@ -352,5 +352,6 @@
 	u.forEachDownstream(func(dc *downstreamConn) {
 		if dc.network == nil {
-			dc.runNetwork(network, false)
+			consumer, _ := network.ring.NewConsumer(nil)
+			dc.ringConsumers[network] = consumer
 		}
 	})
@@ -376,4 +377,5 @@
 				dc.Close()
 			}
+			delete(dc.ringConsumers, net)
 		})
 
