media: add SignMP4 tracing, fix channel leak (#126)

* tracing: add a lot more instrumentation around SignMP4 tracing

* tracing: track signing duration for external signing too

* segchanman: time out segment send

* concat: okay yeah i see why this was a leak now

* webrtc_playback: actually unsubscribe here too

go.mod

···
 	gitlab.com/gitlab-org/release-cli v0.18.0
 	go.opentelemetry.io/otel v1.35.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0
-	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.11.0
-	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.35.0
-	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.35.0
-	go.opentelemetry.io/otel/log v0.11.0
 	go.opentelemetry.io/otel/sdk v1.35.0
-	go.opentelemetry.io/otel/sdk/log v0.11.0
-	go.opentelemetry.io/otel/sdk/metric v1.35.0
 	go.uber.org/goleak v1.3.0
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0
 	golang.org/x/image v0.22.0
···
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0 // indirect
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect

go.sum

···
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0/go.mod h1:zjPK58DtkqQFn+YUMbx0M2XV3QgKU0gS9LeGohREyK4=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0 h1:m639+BofXTvcY1q8CGs4ItwQarYtJPOWmVobfM1HpVI=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0/go.mod h1:LjReUci/F4BUyv+y4dwnq3h/26iNOeC3wAIqgvTIZVo=
-go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.11.0 h1:k6KdfZk72tVW/QVZf60xlDziDvYAePj5QHwoQvrB2m8=
-go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.11.0/go.mod h1:5Y3ZJLqzi/x/kYtrSrPSx7TFI/SGsL7q2kME027tH6I=
-go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.35.0 h1:PB3Zrjs1sG1GBX51SXyTSoOTqcDglmsk7nT6tkKPb/k=
-go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.35.0/go.mod h1:U2R3XyVPzn0WX7wOIypPuptulsMcPDPs/oiSVOMVnHY=
-go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.35.0 h1:T0Ec2E+3YZf5bgTNQVet8iTDW7oIk03tXHq+wkwIDnE=
-go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.35.0/go.mod h1:30v2gqH+vYGJsesLWFov8u47EpYTcIQcBjKpI6pJThg=
-go.opentelemetry.io/otel/log v0.11.0 h1:c24Hrlk5WJ8JWcwbQxdBqxZdOK7PcP/LFtOtwpDTe3Y=
-go.opentelemetry.io/otel/log v0.11.0/go.mod h1:U/sxQ83FPmT29trrifhQg+Zj2lo1/IPN1PF6RTFqdwc=
 go.opentelemetry.io/otel/metric v1.35.0 h1:0znxYu2SNyuMSQT4Y9WDWej0VpcsxkuklLa4/siN90M=
 go.opentelemetry.io/otel/metric v1.35.0/go.mod h1:nKVFgxBZ2fReX6IlyW28MgZojkoAkJGaE8CpgeAU3oE=
 go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY=
 go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg=
-go.opentelemetry.io/otel/sdk/log v0.11.0 h1:7bAOpjpGglWhdEzP8z0VXc4jObOiDEwr3IYbhBnjk2c=
-go.opentelemetry.io/otel/sdk/log v0.11.0/go.mod h1:dndLTxZbwBstZoqsJB3kGsRPkpAgaJrWfQg3lhlHFFY=
 go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o=
 go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w=
 go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt/xgMs=

pkg/media/concat.go

···
 	// them in a pipe so that we don't miss any in between iterations of the output
 	allFiles := make(chan []byte, 1024)
 	go func() {
+		ch := streamer.SubscribeSegment(ctx, user, rendition)
+		defer streamer.UnsubscribeSegment(ctx, user, rendition, ch)
 		for {
-			ch := streamer.SubscribeSegment(ctx, user, rendition)
 			select {
 			case <-ctx.Done():
 				log.Debug(ctx, "exiting segment reader")
-				streamer.UnsubscribeSegment(ctx, user, rendition, ch)
 				return
 			case file := <-ch:
 				log.Debug(ctx, "got segment", "file", file.Filepath)

pkg/media/media_signer.go

···
 	"fmt"
 	"io"
 	"path/filepath"
+	"time"
 
 	"git.stream.place/streamplace/c2pa-go/pkg/c2pa"
 	"go.opentelemetry.io/otel"
···
 	"stream.place/streamplace/pkg/crypto/aqpub"
 	"stream.place/streamplace/pkg/crypto/signers"
 	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/spmetrics"
 )
 
 type MediaSigner interface {
 	SignMP4(ctx context.Context, input io.ReadSeeker, start int64) ([]byte, error)
 	Pub() aqpub.Pub
+	Streamer() string
 }
 
 type MediaSignerLocal struct {
···
 	}, nil
 }
 
+func (ms *MediaSignerLocal) Streamer() string {
+	return ms.StreamerName
+}
+
 func (ms *MediaSignerLocal) SignMP4(ctx context.Context, input io.ReadSeeker, start int64) ([]byte, error) {
+	startTime := time.Now()
 	ctx, span := otel.Tracer("signer").Start(ctx, "SignMP4")
 	defer span.End()
 	title := "livestream"
···
 			},
 		},
 	}
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_MarshalManifest")
 	manifestBs, err := json.Marshal(mani)
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal manifest: %w", err)
···
 	if err != nil {
 		return nil, fmt.Errorf("failed to unmarshal manifest: %w", err)
 	}
+	span.End()
+
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_GetSigningAlgorithm")
 	alg, err := c2pa.GetSigningAlgorithm(string(c2pa.ES256K))
 	if err != nil {
 		return nil, fmt.Errorf("failed to get signing algorithm: %w", err)
 	}
+	span.End()
+
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_NewBuilder")
 	b, err := c2pa.NewBuilder(&manifest, &c2pa.BuilderParams{
 		Cert:      ms.Cert,
 		Signer:    ms.Signer,
···
 	if err != nil {
 		return nil, fmt.Errorf("failed to create C2PA builder: %w", err)
 	}
+	span.End()
 
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_Sign")
 	output := &aqio.ReadWriteSeeker{}
 	err = b.Sign(input, output, "video/mp4")
 	if err != nil {
 		return nil, fmt.Errorf("failed to sign MP4: %w", err)
 	}
+	span.End()
+
+	ctx, span = otel.Tracer("signer").Start(ctx, "SignMP4_OutputBytes")
 	bs, err := output.Bytes()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get output bytes: %w", err)
 	}
+	span.End()
+	spmetrics.SigningDuration.WithLabelValues(ms.StreamerName).Observe(float64(time.Since(startTime).Milliseconds()))
 	return bs, nil
 }
 

pkg/media/media_signer_ext.go

···
 	"io"
 	"os"
 	"os/exec"
+	"time"
 
 	"github.com/decred/dcrd/dcrec/secp256k1"
 	"github.com/mr-tron/base58"
+	"go.opentelemetry.io/otel"
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/crypto/aqpub"
+	"stream.place/streamplace/pkg/spmetrics"
 )
 
 type MediaSignerExt struct {
···
 }
 
 func (ms *MediaSignerExt) SignMP4(ctx context.Context, input io.ReadSeeker, start int64) ([]byte, error) {
+	startTime := time.Now()
+	ctx, span := otel.Tracer("signer").Start(ctx, "SignMP4_Ext")
+	defer span.End()
 	// Get the path to the current executable
 	execPath, err := os.Executable()
 	if err != nil {
···
 	if err := cmd.Wait(); err != nil {
 		return nil, fmt.Errorf("command failed: %w, stderr: %s", err, stderr.String())
 	}
-
+	spmetrics.SigningDuration.WithLabelValues(ms.streamer).Observe(float64(time.Since(startTime).Milliseconds()))
 	return stdout.Bytes(), nil
 }
 
 func (ms *MediaSignerExt) Pub() aqpub.Pub {
 	return ms.pub
 }
+
+func (ms *MediaSignerExt) Streamer() string {
+	return ms.streamer
+}

pkg/media/segchanman/segchanman.go

···
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"go.opentelemetry.io/otel"
+	"stream.place/streamplace/pkg/log"
 )
 
 // it's a segment channel manager, you see
···
 		chs = []chan *Seg{}
 		s.segChans[key] = chs
 	}
-	ch := make(chan *Seg, 1024)
+	ch := make(chan *Seg)
 	chs = append(chs, ch)
 	s.segChans[key] = chs
 	return ch
···
 	}
 	for _, ch := range chs {
 		go func(ch chan *Seg) {
-			ch <- seg
+			select {
+			case ch <- seg:
+			case <-ctx.Done():
+				return
+			case <-time.After(1 * time.Minute):
+				log.Warn(ctx, "failed to send segment to channel, timing out", "user", user, "rendition", rendition)
+			}
+
 		}(ch)
 	}
 }

pkg/media/segmenter.go

···
 	"github.com/go-gst/go-gst/gst"
 	"github.com/go-gst/go-gst/gst/app"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"stream.place/streamplace/pkg/log"
 )
 
···
 		appsink.SetCallbacks(&app.SinkCallbacks{
 			NewSampleFunc: WriterNewSample(ctx, buf),
 			EOSFunc: func(sink *app.Sink) {
-				ctx, span := otel.Tracer("signer").Start(ctx, "SegmentAndSignElem")
+				ctx, span := otel.Tracer("signer").Start(ctx, "SegmentAndSignElem", trace.WithAttributes(
+					attribute.String("streamer", ms.Streamer()),
+				))
 				defer span.End()
 				resetTimer <- struct{}{}
 				now := time.Now().UnixMilli()

pkg/media/webrtc_playback.go

···
 
 	segBuffer := make(chan *segchanman.Seg, 1024)
 	go func() {
+		ch := mm.SubscribeSegment(ctx, user, rendition)
+		defer mm.UnsubscribeSegment(ctx, user, rendition, ch)
 		for {
-			ch := mm.SubscribeSegment(ctx, user, rendition)
 			select {
 			case <-ctx.Done():
 				log.Debug(ctx, "exiting segment reader")
-				mm.UnsubscribeSegment(ctx, user, rendition, ch)
 				return
 			case file := <-ch:
 				log.Debug(ctx, "got segment", "file", file.Filepath)

pkg/spmetrics/spmetrics.go

···
 	Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000},
 }, []string{"streamer"})
 
+var SigningDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Name:    "streamplace_signing_duration_ms",
+	Help:    "duration of transcode in ms",
+	Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000, 20000, 30000, 60000},
+}, []string{"streamer"})
+
 var QueuedTranscodeDuration = promauto.NewGaugeVec(prometheus.GaugeOpts{
 	Name: "streamplace_queued_transcode_duration_ms",
 	Help: "duration of transcode in ms, including time spent waiting",