test(iot): безопасность enqueue↔closeAndWait под race (фикс 🔴3)
Параллельный enqueue + закрытие канала не должен паниковать «send on closed channel». go test -race ./... зелёный.
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
// sender_test.go — тест безопасности enqueue ↔ closeAndWait (фикс 🔴3).
|
||||
//
|
||||
// Раньше поздний колбэк paho мог упасть «send on closed channel» при закрытии
|
||||
// канала во время enqueue. Мьютекс + closed-флаг должны это исключать.
|
||||
package bridge
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestDispatcherEnqueueCloseNoPanic(t *testing.T) {
|
||||
log := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
// client nil — enqueue/closeAndWait его не трогают (только worker'ы, которых не запускаем).
|
||||
d := newSQSDispatcher(nil, "http://queue", log)
|
||||
|
||||
// Параллельный enqueue, пока основной поток закрывает канал.
|
||||
var wg sync.WaitGroup
|
||||
stop := make(chan struct{})
|
||||
for i := 0; i < 4; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
d.enqueue(TelemetryEnvelope{Namespace: "ns", DeviceID: "d"})
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
d.closeAndWait()
|
||||
close(stop)
|
||||
wg.Wait()
|
||||
|
||||
// После закрытия enqueue обязан вернуть false (канал закрыт), а не паниковать.
|
||||
if d.enqueue(TelemetryEnvelope{Namespace: "ns", DeviceID: "d"}) {
|
||||
t.Error("enqueue после close должен вернуть false")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user