Avoid filestore corruption by rejectinglarge publishes into JetStream

We have an effective and seemingly undocumented limit of 32MB for messages
published into JetStream as `indexCacheBuf` in the filestore rejects any
records that exceed this length. We did not catch this on publish however,
and then `indexCacheBuf` would later error when consuming etc.

Signed-off-by: Neil Twigg <[email protected]>

Author: neilalexander

server/filestore.go

@@ -5976,7 +5976,7 @@ func (fs *fileStore) writeMsgRecord(seq uint64, ts int64, subj string, hdr, msg
 
 	// Get size for this message.
 	rl := fileStoreMsgSize(subj, hdr, msg)
-	if rl&hbit != 0 {
+	if rl&hbit != 0 || rl > rlBadThresh {
 		return 0, ErrMsgTooLarge
 	}
 	// Grab our current last message block.

server/jetstream_test.go

@@ -19755,3 +19755,35 @@ func TestJetStreamTHWExpireTasksRace(t *testing.T) {
 		})
 	}
 }
+
+func TestJetStreamRejectLargePublishes(t *testing.T) {
+	tdir := t.TempDir()
+
+	// The test relies on the MaxPayload being larger than the
+	// rlBadThresh, otherwise you can't publish a message large
+	// enough.
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		max_payload: %d
+		jetstream: {store_dir: %q}
+	`, rlBadThresh+2048, tdir)))
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"test"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.Publish("test", make([]byte, rlBadThresh-1024))
+	require_NoError(t, err)
+
+	_, err = js.Publish("test", make([]byte, rlBadThresh+1024))
+	require_Error(t, err)
+	require_Contains(t, err.Error(), ErrMsgTooLarge.Error())
+}