fix(localstorage): sinks may need a context to start

humanlogio · Oct 28, 2024 · 01663db · 01663db
1 parent 739039b
commit 01663db
Show file tree

Hide file tree

Showing 6 changed files with 6 additions and 6 deletions.
diff --git a/cmd/humanlog/localhost.go b/cmd/humanlog/localhost.go
@@ -90,7 +90,7 @@ func startLocalhostServer(
 		}, nil
 	}
 	storage := localstorage.NewMemStorage(ll.WithGroup("memstorage"))
-	ownSink, _, err := storage.SinkFor(int64(machineID), time.Now().UnixNano())
+	ownSink, _, err := storage.SinkFor(ctx, int64(machineID), time.Now().UnixNano())
 	if err != nil {
 		return nil, nil, fmt.Errorf("can't create own sink: %v", err)
 	}

diff --git a/internal/localstorage/memory.go b/internal/localstorage/memory.go
@@ -100,7 +100,7 @@ func (str *MemStorage) Heartbeat(ctx context.Context, machineID, sessionID int64
 	return str.heartbeat, nil
 }
 
-func (str *MemStorage) SinkFor(machineID, sessionID int64) (sink.Sink, time.Duration, error) {
+func (str *MemStorage) SinkFor(ctx context.Context, machineID, sessionID int64) (sink.Sink, time.Duration, error) {
 	str.sinksMu.Lock()
 	defer str.sinksMu.Unlock()
 

diff --git a/internal/localstorage/memory_test.go b/internal/localstorage/memory_test.go
@@ -186,7 +186,7 @@ func TestMemoryStorage(t *testing.T) {
 			mem := NewMemStorage(slog.New(slog.NewTextHandler(os.Stderr, nil)))
 
 			for _, leg := range tt.input {
-				snk, _, err := mem.SinkFor(leg.MachineId, leg.SessionId)
+				snk, _, err := mem.SinkFor(ctx, leg.MachineId, leg.SessionId)
 				require.NoError(t, err)
 				for _, ev := range leg.Logs {
 					err = snk.Receive(ctx, ev)

diff --git a/internal/localstorage/queryable.go b/internal/localstorage/queryable.go
@@ -10,7 +10,7 @@ import (
 
 type Storage interface {
 	Queryable
-	SinkFor(machineID, sessionID int64) (_ sink.Sink, heartbeatIn time.Duration, _ error)
+	SinkFor(ctx context.Context, machineID, sessionID int64) (_ sink.Sink, heartbeatIn time.Duration, _ error)
 	Heartbeat(ctx context.Context, machineID, sessionID int64) (time.Duration, error)
 }
 

diff --git a/internal/localsvc/svc.go b/internal/localsvc/svc.go
@@ -102,7 +102,7 @@ func (svc *Service) IngestStream(ctx context.Context, req *connect.ClientStream[
 		slog.Int64("session_id", sessionID),
 	)
 	ll.DebugContext(ctx, "receiving data from stream")
-	snk, heartbeatIn, err := svc.storage.SinkFor(machineID, sessionID)
+	snk, heartbeatIn, err := svc.storage.SinkFor(ctx, machineID, sessionID)
 	if err != nil {
 		ll.ErrorContext(ctx, "obtaining sink for stream", slog.Any("err", err))
 		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("obtaining sink for stream: %v", err))

diff --git a/internal/localsvc/svc_test.go b/internal/localsvc/svc_test.go
@@ -123,7 +123,7 @@ func TestSummarize(t *testing.T) {
 			mem := localstorage.NewMemStorage(ll)
 
 			for _, leg := range tt.input {
-				snk, _, err := mem.SinkFor(leg.MachineId, leg.SessionId)
+				snk, _, err := mem.SinkFor(ctx, leg.MachineId, leg.SessionId)
 				require.NoError(t, err)
 				for _, ev := range leg.Logs {
 					err = snk.Receive(ctx, ev)