Add ability to propagate certain sink errors up to the source

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/spf13/cobra v1.8.1
 	github.com/tcolgate/mp3 v0.0.0-20170426193717-e79c5a46d300
 	golang.org/x/crypto v0.21.0
+	golang.org/x/sync v0.5.0
 	golang.org/x/term v0.18.0
 	google.golang.org/protobuf v1.33.0
 	gopkg.in/yaml.v3 v3.0.1
@@ -55,7 +56,6 @@ require (
 	golang.org/x/exp/shiny v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/image v0.14.0 // indirect
 	golang.org/x/mobile v0.0.0-20231127183840-76ac6878050a // indirect
-	golang.org/x/sync v0.5.0 // indirect
 	golang.org/x/sys v0.20.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 )

pkg/gordio/server/ingest.go

@@ -6,6 +6,6 @@ import (
 	"dynatron.me/x/stillbox/pkg/calls"
 )
 
-func (s *Server) Ingest(ctx context.Context, call *calls.Call) {
-	s.sinks.EmitCall(context.Background(), call)
+func (s *Server) Ingest(ctx context.Context, call *calls.Call) error {
+	return s.sinks.EmitCall(context.Background(), call)
 }

pkg/gordio/server/server.go

@@ -41,8 +41,8 @@ func New(cfg *config.Config) (*Server, error) {
 		nex:  nexus.New(),
 	}
 
-	srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db))
-	srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex))
+	srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db), true)
+	srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false)
 	srv.sources.Register("rdio-http", sources.NewRdioHTTP(authenticator, srv))
 
 	r.Use(middleware.RequestID)

pkg/gordio/sinks/sinks.go

@@ -2,6 +2,7 @@ package sinks
 
 import (
 	"context"
+	"golang.org/x/sync/errgroup"
 
 	"dynatron.me/x/stillbox/pkg/calls"
 
@@ -16,26 +17,41 @@ type Sink interface {
 type sinkInstance struct {
 	Sink
 	Name string
+
+	// whether call ingest should be considered failed if this sink returns error
+	Required bool
 }
 
 type Sinks []sinkInstance
 
-func (s *Sinks) Register(name string, toAdd Sink) {
+func (s *Sinks) Register(name string, toAdd Sink, required bool) {
 	*s = append(*s, sinkInstance{
 		Name:     name,
 		Sink:     toAdd,
+		Required: required,
 	})
 }
 
-func (s *Sinks) EmitCall(ctx context.Context, call *calls.Call) {
+func (s *Sinks) EmitCall(ctx context.Context, call *calls.Call) error {
+	g, ctx := errgroup.WithContext(ctx)
 	for i := range *s {
-		go (*s)[i].emitCallLogErr(ctx, call)
+		sink := (*s)[i]
+		g.Go(sink.callEmitter(ctx, call))
 	}
+
+	return g.Wait()
 }
 
-func (sink *sinkInstance) emitCallLogErr(ctx context.Context, call *calls.Call) {
+func (sink *sinkInstance) callEmitter(ctx context.Context, call *calls.Call) func() error {
+	return func() error {
 		err := sink.Call(ctx, call)
 		if err != nil {
 			log.Error().Str("sink", sink.Name).Err(err).Msg("call emit to sink failed")
+			if sink.Required {
+				return err
+			}
+		}
+
+		return nil
 	}
 }

pkg/gordio/sources/http.go

@@ -127,7 +127,12 @@ func (h *RdioHTTP) routeCallUpload(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
-	h.ing.Ingest(ctx, call)
+	err = h.ing.Ingest(ctx, call)
+	if err != nil {
+		log.Error().Err(err).Msg("ingest failed")
+		http.Error(w, "Call ingest failed.", http.StatusInternalServerError)
+		return
+	}
 
 	log.Info().Int("system", cur.System).Int("tgid", cur.Talkgroup).Msg("ingested")
 

pkg/gordio/sources/source.go

@@ -35,7 +35,7 @@ func (s *Sources) PublicRoutes(r chi.Router) {
 }
 
 type Ingestor interface {
-	Ingest(context.Context, *calls.Call)
+	Ingest(context.Context, *calls.Call) error
 }
 
 type PublicRouteSource interface {