Repository: nats-io/nats-server
Pull request #1709 (2020-11-14 01:14:32)

Fix bug when removing the same message from a stream.
We would release locks and call into upper layers when removing a message.
The upper layers may call back into the lower layers to get more information, such as the subject.
This fix has the storage updates optionally supply the subject for filtered consumers and fixes the bug of double deletes.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 26c6d983..ca3592ce 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -2117,19 +2117,15 @@ func (o *Consumer) incStreamPending(sseq uint64, subj string) {
 	o.mu.Unlock()
 }
 
-func (o *Consumer) decStreamPending(sseq uint64) {
+func (o *Consumer) decStreamPending(sseq uint64, subj string) {
 	o.mu.Lock()
-	// Ignore if we have already sent this one.
+	defer o.mu.Unlock()
+
+	// Ignore if we have already seen this one.
 	if sseq < o.sseq || o.sgap == 0 {
-		o.mu.Unlock()
 		return
 	}
-	if o.config.FilterSubject == _EMPTY_ {
+	if o.isFilteredMatch(subj) {
 		o.sgap--
-	} else if o.mset != nil {
-		if subj, _, _, _, _ := o.mset.store.LoadMsg(sseq); o.isFilteredMatch(subj) {
-			o.sgap--
-		}
 	}
-	o.mu.Unlock()
 }
diff --git a/server/filestore.go b/server/filestore.go
index ce80af3b..f5e91100 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -66,7 +66,7 @@ type FileConsumerInfo struct {
 type fileStore struct {
 	mu       sync.RWMutex
 	state    StreamState
-	scb      func(int64, int64, uint64)
+	scb      StorageUpdateHandler
 	ageChk   *time.Timer
 	syncTmr  *time.Timer
 	cfg      FileStreamInfo
@@ -551,13 +551,13 @@ func (fs *fileStore) GetSeqFromTime(t time.Time) uint64 {
 // RegisterStorageUpdates registers a callback for updates to storage changes.
 // It will present number of messages and bytes as a signed integer and an
 // optional sequence number of the message if a single.
-func (fs *fileStore) RegisterStorageUpdates(cb func(int64, int64, uint64)) {
+func (fs *fileStore) RegisterStorageUpdates(cb StorageUpdateHandler) {
 	fs.mu.Lock()
 	fs.scb = cb
 	bsz := fs.state.Bytes
 	fs.mu.Unlock()
 	if cb != nil && bsz > 0 {
-		cb(0, int64(bsz), 0)
+		cb(0, int64(bsz), 0, _EMPTY_)
 	}
 }
 
@@ -688,7 +688,7 @@ func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, erro
 
 	// Update the upper layers regarding storage used.
 	if cb != nil {
-		cb(1, int64(n), seq)
+		cb(1, int64(n), seq, subj)
 	}
 
 	return seq, ts, nil
@@ -824,9 +824,19 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 		return false, err
 	}
 
+	// If we have a callback grab the message since we need the subject.
+	// TODO(dlc) - This will cause whole buffer to be loaded which I was trying
+	// to avoid. Maybe use side cache for subjects or understand when we really need them.
+	// Meaning if the stream above is only a single subject no need to store, this is just
+	// for updating stream pending for consumers.
+	var sm *fileStoredMsg
+	if fs.scb != nil {
+		sm, _ = mb.fetchMsg(seq)
+	}
+
 	mb.mu.Lock()
 
-	// Check cache. This will be very rare.
+	// Check cache. This should be very rare.
 	if mb.cache == nil || mb.cache.idx == nil {
 		mb.mu.Unlock()
 		fs.mu.Unlock()
@@ -861,16 +871,6 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	ri, rl, _, _ := mb.slotInfo(int(slot))
 	msz := uint64(rl)
 
-	// Call this early so that we can look this message up if needed.
-	if cb := fs.scb; cb != nil {
-		mb.mu.Unlock()
-		fs.mu.Unlock()
-		delta := int64(msz)
-		cb(-1, -delta, seq)
-		fs.mu.Lock()
-		mb.mu.Lock()
-	}
-
 	// Global stats
 	fs.state.Msgs--
 	fs.state.Bytes -= msz
@@ -913,6 +913,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 		qch = mb.qch
 		fch = mb.fch
 	}
+	cb := fs.scb
 	mb.mu.Unlock()
 
 	// Kick outside of lock.
@@ -933,6 +934,16 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	}
 	fs.mu.Unlock()
 
+	// Storage updates.
+	if cb != nil {
+		subj := _EMPTY_
+		if sm != nil {
+			subj = sm.subj
+		}
+		delta := int64(msz)
+		cb(-1, -delta, seq, subj)
+	}
+
 	return true, nil
 }
 
@@ -1773,13 +1784,14 @@ func (mb *msgBlock) flushPendingMsgs() error {
 
 	// Only one can be flushing at a time.
 	mb.setFlushing()
+	mfd := mb.mfd
 	mb.mu.Unlock()
 
 	var tn int
 
 	// Append new data to the message block file.
 	for lbb := lob; lbb > 0; lbb = len(buf) {
-		n, err := mb.mfd.WriteAt(buf, woff)
+		n, err := mfd.WriteAt(buf, woff)
 		if err != nil {
 			// FIXME(dlc) - What is the correct behavior here?
 			mb.removeIndexFile()
@@ -2350,7 +2362,7 @@ func (fs *fileStore) Purge() uint64 {
 	fs.mu.Unlock()
 
 	if cb != nil {
-		cb(-int64(purged), -rbytes, 0)
+		cb(-int64(purged), -rbytes, 0, _EMPTY_)
 	}
 
 	return purged
diff --git a/server/memstore.go b/server/memstore.go
index 43c866fe..971ed45b 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -27,7 +27,7 @@ type memStore struct {
 	cfg       StreamConfig
 	state     StreamState
 	msgs      map[uint64]*storedMsg
-	scb       func(int64, int64, uint64)
+	scb       StorageUpdateHandler
 	ageChk    *time.Timer
 	consumers int
 }
@@ -134,7 +134,7 @@ func (ms *memStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, error
 	ms.mu.Unlock()
 
 	if cb != nil {
-		cb(1, stopBytes-startBytes, seq)
+		cb(1, stopBytes-startBytes, seq, subj)
 	}
 
 	return seq, ts, nil
@@ -161,7 +161,7 @@ func (ms *memStore) SkipMsg() uint64 {
 // RegisterStorageUpdates registers a callback for updates to storage changes.
 // It will present number of messages and bytes as a signed integer and an
 // optional sequence number of the message if a single.
-func (ms *memStore) RegisterStorageUpdates(cb func(int64, int64, uint64)) {
+func (ms *memStore) RegisterStorageUpdates(cb StorageUpdateHandler) {
 	ms.mu.Lock()
 	ms.scb = cb
 	ms.mu.Unlock()
@@ -267,7 +267,7 @@ func (ms *memStore) Purge() uint64 {
 	ms.mu.Unlock()
 
 	if cb != nil {
-		cb(-int64(purged), -bytes, 0)
+		cb(-int64(purged), -bytes, 0, _EMPTY_)
 	}
 
 	return purged
@@ -351,13 +351,6 @@ func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 
 	ss = memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 
-	if ms.scb != nil {
-		delta := int64(ss)
-		ms.mu.Unlock()
-		ms.scb(-1, -delta, seq)
-		ms.mu.Lock()
-	}
-
 	delete(ms.msgs, seq)
 	ms.state.Msgs--
 	ms.state.Bytes -= ss
@@ -375,6 +368,13 @@ func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 		sm.seq, sm.ts = 0, 0
 	}
 
+	if ms.scb != nil {
+		ms.mu.Unlock()
+		delta := int64(ss)
+		ms.scb(-1, -delta, seq, sm.subj)
+		ms.mu.Lock()
+	}
+
 	return ok
 }
 
diff --git a/server/store.go b/server/store.go
index 6818715d..bfa41bc4 100644
--- a/server/store.go
+++ b/server/store.go
@@ -56,6 +56,10 @@ var (
 	ErrNoAckPolicy = errors.New("ack policy is none")
 )
 
+// Used to call back into the upper layers to report on changes in storage resources.
+// For the cases where its a single message we will also supply sequence number and subject.
+type StorageUpdateHandler func(msgs, bytes int64, seq uint64, subj string)
+
 type StreamStore interface {
 	StoreMsg(subj string, hdr, msg []byte) (uint64, int64, error)
 	SkipMsg() uint64
@@ -65,7 +69,7 @@ type StreamStore interface {
 	Purge() uint64
 	GetSeqFromTime(t time.Time) uint64
 	State() StreamState
-	RegisterStorageUpdates(func(int64, int64, uint64))
+	RegisterStorageUpdates(StorageUpdateHandler)
 	UpdateConfig(cfg *StreamConfig) error
 	Delete() error
 	Stop() error
diff --git a/server/stream.go b/server/stream.go
index a4aaa3ec..2415d27f 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -759,23 +759,17 @@ func (mset *Stream) setupStore(fsCfg *FileStoreConfig) error {
 // jetstream account. We do local processing for stream pending for consumers, but only
 // for removals.
 // Lock should not ne held.
-func (mset *Stream) storeUpdates(md, bd int64, seq uint64) {
-	var _obs [4]*Consumer
-	obs := _obs[:0]
-
+func (mset *Stream) storeUpdates(md, bd int64, seq uint64, subj string) {
 	// If we have a single negative update then we will process our consumers for stream pending.
 	// Purge and Store handled separately inside individual calls.
 	if md == -1 {
-		mset.mu.Lock()
+		mset.mu.RLock()
 		for _, o := range mset.consumers {
-			obs = append(obs, o)
+			o.decStreamPending(seq, subj)
 		}
-		mset.mu.Unlock()
+		mset.mu.RUnlock()
 	}
 
-	for _, o := range obs {
-		o.decStreamPending(seq)
-	}
 	if mset.jsa != nil {
 		mset.jsa.updateUsage(mset.config.Storage, bd)
 	}


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 8438a09e..ff5570b5 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1922,6 +1922,8 @@ func TestFileStoreStoreLimitRemovePerf(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	fs.RegisterStorageUpdates(func(md, bd int64, seq uint64, subj string) {})
+
 	fmt.Printf("storing and removing (limit 1) %d msgs of %s each, totalling %s\n",
 		toStore,
 		FriendlyBytes(int64(storedMsgSize)),
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index de9abbf0..782d8801 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -8942,8 +8942,12 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 				}
 			}
 			// To make sure acks are processed for checking state after sending new ones.
-			nc1.Flush()
-			nc2.Flush()
+			checkFor(t, time.Second, 25*time.Millisecond, func() error {
+				if state = mset.State(); state.Msgs != 0 {
+					return fmt.Errorf("Stream still has messages")
+				}
+				return nil
+			})
 
 			// Now close the 2nd subscription...
 			sub2.Unsubscribe()