diff --git a/server/db/mysql/group_message_shards.go b/server/db/mysql/group_message_shards.go index 13f8ca0..a472f3b 100644 --- a/server/db/mysql/group_message_shards.go +++ b/server/db/mysql/group_message_shards.go @@ -138,6 +138,17 @@ func groupMessageTablesForSeq(q sqlx.QueryerContext, ctx context.Context, tenant 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) for _, route := range routes { 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) { diff --git a/server/db/mysql/group_schema_test.go b/server/db/mysql/group_schema_test.go index 01b324a..76f6dca 100644 --- a/server/db/mysql/group_schema_test.go +++ b/server/db/mysql/group_schema_test.go @@ -7,6 +7,7 @@ import ( "strings" "testing" + "github.com/tinode/chat/server/store" 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) { tests := []struct { tenantID types.TenantID diff --git a/server/hub.go b/server/hub.go index f1aa441..f18615c 100644 --- a/server/hub.go +++ b/server/hub.go @@ -255,14 +255,17 @@ func (h *Hub) run() { } } else if msg.Note == nil { // Topic is unknown or offline. - // Note is silently ignored, all other messages are reported as accepted to prevent - // clients from guessing valid topic names. + // Note is silently ignored. Other client messages cannot be accepted here + // 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. 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: // This is a server message from a connection not subscribed to topic diff --git a/server/hub_tenant_test.go b/server/hub_tenant_test.go index df23e7b..01c38aa 100644 --- a/server/hub_tenant_test.go +++ b/server/hub_tenant_test.go @@ -1,8 +1,10 @@ package main import ( + "net/http" "sync" "testing" + "time" "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) } } + +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") + } +}