fix: reject unrouted messages without seq
This commit is contained in:
@@ -138,6 +138,17 @@ func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenant
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
tables := groupMessageShardTablesForSeq(tenantID, routes, seq)
|
||||||
|
if len(tables) == 0 && cached {
|
||||||
|
if routes, err = groupMessageAllShardRoutes(q, ctx, tenantID); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
tables = groupMessageShardTablesForSeq(tenantID, routes, seq)
|
||||||
|
}
|
||||||
|
return tables, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func groupMessageShardTablesForSeq(tenantID t.TenantID, routes []store.GroupMessageShardRoute, seq int64) []string {
|
||||||
tables := make([]string, 0, 2)
|
tables := make([]string, 0, 2)
|
||||||
for _, route := range routes {
|
for _, route := range routes {
|
||||||
if route.MinSeq <= seq && route.MaxSeq >= seq {
|
if route.MinSeq <= seq && route.MaxSeq >= seq {
|
||||||
@@ -146,7 +157,7 @@ func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenant
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return tables, nil
|
return tables
|
||||||
}
|
}
|
||||||
|
|
||||||
func groupMessageAllShardRoutes(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID) ([]store.GroupMessageShardRoute, error) {
|
func groupMessageAllShardRoutes(q sqlx.QueryerContext, ctx context.Context, tenantID t.TenantID) ([]store.GroupMessageShardRoute, error) {
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/tinode/chat/server/store"
|
||||||
types "github.com/tinode/chat/server/store/types"
|
types "github.com/tinode/chat/server/store/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -75,6 +76,21 @@ func TestGroupMessageShardTableRejectsUnsafeMonth(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestGroupMessageShardTablesForSeqFiltersRoutes(t *testing.T) {
|
||||||
|
routes := []store.GroupMessageShardRoute{
|
||||||
|
{Month: "202608", MinSeq: 1, MaxSeq: 20},
|
||||||
|
{Month: "202609", MinSeq: 21, MaxSeq: 100},
|
||||||
|
{Month: "202613", MinSeq: 1, MaxSeq: 1000},
|
||||||
|
}
|
||||||
|
tables := groupMessageShardTablesForSeq(7, routes, 23)
|
||||||
|
if len(tables) != 1 || tables[0] != "im_007_group_messages_202609" {
|
||||||
|
t.Fatalf("unexpected shard tables for seq: %#v", tables)
|
||||||
|
}
|
||||||
|
if tables = groupMessageShardTablesForSeq(7, routes, 101); len(tables) != 0 {
|
||||||
|
t.Fatalf("unexpected shard tables for uncovered seq: %#v", tables)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestTenantPartitionTableNames(t *testing.T) {
|
func TestTenantPartitionTableNames(t *testing.T) {
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
tenantID types.TenantID
|
tenantID types.TenantID
|
||||||
|
|||||||
+6
-3
@@ -255,14 +255,17 @@ func (h *Hub) run() {
|
|||||||
}
|
}
|
||||||
} else if msg.Note == nil {
|
} else if msg.Note == nil {
|
||||||
// Topic is unknown or offline.
|
// Topic is unknown or offline.
|
||||||
// Note is silently ignored, all other messages are reported as accepted to prevent
|
// Note is silently ignored. Other client messages cannot be accepted here
|
||||||
// clients from guessing valid topic names.
|
// because they are not routed to a live topic and therefore cannot reserve a
|
||||||
|
// message seq or be persisted.
|
||||||
|
|
||||||
// TODO(gene): validate topic name, discarding invalid topics.
|
// TODO(gene): validate topic name, discarding invalid topics.
|
||||||
|
|
||||||
logs.Info.Printf("Hub. Topic[%s] is unknown or offline", msg.RcptTo)
|
logs.Info.Printf("Hub. Topic[%s] is unknown or offline", msg.RcptTo)
|
||||||
|
|
||||||
msg.sess.queueOut(NoErrAcceptedExplicitTs(msg.Id, msg.RcptTo, types.TimeNow(), msg.Timestamp))
|
if msg.sess != nil {
|
||||||
|
msg.sess.queueOut(ErrAttachFirst(msg, types.TimeNow()))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
case msg := <-h.routeSrv:
|
case msg := <-h.routeSrv:
|
||||||
// This is a server message from a connection not subscribed to topic
|
// This is a server message from a connection not subscribed to topic
|
||||||
|
|||||||
@@ -1,8 +1,10 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"net/http"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/tinode/chat/server/store/types"
|
"github.com/tinode/chat/server/store/types"
|
||||||
)
|
)
|
||||||
@@ -53,3 +55,50 @@ func TestTopicAssignsTenantToServerMessage(t *testing.T) {
|
|||||||
t.Fatalf("message tenant = %d, want %d", msg.TenantID, topic.tenantID)
|
t.Fatalf("message tenant = %d, want %d", msg.TenantID, topic.tenantID)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestHubRejectsUnroutedClientMessageInsteadOfAcceptingWithoutSeq(t *testing.T) {
|
||||||
|
h := &Hub{
|
||||||
|
topics: &sync.Map{},
|
||||||
|
routeCli: make(chan *ClientComMessage, 1),
|
||||||
|
shutdown: make(chan chan<- bool),
|
||||||
|
routeSrv: make(chan *ServerComMessage),
|
||||||
|
join: make(chan *ClientComMessage),
|
||||||
|
meta: make(chan *ClientComMessage),
|
||||||
|
unreg: make(chan *topicUnreg),
|
||||||
|
userStatus: make(chan *userStatusReq),
|
||||||
|
rehash: make(chan bool),
|
||||||
|
}
|
||||||
|
go h.run()
|
||||||
|
done := make(chan bool)
|
||||||
|
defer func() {
|
||||||
|
h.shutdown <- done
|
||||||
|
<-done
|
||||||
|
}()
|
||||||
|
|
||||||
|
sess := &Session{tenantID: 7, send: make(chan any, 1)}
|
||||||
|
now := time.Now().UTC()
|
||||||
|
h.routeCli <- &ClientComMessage{
|
||||||
|
Id: "pub-1",
|
||||||
|
TenantID: 7,
|
||||||
|
RcptTo: "p2pMissing",
|
||||||
|
Original: "p2pMissing",
|
||||||
|
Timestamp: now,
|
||||||
|
sess: sess,
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case raw := <-sess.send:
|
||||||
|
msg, ok := raw.(*ServerComMessage)
|
||||||
|
if !ok || msg.Ctrl == nil {
|
||||||
|
t.Fatalf("unexpected response: %#v", raw)
|
||||||
|
}
|
||||||
|
if msg.Ctrl.Code != http.StatusConflict || msg.Ctrl.Text != "must attach first" {
|
||||||
|
t.Fatalf("unexpected ctrl: %#v", msg.Ctrl)
|
||||||
|
}
|
||||||
|
if msg.Ctrl.Params != nil {
|
||||||
|
t.Fatalf("unrouted message must not expose seq params: %#v", msg.Ctrl.Params)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("timeout waiting for unrouted message response")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user