(2.12) [ADDED] Offline assets support (#7158)

Implements [Offline Assets support from
ADR-44](https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-44.md).
Streams and consumers will be put in "offline mode" if the server
doesn't support the required API level.

Example:
```
[INF] ---------------- JETSTREAM ----------------
...
[INF]   API Level:       2
[INF] -------------------------------------------
[WRN]   Detected unsupported stream 'js > test-stream', delete the stream or upgrade the server to API level 10
[INF] Starting JetStream cluster
[INF] Creating JetStream metadata controller
[INF] JetStream cluster recovering state
[INF] Listening for client connections on 0.0.0.0:4444
[WRN] Detected unsupported stream 'js > test-stream', delete the stream or upgrade the server to API level 10
[INF] Server is ready
[INF] Cluster name is nats-cluster
```

Stream/consumer list and names requests still return the names of these
streams. Stream/consumer info still works in this state, and a
stream/consumer can still be deleted as well. The "offline reason" is
included as an error in the stream/consumer info requests and is
contained in the `offline: map[string]string` field in the list
requests.

These issues will be logged during startup, if an update request is
performed. The raw JSON is always preserved, so downgrading to a server
that doesn't understand the new JSON fields can safely upgrade and the
JSON will be preserved as expected. Importantly, if a stream or consumer
is recognized as unsupported, they are not even loaded by the server.
This means data will be safe and not even be loaded on an older version
server.

This will allow for more graceful downgrades from 2.12 to 2.11 (and
onward), where loads of new features would stop functioning after a
downgrade: counter-streams, atomic batch, etc.

Signed-off-by: Maurice van Veen <[email protected]>

Author: neilalexander

server/consumer.go

@@ -21,6 +21,8 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"os"
+	"path/filepath"
 	"reflect"
 	"regexp"
 	"slices"
@@ -71,6 +73,13 @@ type ConsumerInfo struct {
 	PriorityGroups []PriorityGroupState `json:"priority_groups,omitempty"`
 }
 
+// consumerInfoClusterResponse is a response used in a cluster to communicate the consumer info
+// back to the meta leader as part of a consumer list request.
+type consumerInfoClusterResponse struct {
+	ConsumerInfo
+	OfflineReason string `json:"offline_reason,omitempty"` // Reporting when a consumer is offline.
+}
+
 type PriorityGroupState struct {
 	Group          string    `json:"group"`
 	PinnedClientID string    `json:"pinned_client_id,omitempty"`
@@ -510,6 +519,10 @@ type consumer struct {
 	/// pinnedTtl is the remaining time before the current PinId expires.
 	pinnedTtl *time.Timer
 	pinnedTS  time.Time
+
+	// If standalone/single-server, the offline reason needs to be stored directly in the consumer.
+	// Otherwise, if clustered it will be part of the consumer assignment.
+	offlineReason string
 }
 
 // A single subject filter.
@@ -5059,7 +5072,7 @@ func (o *consumer) setMaxPendingBytes(limit int) {
 // This does some quick sanity checks to see if we should re-calculate num pending.
 // Lock should be held.
 func (o *consumer) checkNumPending() uint64 {
-	if o.mset != nil {
+	if o.mset != nil && o.mset.store != nil {
 		var state StreamState
 		o.mset.store.FastState(&state)
 		npc := o.numPending()
@@ -6103,6 +6116,14 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 		} else {
 			err = store.Stop()
 		}
+	} else if dflag {
+		// If there's no store (for example, when it's offline), manually delete the directories.
+		o.mu.RLock()
+		stream, consumer := o.stream, o.name
+		o.mu.RUnlock()
+		accDir := filepath.Join(js.config.StoreDir, a.GetName())
+		consumersDir := filepath.Join(accDir, streamsDir, stream, consumerDir)
+		os.RemoveAll(filepath.Join(consumersDir, consumer))
 	}
 
 	return err

server/errors.json

@@ -1918,5 +1918,25 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSStreamOfflineReasonErrF",
+    "code": 500,
+    "error_code": 10194,
+    "description": "stream is offline: {err}",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerOfflineReasonErrF",
+    "code": 500,
+    "error_code": 10195,
+    "description": "consumer is offline: {err}",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]

server/jetstream.go

@@ -14,6 +14,7 @@
 package server
 
 import (
+	"bytes"
 	"crypto/hmac"
 	"crypto/sha256"
 	"encoding/binary"
@@ -1333,8 +1334,54 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 		}
 
 		var cfg FileStreamInfo
-		if err := json.Unmarshal(buf, &cfg); err != nil {
-			s.Warnf("  Error unmarshalling stream metafile %q: %v", metafile, err)
+		decoder := json.NewDecoder(bytes.NewReader(buf))
+		decoder.DisallowUnknownFields()
+		strictErr := decoder.Decode(&cfg)
+		if strictErr != nil {
+			cfg = FileStreamInfo{}
+			if err := json.Unmarshal(buf, &cfg); err != nil {
+				s.Warnf("  Error unmarshalling stream metafile %q: %v", metafile, err)
+				continue
+			}
+		}
+		if supported := supportsRequiredApiLevel(cfg.Metadata); !supported || strictErr != nil {
+			var offlineReason string
+			if !supported {
+				apiLevel := getRequiredApiLevel(cfg.Metadata)
+				offlineReason = fmt.Sprintf("unsupported - required API level: %s, current API level: %d", apiLevel, JSApiLevel)
+				s.Warnf("  Detected unsupported stream '%s > %s', delete the stream or upgrade the server to API level %s", a.Name, cfg.StreamConfig.Name, apiLevel)
+			} else {
+				offlineReason = fmt.Sprintf("decoding error: %v", strictErr)
+				s.Warnf("  Error unmarshalling stream metafile %q: %v", metafile, strictErr)
+			}
+			singleServerMode := !s.JetStreamIsClustered() && s.standAloneMode()
+			if singleServerMode {
+				// Fake a stream, so we can respond to API requests as single-server.
+				mset := &stream{
+					acc:           a,
+					jsa:           jsa,
+					cfg:           cfg.StreamConfig,
+					js:            js,
+					srv:           s,
+					stype:         cfg.Storage,
+					consumers:     make(map[string]*consumer),
+					active:        false,
+					created:       time.Now().UTC(),
+					offlineReason: offlineReason,
+				}
+				if !cfg.Created.IsZero() {
+					mset.created = cfg.Created
+				}
+				mset.closed.Store(true)
+
+				jsa.mu.Lock()
+				jsa.streams[cfg.Name] = mset
+				jsa.mu.Unlock()
+
+				// Now do the consumers.
+				odir := filepath.Join(sdir, fi.Name(), consumerDir)
+				consumers = append(consumers, &ce{mset, odir})
+			}
 			continue
 		}
 
@@ -1510,13 +1557,66 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			}
 
 			var cfg FileConsumerInfo
-			if err := json.Unmarshal(buf, &cfg); err != nil {
-				s.Warnf("    Error unmarshalling consumer metafile %q: %v", metafile, err)
+			decoder := json.NewDecoder(bytes.NewReader(buf))
+			decoder.DisallowUnknownFields()
+			strictErr := decoder.Decode(&cfg)
+			if strictErr != nil {
+				cfg = FileConsumerInfo{}
+				if err := json.Unmarshal(buf, &cfg); err != nil {
+					s.Warnf("    Error unmarshalling consumer metafile %q: %v", metafile, err)
+					continue
+				}
+			}
+			if supported := supportsRequiredApiLevel(cfg.Metadata); !supported || strictErr != nil {
+				var offlineReason string
+				if !supported {
+					apiLevel := getRequiredApiLevel(cfg.Metadata)
+					offlineReason = fmt.Sprintf("unsupported - required API level: %s, current API level: %d", apiLevel, JSApiLevel)
+					s.Warnf("  Detected unsupported consumer '%s > %s > %s', delete the consumer or upgrade the server to API level %s", a.Name, e.mset.name(), cfg.Name, apiLevel)
+				} else {
+					offlineReason = fmt.Sprintf("decoding error: %v", strictErr)
+					s.Warnf("  Error unmarshalling consumer metafile %q: %v", metafile, strictErr)
+				}
+				singleServerMode := !s.JetStreamIsClustered() && s.standAloneMode()
+				if singleServerMode {
+					if !e.mset.closed.Load() {
+						s.Warnf("  Stopping unsupported stream '%s > %s'", a.Name, e.mset.name())
+						e.mset.mu.Lock()
+						e.mset.offlineReason = "stopped"
+						e.mset.mu.Unlock()
+						e.mset.stop(false, false)
+					}
+
+					// Fake a consumer, so we can respond to API requests as single-server.
+					o := &consumer{
+						mset:          e.mset,
+						js:            s.getJetStream(),
+						acc:           a,
+						srv:           s,
+						cfg:           cfg.ConsumerConfig,
+						active:        false,
+						stream:        e.mset.name(),
+						name:          cfg.Name,
+						dseq:          1,
+						sseq:          1,
+						created:       time.Now().UTC(),
+						closed:        true,
+						offlineReason: offlineReason,
+					}
+					if !cfg.Created.IsZero() {
+						o.created = cfg.Created
+					}
+
+					e.mset.mu.Lock()
+					e.mset.setConsumer(o)
+					e.mset.mu.Unlock()
+				}
 				continue
 			}
+
 			isEphemeral := !isDurableConsumer(&cfg.ConsumerConfig)
 			if isEphemeral {
-				// This is an ephermal consumer and this could fail on restart until
+				// This is an ephemeral consumer and this could fail on restart until
 				// the consumer can reconnect. We will create it as a durable and switch it.
 				cfg.ConsumerConfig.Durable = ofi.Name()
 			}