only set replicas when we have more than 1

.golangci.yml

@@ -68,7 +68,6 @@ linters:
     - goerr113
     - goimports
     - revive
-    - gomnd
     - misspell
     - noctx
     - stylecheck

cmd/run.go

@@ -107,7 +107,7 @@ func init() {
 	cmdRun.PersistentFlags().BoolVarP(&dryrun, "dry-run", "", false, "In dryrun mode, the worker actions the task without installing firmware")
 	cmdRun.PersistentFlags().BoolVarP(&useStatusKV, "use-kv", "", false, "when this is true, flasher writes status to a NATS KV store instead of sending reply messages")
 	cmdRun.PersistentFlags().BoolVarP(&faultInjection, "fault-injection", "", false, "Tasks can include a Fault attribute to allow fault injection for development purposes")
-	cmdRun.PersistentFlags().IntVarP(&replicas, "nats-replicas", "r", 1, "the default number of replicas to use for NATS data")
+	cmdRun.PersistentFlags().IntVarP(&replicas, "replica-count", "r", 3, "the number of replicas to use for NATS data")
 
 	if err := cmdRun.MarkPersistentFlagRequired("store"); err != nil {
 		log.Fatal(err)

internal/worker/liveness.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"go.hollow.sh/toolbox/events"
+	"go.hollow.sh/toolbox/events/pkg/kv"
 	"go.hollow.sh/toolbox/events/registry"
 
 	"github.com/nats-io/nats.go"
@@ -14,6 +15,7 @@ import (
 var (
 	once           sync.Once
 	checkinCadence = 30 * time.Second
+	livenessTTL    = 3 * time.Minute
 )
 
 // This starts a go-routine to peridocally check in with the NATS kv
@@ -26,7 +28,16 @@ func (w *Worker) startWorkerLivenessCheckin(ctx context.Context) {
 			return
 		}
 
-		if err := registry.InitializeActiveControllerRegistry(natsJS); err != nil {
+		opts := []kv.Option{
+			kv.WithTTL(livenessTTL),
+		}
+
+		// any setting of replicas (even 1) chokes NATS in non-clustered mode
+		if w.replicaCount != 1 {
+			opts = append(opts, kv.WithReplicas(w.replicaCount))
+		}
+
+		if err := registry.InitializeRegistryWithOptions(natsJS, opts...); err != nil {
 			w.logger.WithError(err).Error("unable to initialize active worker registry")
 			return
 		}

internal/worker/worker.go

@@ -310,7 +310,11 @@ func (o *Worker) runTaskWithMonitor(ctx context.Context, task *model.Task, asset
 
 func (o *Worker) getStatusPublisher() sm.Publisher {
 	if o.useStatusKV {
-		return NewStatusKVPublisher(o.stream, o.logger, kv.WithReplicas(o.replicaCount))
+		var opts []kv.Option
+		if o.replicaCount > 1 {
+			opts = append(opts, kv.WithReplicas(o.replicaCount))
+		}
+		return NewStatusKVPublisher(o.stream, o.logger, opts...)
 	}
 	return &statusEmitter{o.stream, o.logger}
 }