Panic when use Next()

[lekhanhtoan37]

panic: runtime error: index out of range [0] with length 0

goroutine 2185349 [running]:
github.com/smallnest/weighted.(*SW).nextWeighted(0x710300?)
        /go/pkg/mod/github.com/smallnest/weighted@v0.0.0-20230419055410-36b780e40a7a/smooth_weighted.go:82 +0x4b
github.com/smallnest/weighted.(*SW).Next(...)
        /go/pkg/mod/github.com/smallnest/weighted@v0.0.0-20230419055410-36b780e40a7a/smooth_weighted.go:69

[smallnest]

it looks you has not added servers

[lekhanhtoan37]

package proxy

import (
	"context"
	"fmt"
	"net/http"
	"net/url"
	"sync"
	"sync/atomic"

	"github.com/smallnest/weighted"
)

// Topic ->
type WeightedUpstream struct {
	WeightedUpstreamPerChain map[string]*WeightedUpstreamPerChain
	UpstreamConfig           map[string]UpstreamPerChain
	mutex                    sync.Mutex
}

// WeightedUpstream chứa URL và weight cho mỗi upstream
type WeightedUpstreamPerChain struct {
	WeightedUpstreamPerTopic map[string]*WeightedUpstreamPerTopic
}

type WeightedUpstreamPerTopic struct {
	FallbackURL   *url.URL `toml:"fallback_url"`
	Pool          *weighted.SW
	UpstreamCount atomic.Uint64
}

func (w *WeightedUpstreamPerChain) Clone() *WeightedUpstreamPerChain {
	clone := &WeightedUpstreamPerChain{
		WeightedUpstreamPerTopic: make(map[string]*WeightedUpstreamPerTopic),
	}

	for topic, upstream := range w.WeightedUpstreamPerTopic {
		count := upstream.UpstreamCount.Load()
		clone.WeightedUpstreamPerTopic[topic] = &WeightedUpstreamPerTopic{
			FallbackURL:   upstream.FallbackURL,
			Pool:          upstream.Pool,
			UpstreamCount: atomic.Uint64{},
		}
		clone.WeightedUpstreamPerTopic[topic].UpstreamCount.Store(count)
	}

	return clone
}

func (w *WeightedUpstream) CloneWeightedUpstream() *WeightedUpstream {
	clone := &WeightedUpstream{
		WeightedUpstreamPerChain: make(map[string]*WeightedUpstreamPerChain),
	}

	for chain, upstream := range w.WeightedUpstreamPerChain {
		clone.WeightedUpstreamPerChain[chain] = upstream.Clone()
	}

	return clone
}

func (w *WeightedUpstream) CloneUpstreamConfig() *UpstreamConfig {
	c := make(map[string]UpstreamPerChain, len(w.UpstreamConfig))
	for chain, upstream := range w.UpstreamConfig {
		c[chain] = upstream
	}
	return &UpstreamConfig{
		Upstream: c,
	}
}

func (w *WeightedUpstream) Clone() *WeightedUpstream {
	c := w.CloneWeightedUpstream()
	c.UpstreamConfig = make(map[string]UpstreamPerChain)

	for chain, upstream := range w.UpstreamConfig {
		c.UpstreamConfig[chain] = upstream
	}
	return c
}

func NewWeightedUpstream(upstreamConfig map[string]UpstreamPerChain) *WeightedUpstream {
	weightedUpstream := WeightedUpstream{
		WeightedUpstreamPerChain: make(map[string]*WeightedUpstreamPerChain),
		UpstreamConfig:           upstreamConfig,
	}

	for chain, upstream := range upstreamConfig {
		if _, ok := weightedUpstream.WeightedUpstreamPerChain[chain]; !ok {
			weightedUpstream.WeightedUpstreamPerChain[chain] = &WeightedUpstreamPerChain{
				WeightedUpstreamPerTopic: make(map[string]*WeightedUpstreamPerTopic),
			}
		}

		for topic, rawUrl := range upstream.WebSocketURL {
			fallbackUrl, err := url.Parse(rawUrl[0])
			if err != nil {
				Fatal(
					context.Background(),
					err,
					"invalid fallback url",
					LogField{Key: "url", Value: rawUrl},
				)
			}

			subTypes := upstream.SubscriptionType[topic]
			for _, subType := range subTypes {
				pool := &weighted.SW{}
				topicWithSubType := fmt.Sprintf("%s_%s", chain, subType)
				if _, ok := weightedUpstream.WeightedUpstreamPerChain[chain].WeightedUpstreamPerTopic[topicWithSubType]; !ok {
					weightedUpstream.WeightedUpstreamPerChain[chain].WeightedUpstreamPerTopic[topicWithSubType] = &WeightedUpstreamPerTopic{
						FallbackURL:   fallbackUrl,
						Pool:          pool,
						UpstreamCount: atomic.Uint64{},
					}
				}

			}
		}
	}

	return &weightedUpstream
}

func (w *WeightedUpstream) CopyCloneURLRequest(chain string, req *http.Request) *WeightedUpstream {
	copyUrl := func(u *url.URL) *url.URL {
		u.Fragment = req.URL.Fragment
		// x-api-key
		u.RawQuery = req.URL.RawQuery
		return u
	}

	originWeightedUpstreamPerChain, ok := w.WeightedUpstreamPerChain[chain]
	if !ok {
		Fatal(
			context.Background(),
			fmt.Errorf("chain not found"),
			"chain not found",
			LogField{Key: "chain", Value: chain},
		)
	}

	urlWithWeight := []struct {
		Url    *url.URL
		Weight int
	}{}

	weightedUpstreamPerChain := originWeightedUpstreamPerChain.Clone()

	for _, weightedUpstreamPerTopic := range weightedUpstreamPerChain.WeightedUpstreamPerTopic {
		allUrls := weightedUpstreamPerTopic.Pool.All()
		for key, weight := range allUrls {
			u, ok := key.(*url.URL)
			if !ok {
				Fatal(
					context.Background(),
					fmt.Errorf("invalid url type"),
					"invalid url",
					LogField{Key: "url", Value: key},
				)
			}

			urlWithWeight = append(urlWithWeight, struct {
				Url    *url.URL
				Weight int
			}{
				Url:    copyUrl(u),
				Weight: weight,
			})
		}

		weightedUpstreamPerTopic.Pool.RemoveAll()

		for i, urlWithWeight := range urlWithWeight {
			weightedUpstreamPerTopic.Pool.Add(urlWithWeight.Url, urlWithWeight.Weight)
			if i == 0 {
				weightedUpstreamPerTopic.FallbackURL = urlWithWeight.Url
			}
		}

		urlWithWeight = []struct {
			Url    *url.URL
			Weight int
		}{}
	}

	nw := w.CloneWeightedUpstream()
	for c := range nw.WeightedUpstreamPerChain {
		if c == chain {
			nw.WeightedUpstreamPerChain[c] = weightedUpstreamPerChain
			break
		}
	}

	return &WeightedUpstream{
		WeightedUpstreamPerChain: nw.WeightedUpstreamPerChain,
		UpstreamConfig:           w.UpstreamConfig,
	}
}

func (w *WeightedUpstream) GetUpstream(chain string, topic string) (u *url.URL) {
	w.mutex.Lock()
	defer w.mutex.Unlock()
	weightedUpstreamPerChain, ok := w.WeightedUpstreamPerChain[chain]
	if !ok {
		return nil
	}

	weightedUpstreamPerTopic, ok := weightedUpstreamPerChain.WeightedUpstreamPerTopic[topic]
	if !ok {
		return nil
	}

	defer func() {
		if r := recover(); r != nil {
			Warn(
				context.Background(),
				"panic in GetUpstream",
				LogField{Key: "chain", Value: chain},
				LogField{Key: "topic", Value: topic},
				LogField{Key: "error", Value: r},
			)
			u = weightedUpstreamPerTopic.FallbackURL
		}
	}()

	weightedUpstream := weightedUpstreamPerTopic.Pool.Next()
	if weightedUpstream != nil {
		if u, ok := weightedUpstream.(*url.URL); ok {
			return u
		}
	}

	u = weightedUpstreamPerTopic.FallbackURL
	return u
}

func (w *WeightedUpstream) AddUpstreamUrl(
	chain string,
	topic string,
	url *url.URL,
	weight int,
) error {
	weightedUpstreamPerChain, ok := w.WeightedUpstreamPerChain[chain]
	if !ok {
		return fmt.Errorf("chain %s not found", chain)
	}

	weightedUpstreamPerTopic, ok := weightedUpstreamPerChain.WeightedUpstreamPerTopic[topic]
	if !ok {
		return fmt.Errorf("topic %s not found", topic)
	}

	w.mutex.Lock()
	defer w.mutex.Unlock()
	weightedUpstreamPerTopic.Pool.Add(url, weight)
	weightedUpstreamPerTopic.UpstreamCount.Add(1)
	return nil
}

func (w *WeightedUpstream) Len() int {
	count := 0
	for _, weightedUpstreamPerChain := range w.WeightedUpstreamPerChain {
		for _, weightedUpstreamPerTopic := range weightedUpstreamPerChain.WeightedUpstreamPerTopic {
			count += int(weightedUpstreamPerTopic.UpstreamCount.Load())
		}
	}
	return count
}

func (w *WeightedUpstream) AddUpstream(
	chain string,
	topicWithSubType string,
	rawUrl string,
	weight int,
) error {
	weightedUpstreamPerChain, ok := w.WeightedUpstreamPerChain[chain]
	if !ok {
		return fmt.Errorf("chain %s not found", chain)
	}

	weightedUpstreamPerTopic, ok := weightedUpstreamPerChain.WeightedUpstreamPerTopic[topicWithSubType]
	if !ok {
		return fmt.Errorf("topic %s not found", topicWithSubType)
	}

	u, err := url.Parse(rawUrl)
	if err != nil {
		return fmt.Errorf("invalid url %s: %w", rawUrl, err)
	}

	weightedUpstreamPerTopic.Pool.Add(u, weight)
	return nil
}

I use CopyCloneURLRequest per served request, and create default with config. It's randomly panic, not at the first time, sometimes it took very long time then panic
GetUpstream was called in goroutine (in ServeHTTP function)

P/s: I did a trick that recover by fallback URL to bypass temporarily panic issue