Index: /trunk/downstream.go
===================================================================
--- /trunk/downstream.go	(revision 56)
+++ /trunk/downstream.go	(revision 57)
@@ -41,13 +41,14 @@
 
 type downstreamConn struct {
-	net      net.Conn
-	irc      *irc.Conn
-	srv      *Server
-	logger   Logger
-	messages chan *irc.Message
+	net       net.Conn
+	irc       *irc.Conn
+	srv       *Server
+	logger    Logger
+	messages  chan *irc.Message
+	consumers chan *RingConsumer
+	closed    chan struct{}
 
 	registered bool
 	user       *user
-	closed     bool
 	nick       string
 	username   string
@@ -57,9 +58,11 @@
 func newDownstreamConn(srv *Server, netConn net.Conn) *downstreamConn {
 	dc := &downstreamConn{
-		net:      netConn,
-		irc:      irc.NewConn(netConn),
-		srv:      srv,
-		logger:   &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
-		messages: make(chan *irc.Message, 64),
+		net:       netConn,
+		irc:       irc.NewConn(netConn),
+		srv:       srv,
+		logger:    &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
+		messages:  make(chan *irc.Message, 64),
+		consumers: make(chan *RingConsumer),
+		closed:    make(chan struct{}),
 	}
 
@@ -83,4 +86,13 @@
 		User: dc.username,
 		// TODO: fill the host?
+	}
+}
+
+func (dc *downstreamConn) isClosed() bool {
+	select {
+	case <-dc.closed:
+		return true
+	default:
+		return false
 	}
 }
@@ -105,5 +117,5 @@
 		}
 
-		if dc.closed {
+		if dc.isClosed() {
 			return nil
 		}
@@ -114,7 +126,30 @@
 
 func (dc *downstreamConn) writeMessages() error {
-	for msg := range dc.messages {
-		if err := dc.irc.WriteMessage(msg); err != nil {
-			return err
+	for {
+		var err error
+		var closed bool
+		select {
+		case msg := <-dc.messages:
+			err = dc.irc.WriteMessage(msg)
+		case consumer := <-dc.consumers:
+			for {
+				msg := consumer.Peek()
+				if msg == nil {
+					break
+				}
+				err = dc.irc.WriteMessage(msg)
+				if err != nil {
+					break
+				}
+				consumer.Consume()
+			}
+		case <-dc.closed:
+			closed = true
+		}
+		if err != nil {
+			return err
+		}
+		if closed {
+			break
 		}
 	}
@@ -123,5 +158,5 @@
 
 func (dc *downstreamConn) Close() error {
-	if dc.closed {
+	if dc.isClosed() {
 		return fmt.Errorf("downstream connection already closed")
 	}
@@ -135,15 +170,7 @@
 		}
 		u.lock.Unlock()
-
-		// TODO: figure out a better way to advance the ring buffer consumer cursor
-		u.forEachUpstream(func(uc *upstreamConn) {
-			// TODO: let clients specify the ring buffer name in their username
-			uc.ring.Consumer("").Reset()
-		})
-	}
-
-	close(dc.messages)
-	dc.closed = true
-
+	}
+
+	close(dc.closed)
 	return nil
 }
@@ -212,4 +239,5 @@
 
 	u.lock.Lock()
+	firstDownstream := len(u.downstreamConns) == 0
 	u.downstreamConns = append(u.downstreamConns, dc)
 	u.lock.Unlock()
@@ -250,13 +278,39 @@
 
 		// TODO: let clients specify the ring buffer name in their username
-		consumer := uc.ring.Consumer("")
-		for {
-			// TODO: these messages will get lost if the connection is closed
-			msg := consumer.Consume()
-			if msg == nil {
-				break
-			}
-			dc.SendMessage(msg)
-		}
+		historyName := ""
+
+		var seqPtr *uint64
+		if firstDownstream {
+			seq, ok := uc.history[historyName]
+			if ok {
+				seqPtr = &seq
+			}
+		}
+
+		consumer, ch := uc.ring.Consumer(seqPtr)
+		go func() {
+			for {
+				var closed bool
+				select {
+				case <-ch:
+					dc.consumers <- consumer
+				case <-dc.closed:
+					closed = true
+				}
+				if closed {
+					break
+				}
+			}
+
+			seq := consumer.Close()
+
+			dc.user.lock.Lock()
+			lastDownstream := len(dc.user.downstreamConns) == 0
+			dc.user.lock.Unlock()
+
+			if lastDownstream {
+				uc.history[historyName] = seq
+			}
+		}()
 	})
 
Index: /trunk/ring.go
===================================================================
--- /trunk/ring.go	(revision 56)
+++ /trunk/ring.go	(revision 57)
@@ -2,4 +2,6 @@
 
 import (
+	"sync"
+
 	"gopkg.in/irc.v3"
 )
@@ -8,44 +10,67 @@
 // buffer size is fixed. The ring buffer is stored in memory.
 type Ring struct {
-	buffer   []*irc.Message
-	cap, cur uint64
+	buffer []*irc.Message
+	cap    uint64
 
-	consumers map[string]*RingConsumer
+	lock      sync.Mutex
+	cur       uint64
+	consumers []*RingConsumer
 }
 
 func NewRing(capacity int) *Ring {
 	return &Ring{
-		buffer:    make([]*irc.Message, capacity),
-		cap:       uint64(capacity),
-		consumers: make(map[string]*RingConsumer),
+		buffer: make([]*irc.Message, capacity),
+		cap:    uint64(capacity),
 	}
 }
 
 func (r *Ring) Produce(msg *irc.Message) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
 	i := int(r.cur % r.cap)
 	r.buffer[i] = msg
 	r.cur++
+
+	for _, consumer := range r.consumers {
+		select {
+		case consumer.ch <- struct{}{}:
+			// This space is intentionally left blank
+		default:
+			// The channel already has a pending item
+		}
+	}
 }
 
-func (r *Ring) Consumer(name string) *RingConsumer {
-	consumer, ok := r.consumers[name]
-	if ok {
-		return consumer
+func (r *Ring) Consumer(seq *uint64) (*RingConsumer, <-chan struct{}) {
+	consumer := &RingConsumer{
+		ring: r,
+		ch:   make(chan struct{}, 1),
 	}
 
-	consumer = &RingConsumer{
-		ring: r,
-		cur:  r.cur,
+	r.lock.Lock()
+	if seq != nil {
+		consumer.cur = *seq
+	} else {
+		consumer.cur = r.cur
 	}
-	r.consumers[name] = consumer
-	return consumer
+	if consumer.diff() > 0 {
+		consumer.ch <- struct{}{}
+	}
+	r.consumers = append(r.consumers, consumer)
+	r.lock.Unlock()
+
+	return consumer, consumer.ch
 }
 
 type RingConsumer struct {
-	ring *Ring
-	cur  uint64
+	ring   *Ring
+	cur    uint64
+	ch     chan struct{}
+	closed bool
 }
 
-func (rc *RingConsumer) Diff() uint64 {
+// diff returns the number of pending messages. It assumes the Ring is locked.
+func (rc *RingConsumer) diff() uint64 {
 	if rc.cur > rc.ring.cur {
 		panic("jounce: consumer cursor greater than producer cursor")
@@ -55,5 +80,12 @@
 
 func (rc *RingConsumer) Peek() *irc.Message {
-	diff := rc.Diff()
+	if rc.closed {
+		panic("jounce: RingConsumer.Peek called after Close")
+	}
+
+	rc.ring.lock.Lock()
+	defer rc.ring.lock.Unlock()
+
+	diff := rc.diff()
 	if diff == 0 {
 		return nil
@@ -79,5 +111,16 @@
 }
 
-func (rc *RingConsumer) Reset() {
-	rc.cur = rc.ring.cur
+func (rc *RingConsumer) Close() uint64 {
+	rc.ring.lock.Lock()
+	for i := range rc.ring.consumers {
+		if rc.ring.consumers[i] == rc {
+			rc.ring.consumers = append(rc.ring.consumers[:i], rc.ring.consumers[i+1:]...)
+			break
+		}
+	}
+	rc.ring.lock.Unlock()
+
+	close(rc.ch)
+	rc.closed = true
+	return rc.cur
 }
Index: /trunk/server.go
===================================================================
--- /trunk/server.go	(revision 56)
+++ /trunk/server.go	(revision 57)
@@ -39,4 +39,11 @@
 	upstreamConns   []*upstreamConn
 	downstreamConns []*downstreamConn
+}
+
+func newUser(srv *Server, username string) *user {
+	return &user{
+		username: username,
+		srv:      srv,
+	}
 }
 
@@ -117,5 +124,5 @@
 func (s *Server) Run() {
 	// TODO: multi-user
-	u := &user{username: "jounce", srv: s}
+	u := newUser(s, "jounce")
 
 	s.lock.Lock()
Index: /trunk/upstream.go
===================================================================
--- /trunk/upstream.go	(revision 56)
+++ /trunk/upstream.go	(revision 57)
@@ -45,4 +45,5 @@
 	modes      modeSet
 	channels   map[string]*upstreamChannel
+	history    map[string]uint64
 }
 
@@ -67,4 +68,5 @@
 		ring:     NewRing(u.srv.RingCap),
 		channels: make(map[string]*upstreamChannel),
+		history:  make(map[string]uint64),
 	}
 
@@ -306,7 +308,4 @@
 	case "PRIVMSG":
 		uc.ring.Produce(msg)
-		uc.user.forEachDownstream(func(dc *downstreamConn) {
-			dc.SendMessage(msg)
-		})
 	case irc.RPL_YOURHOST, irc.RPL_CREATED:
 		// Ignore
