nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: nats-io/nats-server
Pull request #2753 (2021-12-21 02:31:40)

[FIXED] JetStream: stream first/last sequence possibly reset
A low-level Filestore issue would cause a new block to be created
when the last block was empty, but the index for the new block
would not be forced to be written on disk.

The observed issue could be that with a stream with a WorkQueue
retention policy, its first/last sequence values could be reset
after a pull subscriber would have consumed all messages and
the server was restarted without a clean shutdown.
This would cause the pull subscriber to "stall" until enough
new messages are sent to reach a stream sequence that catches
up with the consumer's view of the stream first sequence prior
to the restart.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index acc21ae8..2ca10627 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1872,9 +1872,11 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 	}
 
 	// If we emptied the current message block and the seq was state.First.Seq
-	// then we need to jump message blocks.
+	// then we need to jump message blocks. We will also write the index so
+	// we don't lose track of the first sequence.
 	if firstSeqNeedsUpdate {
 		fs.selectNextFirst()
+		fs.lmb.writeIndexInfo()
 	}
 	fs.mu.Unlock()
 
@@ -3472,7 +3474,7 @@ func (mb *msgBlock) indexNeedsUpdate() bool {
 }
 
 // Write index info to the appropriate file.
-// Lock should be held.
+// Filestore lock should be held.
 func (mb *msgBlock) writeIndexInfo() error {
 	// HEADER: magic version msgs bytes fseq fts lseq lts ndel checksum
 	var hdr [indexHdrSize]byte


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 18c4ab1f..78b684c4 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3449,3 +3449,33 @@ func TestFileStorePurgeExKeepOneBug(t *testing.T) {
 		t.Fatalf("Expected to find 1 `A` msgs, got %d", fss.Msgs)
 	}
 }
+
+func TestFileStoreRemoveLastWriteIndex(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "TEST", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	for i := 0; i < 10; i++ {
+		fs.StoreMsg("foo", nil, []byte("msg"))
+	}
+	for i := 0; i < 10; i++ {
+		fs.RemoveMsg(uint64(i + 1))
+	}
+
+	fs.mu.Lock()
+	fname := fs.lmb.ifn
+	fs.mu.Unlock()
+
+	fi, err := os.Stat(fname)
+	if err != nil {
+		t.Fatalf("Error getting stats for index file %q: %v", fname, err)
+	}
+	if fi.Size() == 0 {
+		t.Fatalf("Index file %q size is 0", fname)
+	}
+}
Repository: census-instrumentation/opencensus-python
Pull request #453 (2019-01-17 20:20:13)

Stats: Fix recording with None TagMap.
The `record` method takes an optional `TagMap`. If `TagMap` is present, recorded measurements would be broken down by the key-value pairs in it. If not, recorded measurements are dimensionless. See also https://github.com/census-instrumentation/opencensus-specs/blob/master/stats/Record.md#recording-stats.

Currently if we record measurements with none `TagMap`, an error will be raised:
```
tag_values = self.get_tag_values(tags=context.map,
AttributeError: 'NoneType' object has no attribute 'map'
```
Fix this bug by checking if `TagMap` is `None` before using it.

songy23: @c24t Updated PR description, PTAL.
c24t: Thanks @songy23!
songy23: Thanks!

Patch:
diff --git a/opencensus/stats/view_data.py b/opencensus/stats/view_data.py
index 67cfbe0..52b43d2 100644
--- a/opencensus/stats/view_data.py
+++ b/opencensus/stats/view_data.py
@@ -83,7 +83,11 @@ class ViewData(object):
 
     def record(self, context, value, timestamp, attachments=None):
         """records the view data against context"""
-        tag_values = self.get_tag_values(tags=context.map,
+        if context is None:
+            tags = dict()
+        else:
+            tags = context.map
+        tag_values = self.get_tag_values(tags=tags,
                                          columns=self.view.columns)
         tuple_vals = tuple(tag_values)
         if tuple_vals not in self.tag_value_aggregation_data_map:


Test patch:
diff --git a/tests/unit/stats/test_view_data.py b/tests/unit/stats/test_view_data.py
index c442d1e..41c9aaf 100644
--- a/tests/unit/stats/test_view_data.py
+++ b/tests/unit/stats/test_view_data.py
@@ -293,3 +293,24 @@ class TestViewData(unittest.TestCase):
         self.assertTrue(tuple_vals in view_data.tag_value_aggregation_data_map)
         sum_data = view_data.tag_value_aggregation_data_map.get(tuple_vals)
         self.assertEqual(4, sum_data.sum_data)
+
+    def test_record_with_none_context(self):
+        measure = mock.Mock()
+        sum_aggregation = aggregation_module.SumAggregation()
+        view = view_module.View("test_view", "description", ['key1', 'key2'],
+                                measure, sum_aggregation)
+        start_time = datetime.utcnow()
+        end_time = datetime.utcnow()
+        view_data = view_data_module.ViewData(
+            view=view, start_time=start_time, end_time=end_time)
+        time = datetime.utcnow().isoformat() + 'Z'
+        value = 4
+        view_data.record(
+            context=None, value=value, timestamp=time, attachments=None)
+        tag_values = view_data.get_tag_values(
+            tags={}, columns=view.columns)
+        tuple_vals = tuple(tag_values)
+        self.assertEqual([None, None], tag_values)
+        self.assertTrue(tuple_vals in view_data.tag_value_aggregation_data_map)
+        sum_data = view_data.tag_value_aggregation_data_map.get(tuple_vals)
+        self.assertEqual(4, sum_data.sum_data)
Repository: nats-io/nats-server
Pull request #2835 (2022-02-01 20:19:52)

Consumer pending gets stuck.
Under certain scenarios the pending for a consumer could appear to get stuck.

Under the covers we were calculating pending per msg block incorrectly when a single message existed beyond the requested sequence.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 0aad6e21..d4b299e5 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1115,7 +1115,7 @@ func (mb *msgBlock) filteredPendingLocked(subj string, wc bool, seq uint64) (tot
 	for i, subj := range subs {
 		// If the starting seq is less then or equal that means we want all and we do not need to load any messages.
 		ss := mb.fss[subj]
-		if ss == nil {
+		if ss == nil || seq > ss.Last {
 			continue
 		}
 


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index e9f4a5e1..a6316024 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3480,6 +3480,33 @@ func TestFileStoreRemoveLastWriteIndex(t *testing.T) {
 	}
 }
 
+func TestFileStoreFilteredPendingBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "TEST", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	fs.StoreMsg("foo", nil, []byte("msg"))
+	fs.StoreMsg("bar", nil, []byte("msg"))
+	fs.StoreMsg("baz", nil, []byte("msg"))
+
+	fs.mu.Lock()
+	mb := fs.lmb
+	fs.mu.Unlock()
+
+	total, f, l := mb.filteredPending("foo", false, 3)
+	if total != 0 {
+		t.Fatalf("Expected total of 0 but got %d", total)
+	}
+	if f != 0 || l != 0 {
+		t.Fatalf("Expected first and last to be 0 as well, but got %d %d", f, l)
+	}
+}
+
 // Test to optimize the selectMsgBlock with lots of blocks.
 func TestFileStoreFetchPerf(t *testing.T) {
 	// Comment out to run.