Sending data from one goroutine to multiple other goroutines

In a project the program receives data via websocket. This data needs to be processed by n algorithms. The amount of algorithms can change dynamically.

My attempt is to create some pub/sub pattern where subscriptions can be started and canceled on the fly. Turns out that this is a bit more challenging than expected.

Here's what I came up with (which is based on https://eli.thegreenplace.net/2020/pubsub-using-channels-in-go/):

package pubsub

import (
    "context"
    "sync"
    "time"
)

type Pubsub struct {
    sync.RWMutex
    subs   []*Subsciption
    closed bool
}

func New() *Pubsub {
    ps := &Pubsub{}
    ps.subs = []*Subsciption{}
    return ps
}

func (ps *Pubsub) Publish(msg interface{}) {
    ps.RLock()
    defer ps.RUnlock()

    if ps.closed {
        return
    }

    for _, sub := range ps.subs {
        // ISSUE1: These goroutines apparently do not exit properly... 
        go func(ch chan interface{}) {
            ch <- msg
        }(sub.Data)
    }
}

func (ps *Pubsub) Subscribe() (context.Context, *Subsciption, error) {
    ps.Lock()
    defer ps.Unlock()

    // prep channel
    ctx, cancel := context.WithCancel(context.Background())
    sub := &Subsciption{
        Data:   make(chan interface{}, 1),
        cancel: cancel,
        ps:     ps,
    }

    // prep subsciption
    ps.subs = append(ps.subs, sub)
    return ctx, sub, nil
}

func (ps *Pubsub) unsubscribe(s *Subsciption) bool {
    ps.Lock()
    defer ps.Unlock()

    found := false
    index := 0
    for i, sub := range ps.subs {
        if sub == s {
            index = i
            found = true
        }
    }
    if found {
        s.cancel()
        ps.subs[index] = ps.subs[len(ps.subs)-1]
        ps.subs = ps.subs[:len(ps.subs)-1]

        // ISSUE2: close the channel async with a delay to ensure
        // nothing will be written to the channel anymore
        // via a pending goroutine from Publish()
        go func(ch chan interface{}) {
            time.Sleep(500 * time.Millisecond)
            close(ch)
        }(s.Data)
    }
    return found
}

func (ps *Pubsub) Close() {
    ps.Lock()
    defer ps.Unlock()

    if !ps.closed {
        ps.closed = true
        for _, sub := range ps.subs {
            sub.cancel()

            // ISSUE2: close the channel async with a delay to ensure
            // nothing will be written to the channel anymore
            // via a pending goroutine from Publish()
            go func(ch chan interface{}) {
                time.Sleep(500 * time.Millisecond)
                close(ch)
            }(sub.Data)
        }
    }
}

type Subsciption struct {
    Data   chan interface{}
    cancel func()
    ps     *Pubsub
}

func (s *Subsciption) Unsubscribe() {
    s.ps.unsubscribe(s)
}

As mentioned in the comments there are (at least) two issues with this:

ISSUE1:

After a while of running in implementation of this I get a few errors of this kind:

goroutine 120624 [runnable]:
bm/internal/pubsub.(*Pubsub).Publish.func1(0x8586c0, 0xc00b44e880, 0xc008617740)
    /home/X/Projects/bm/internal/pubsub/pubsub.go:30
created by bookmaker/internal/pubsub.(*Pubsub).Publish
    /home/X/Projects/bm/internal/pubsub/pubsub.go:30 +0xbb

Without really understanding this it appears to me that the goroutines created in Publish() do accumulate/leak. Is this correct and what am I doing wrong here?

ISSUE2:

When I end a subscription via Unsubscribe() it occurs that Publish() tried to write to a closed channel and panics. To mitigate this I have created a goroutine to close the channel with a delay. This feel really off-best-practice but I was not able to find a proper solution to this. What would be a deterministic way to do this?

Heres a little playground for you to test with: https://play.golang.org/p/K-L8vLjt7_9