Phase 10: hardening and ops

Implements PLAN.md Phase 10.

Simplified Sliding Sync (MSC4186) is served at the unstable
org.matrix.simplified_msc3575 endpoint and advertised in /versions. Lists
are sorted by activity and filtered by DM, invite, encryption, room type,
space and tag; room subscriptions, required_state with wildcards, $ME and
$LAZY, and the to_device, e2ee, account_data, receipts and typing
extensions are supported. Per-connection state is kept in memory and an
unknown pos answers M_UNKNOWN_POS.

Threads and relations: m.relates_to is indexed on roomserver_events
(migration 0012), /relations (with recursion) and /threads are served, and
every event a timeline read returns carries its m.thread, m.replace and
m.reference summaries in unsigned.m.relations. Notifications carry a
thread ID (migration 0013): threaded receipts clear only their thread, and
sync reports unread_thread_notifications when the filter asks.

Spaces: /hierarchy walks a space breadth first, asking other servers about
rooms this one is not in through the new federation hierarchy endpoint;
room_summary describes one room. /search runs on FTS5 (sqlite, pocketbase)
or tsvector (postgres), bounded to what each user may see, with context,
profiles, state, grouping and paging.

The PocketBase driver is real: the sqlite schema and store on an embedded
PocketBase app's database, with media in its filesystem, local or S3
(database.pocketbase.s3). It runs the full storage test suite.

Ops: OpenTelemetry tracing over OTLP/HTTP (tracing.*), with server spans
per route, component spans, and trace propagation on federation, push and
application service calls. New metrics for persisted events, waiting
syncs, sliding sync connections and the database pool, with a Grafana
dashboard. deploy/ adds a Compose setup (PostgreSQL, Prometheus, Grafana,
Jaeger) and a Helm chart; cmd/warp-loadtest loads a server and reports
request and delivery latencies.

The load tester found a long-standing bug, fixed here: on a server where
no account data had been written, every incremental /sync re-sent the push
rules and returned at once, and an all-zero token was taken for an initial
sync, so long polls became busy loops.

The PostgreSQL migrations and search dialect compile but were not run
against a live PostgreSQL in this change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/internal/syncapi/filter.go

@@ -98,8 +98,10 @@ type RoomEventFilter struct {
 	// this were set; see roomserver.shapeState.
 	IncludeRedundantMembers bool `json:"include_redundant_members,omitempty"`
 
-	// UnreadThreadNotifications is a threads feature; threads are
-	// PLAN.md phase 10.
+	// UnreadThreadNotifications splits a joined room's unread counts by
+	// thread: unread_notifications counts the main timeline only, and
+	// unread_thread_notifications has the counts of each thread. Only the
+	// room timeline's flag is read, which is where the spec puts it.
 	UnreadThreadNotifications bool `json:"unread_thread_notifications,omitempty"`
 }
 
@@ -221,6 +223,12 @@ func (f Filter) LazyLoadMembers() bool {
 	return f.Room != nil && f.Room.Timeline != nil && f.Room.Timeline.LazyLoadMembers
 }
 
+// UnreadThreadNotifications reports whether the filter asks for unread
+// counts per thread.
+func (f Filter) UnreadThreadNotifications() bool {
+	return f.Room != nil && f.Room.Timeline != nil && f.Room.Timeline.UnreadThreadNotifications
+}
+
 // IncludeLeave reports whether the rooms the user has left should be
 // reported in full rather than only when the leave is new.
 func (f Filter) IncludeLeave() bool {

server/internal/syncapi/notifications.go

@@ -3,6 +3,8 @@ package syncapi
 import (
 	"context"
 	"slices"
+
+	"github.com/dariuszjaniak/warp/server/internal/storage"
 )
 
 // pushRulesType is the account data push rules travel as. The push server
@@ -24,7 +26,15 @@ type UnreadNotifications struct {
 // counts moved: a room is in a response because something in it changed,
 // and a client showing a new message wants the count beside it whether or
 // not this particular sync is the one that moved it.
-func (s *API) unreadNotifications(ctx context.Context, userID string, resp *Response) error {
+//
+// With byThread — the filter's unread_thread_notifications — the counts
+// are split: unread_notifications is the main timeline's, and each thread
+// with something unread is in unread_thread_notifications. A thread read
+// to the end drops out of the map rather than being reported as zero,
+// which the spec allows and a client takes the same way: the room is in
+// the response because its counts moved, and the map it is sent stands
+// for all of the room's threads, so a thread missing from it is read.
+func (s *API) unreadNotifications(ctx context.Context, userID string, byThread bool, resp *Response) error {
 	if s.push == nil || len(resp.Rooms.Join) == 0 {
 		return nil
 	}
@@ -34,6 +44,35 @@ func (s *API) unreadNotifications(ctx context.Context, userID string, resp *Resp
 	}
 	slices.Sort(roomIDs)
 
+	if byThread {
+		counts, err := s.push.ThreadNotificationCounts(ctx, userID, roomIDs)
+		if err != nil {
+			return err
+		}
+		for _, roomID := range roomIDs {
+			room := resp.Rooms.Join[roomID]
+			main := counts[roomID][storage.ThreadMain]
+			room.UnreadNotifications = &UnreadNotifications{
+				NotificationCount: main.Notifications,
+				HighlightCount:    main.Highlights,
+			}
+			for thread, c := range counts[roomID] {
+				if thread == storage.ThreadMain {
+					continue
+				}
+				if room.UnreadThreadNotifications == nil {
+					room.UnreadThreadNotifications = map[string]UnreadNotifications{}
+				}
+				room.UnreadThreadNotifications[thread] = UnreadNotifications{
+					NotificationCount: c.Notifications,
+					HighlightCount:    c.Highlights,
+				}
+			}
+			resp.Rooms.Join[roomID] = room
+		}
+		return nil
+	}
+
 	counts, err := s.push.NotificationCounts(ctx, userID, roomIDs)
 	if err != nil {
 		return err

server/internal/syncapi/relations.go

@@ -0,0 +1,197 @@
+package syncapi
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// GET /rooms/{roomID}/relations/{eventID} and GET /rooms/{roomID}/threads:
+// the children of an event, and the threads of a room.
+//
+// The reads are the roomserver's (relations.go there); what is here is the
+// token handling, which is this package's. Relations page exactly like
+// /messages and take the same tokens — a client may start paging an
+// event's children from a sync's prev_batch — so they use the same
+// stream-position token. Threads are ordered by their latest activity
+// rather than by any one stream, and have a token of their own.
+
+// DefaultRelationsLimit is the page size of /relations and /threads when
+// the client names none.
+const DefaultRelationsLimit = 50
+
+// RelationsRequest is one page of an event's children.
+type RelationsRequest struct {
+	RoomID  string
+	EventID string
+	User    string
+
+	RelType   string
+	EventType string
+
+	// From and To are pagination tokens, as for /messages.
+	From string
+	To   string
+
+	// Direction defaults to backwards, newest first, as the spec has it.
+	Direction storage.TimelineDirection
+
+	Limit   int
+	Recurse bool
+}
+
+// RelationsResponse is one page of an event's children.
+type RelationsResponse struct {
+	Chunk []roomserver.ClientEvent `json:"chunk"`
+
+	// NextBatch continues in the direction asked for; it is left out at
+	// the end.
+	NextBatch string `json:"next_batch,omitempty"`
+
+	// PrevBatch is the token this page began at, when the client gave one.
+	PrevBatch string `json:"prev_batch,omitempty"`
+
+	// RecursionDepth is how deep a recursive read went. It is only there
+	// when the client asked to recurse, which is how it learns the server
+	// honoured the request.
+	RecursionDepth int `json:"recursion_depth,omitempty"`
+}
+
+// Relations reads one page of an event's children.
+func (s *API) Relations(ctx context.Context, req RelationsRequest) (RelationsResponse, error) {
+	from, err := ParseToken(req.From)
+	if err != nil {
+		return RelationsResponse{}, errInvalidTokenf("from: %v", err)
+	}
+	to, err := ParseToken(req.To)
+	if err != nil {
+		return RelationsResponse{}, errInvalidTokenf("to: %v", err)
+	}
+	direction := req.Direction
+	if direction != storage.DirectionForward {
+		direction = storage.DirectionBackward
+	}
+	limit := req.Limit
+	if limit <= 0 {
+		limit = DefaultRelationsLimit
+	}
+	limit = min(limit, MaxTimelineLimit)
+
+	low, high := from.PDU, to.PDU
+	if direction == storage.DirectionBackward {
+		low, high = to.PDU, from.PDU
+	}
+
+	page, err := s.rooms.Relations(ctx, roomserver.RelationsRequest{
+		RoomID:    req.RoomID,
+		EventID:   req.EventID,
+		User:      req.User,
+		RelType:   req.RelType,
+		EventType: req.EventType,
+		Low:       low,
+		High:      high,
+		Direction: direction,
+		Limit:     limit,
+		Recurse:   req.Recurse,
+	})
+	if err != nil {
+		return RelationsResponse{}, err
+	}
+
+	chunk := page.Events
+	if direction == storage.DirectionBackward {
+		chunk = reversed(chunk)
+	}
+	if chunk == nil {
+		chunk = []roomserver.ClientEvent{}
+	}
+	out := RelationsResponse{Chunk: chunk}
+	if req.From != "" {
+		out.PrevBatch = req.From
+	}
+	if page.Limited {
+		if direction == storage.DirectionBackward {
+			out.NextBatch = pduToken(page.Start).String()
+		} else {
+			out.NextBatch = pduToken(page.End).String()
+		}
+	}
+	if req.Recurse {
+		out.RecursionDepth = page.Depth
+	}
+	return out, nil
+}
+
+// ThreadsRequest is one page of a room's threads.
+type ThreadsRequest struct {
+	RoomID string
+	User   string
+
+	// Include is "all" or "participated".
+	Include string
+
+	// From is a token from a previous page's next_batch.
+	From string
+
+	Limit int
+}
+
+// ThreadsResponse is one page of a room's threads.
+type ThreadsResponse struct {
+	Chunk     []roomserver.ClientEvent `json:"chunk"`
+	NextBatch string                   `json:"next_batch,omitempty"`
+}
+
+// threadTokenPrefix marks a /threads token, which is a position in thread
+// activity rather than a sync token.
+const threadTokenPrefix = "t"
+
+// Threads reads one page of a room's threads.
+func (s *API) Threads(ctx context.Context, req ThreadsRequest) (ThreadsResponse, error) {
+	var participated bool
+	switch req.Include {
+	case "", "all":
+	case "participated":
+		participated = true
+	default:
+		return ThreadsResponse{}, errInvalidParamf("include must be \"all\" or \"participated\"")
+	}
+
+	var before int64
+	if req.From != "" {
+		raw, ok := strings.CutPrefix(req.From, threadTokenPrefix)
+		value, err := strconv.ParseInt(raw, 10, 64)
+		if !ok || err != nil {
+			return ThreadsResponse{}, errInvalidTokenf("%q is not a /threads token", req.From)
+		}
+		before = value
+	}
+	limit := req.Limit
+	if limit <= 0 {
+		limit = DefaultRelationsLimit
+	}
+	limit = min(limit, MaxTimelineLimit)
+
+	page, err := s.rooms.Threads(ctx, roomserver.ThreadsRequest{
+		RoomID:       req.RoomID,
+		User:         req.User,
+		Participated: participated,
+		Before:       before,
+		Limit:        limit,
+	})
+	if err != nil {
+		return ThreadsResponse{}, err
+	}
+	out := ThreadsResponse{Chunk: page.Roots}
+	if out.Chunk == nil {
+		out.Chunk = []roomserver.ClientEvent{}
+	}
+	if page.Next != 0 {
+		out.NextBatch = fmt.Sprintf("%s%d", threadTokenPrefix, page.Next)
+	}
+	return out, nil
+}

server/internal/syncapi/search.go

@@ -0,0 +1,362 @@
+package syncapi
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// POST /search: full-text search of the rooms a user is or was in.
+//
+// The roomserver decides where the user may look and the store runs the
+// query; what is here is the request and response shape, which carries a
+// fair amount besides the hits — the events around each one, the profiles of
+// their senders, the rooms' state, and the hits grouped by room or sender —
+// all of it optional and all of it what a client needs to show a result
+// without a second round of requests.
+
+// DefaultSearchLimit is the page size when the filter names none.
+const DefaultSearchLimit = 10
+
+// MaxSearchLimit caps a page. Every hit can bring its context, its
+// senders' profiles and its room's state with it, so a page is far heavier
+// than its count suggests.
+const MaxSearchLimit = 100
+
+// MaxSearchContext caps before_limit and after_limit.
+const MaxSearchContext = 20
+
+// SearchRequest is the body of a /search, with the caller.
+type SearchRequest struct {
+	User string
+
+	// NextBatch is the ?next_batch= of a later page.
+	NextBatch string
+
+	Body SearchBody
+}
+
+// SearchBody is the request body.
+type SearchBody struct {
+	SearchCategories struct {
+		RoomEvents *RoomEventsCriteria `json:"room_events"`
+	} `json:"search_categories"`
+}
+
+// RoomEventsCriteria is the one search category the spec defines.
+type RoomEventsCriteria struct {
+	SearchTerm   string           `json:"search_term"`
+	Keys         []string         `json:"keys,omitempty"`
+	Filter       *RoomEventFilter `json:"filter,omitempty"`
+	OrderBy      string           `json:"order_by,omitempty"`
+	EventContext *struct {
+		BeforeLimit    *int `json:"before_limit,omitempty"`
+		AfterLimit     *int `json:"after_limit,omitempty"`
+		IncludeProfile bool `json:"include_profile,omitempty"`
+	} `json:"event_context,omitempty"`
+	IncludeState bool `json:"include_state,omitempty"`
+	Groupings    *struct {
+		GroupBy []struct {
+			Key string `json:"key"`
+		} `json:"group_by"`
+	} `json:"groupings,omitempty"`
+}
+
+// SearchResponse is the body of the answer.
+type SearchResponse struct {
+	SearchCategories struct {
+		RoomEvents RoomEventsResults `json:"room_events"`
+	} `json:"search_categories"`
+}
+
+// RoomEventsResults is one page of hits.
+type RoomEventsResults struct {
+	Count      int                                    `json:"count"`
+	Highlights []string                               `json:"highlights"`
+	Results    []SearchHit                            `json:"results"`
+	NextBatch  string                                 `json:"next_batch,omitempty"`
+	State      map[string][]roomserver.ClientEvent    `json:"state,omitempty"`
+	Groups     map[string]map[string]SearchGroupEntry `json:"groups,omitempty"`
+}
+
+// SearchHit is one result.
+type SearchHit struct {
+	Rank    float64                `json:"rank"`
+	Result  roomserver.ClientEvent `json:"result"`
+	Context *SearchContext         `json:"context,omitempty"`
+}
+
+// SearchContext is what surrounds one hit.
+type SearchContext struct {
+	Start        string                       `json:"start,omitempty"`
+	End          string                       `json:"end,omitempty"`
+	EventsBefore []roomserver.ClientEvent     `json:"events_before"`
+	EventsAfter  []roomserver.ClientEvent     `json:"events_after"`
+	ProfileInfo  map[string]SearchProfileInfo `json:"profile_info,omitempty"`
+}
+
+// SearchProfileInfo is a sender's name and avatar.
+type SearchProfileInfo struct {
+	DisplayName string `json:"displayname,omitempty"`
+	AvatarURL   string `json:"avatar_url,omitempty"`
+}
+
+// SearchGroupEntry is one group of a grouped search.
+type SearchGroupEntry struct {
+	Results   []string `json:"results"`
+	Order     int      `json:"order"`
+	NextBatch string   `json:"next_batch,omitempty"`
+}
+
+// The search tokens. A ranked search pages by offset, because the ranking
+// of a result does not move as later events arrive; a search by recency
+// pages by stream position, which keeps a page from repeating a result when
+// new messages land between two requests.
+const (
+	rankTokenPrefix   = "r"
+	recentTokenPrefix = "p"
+)
+
+// Search answers one /search.
+func (s *API) Search(ctx context.Context, req SearchRequest) (SearchResponse, error) {
+	criteria := req.Body.SearchCategories.RoomEvents
+	var resp SearchResponse
+	resp.SearchCategories.RoomEvents.Highlights = []string{}
+	resp.SearchCategories.RoomEvents.Results = []SearchHit{}
+	if criteria == nil {
+		// The only category there is was not asked for, so there is
+		// nothing to search.
+		return resp, nil
+	}
+	if strings.TrimSpace(criteria.SearchTerm) == "" {
+		return SearchResponse{}, errInvalidParamf("search_term must not be empty")
+	}
+
+	order := storage.SearchByRank
+	switch criteria.OrderBy {
+	case "", "rank":
+	case "recent":
+		order = storage.SearchByRecent
+	default:
+		return SearchResponse{}, errInvalidParamf("order_by must be \"rank\" or \"recent\"")
+	}
+	for _, key := range criteria.Keys {
+		switch key {
+		case "content.body", "content.name", "content.topic":
+		default:
+			return SearchResponse{}, errInvalidParamf("%q is not a search key", key)
+		}
+	}
+
+	offset, before, err := parseSearchToken(req.NextBatch, order)
+	if err != nil {
+		return SearchResponse{}, err
+	}
+
+	limit := DefaultSearchLimit
+	var filter storage.TimelineFilter
+	var rooms, notRooms []string
+	if f := criteria.Filter; f != nil {
+		if f.Limit != nil && *f.Limit > 0 {
+			limit = min(*f.Limit, MaxSearchLimit)
+		}
+		filter = roomEventFilter(f)
+		rooms, notRooms = f.Rooms, f.NotRooms
+	}
+
+	found, total, err := s.rooms.Search(ctx, roomserver.SearchRequest{
+		User:     req.User,
+		Term:     criteria.SearchTerm,
+		Keys:     criteria.Keys,
+		Rooms:    rooms,
+		NotRooms: notRooms,
+		Filter:   filter,
+		Order:    order,
+		Offset:   offset,
+		Before:   before,
+		Limit:    limit,
+	})
+	if err != nil {
+		return SearchResponse{}, err
+	}
+
+	out := &resp.SearchCategories.RoomEvents
+	out.Count = total
+	out.Highlights = storage.SearchWords(criteria.SearchTerm)
+
+	for _, hit := range found {
+		h := SearchHit{Rank: hit.Rank, Result: hit.Event}
+		if criteria.EventContext != nil {
+			if h.Context, err = s.searchContext(ctx, req.User, hit, criteria); err != nil {
+				return SearchResponse{}, err
+			}
+		}
+		out.Results = append(out.Results, h)
+	}
+
+	// There is more when the page came back full and the total says so.
+	if len(found) == limit {
+		switch order {
+		case storage.SearchByRank:
+			if offset+len(found) < total {
+				out.NextBatch = rankTokenPrefix + strconv.Itoa(offset+len(found))
+			}
+		case storage.SearchByRecent:
+			out.NextBatch = recentTokenPrefix + strconv.FormatInt(found[len(found)-1].Position, 10)
+		}
+	}
+
+	if criteria.IncludeState {
+		out.State = map[string][]roomserver.ClientEvent{}
+		for _, hit := range found {
+			roomID := hit.Event.RoomID
+			if _, done := out.State[roomID]; done {
+				continue
+			}
+			state, err := s.rooms.RoomState(ctx, roomID, req.User)
+			if err != nil {
+				return SearchResponse{}, err
+			}
+			out.State[roomID] = state
+		}
+	}
+
+	if criteria.Groupings != nil {
+		out.Groups = searchGroups(found, criteria, out.NextBatch)
+	}
+	return resp, nil
+}
+
+// searchContext reads the events around one hit.
+func (s *API) searchContext(ctx context.Context, userID string, hit roomserver.SearchResult, criteria *RoomEventsCriteria) (*SearchContext, error) {
+	limitOf := func(p *int) int {
+		if p == nil {
+			return 5
+		}
+		return max(0, min(*p, MaxSearchContext))
+	}
+	beforeLimit := limitOf(criteria.EventContext.BeforeLimit)
+	afterLimit := limitOf(criteria.EventContext.AfterLimit)
+	roomID := hit.Event.RoomID
+
+	out := &SearchContext{
+		EventsBefore: []roomserver.ClientEvent{},
+		EventsAfter:  []roomserver.ClientEvent{},
+	}
+	if beforeLimit > 0 {
+		earlier, err := s.rooms.Timeline(ctx, roomserver.TimelineRequest{
+			RoomID:    roomID,
+			User:      userID,
+			High:      storage.PositionBefore(hit.Position),
+			Direction: storage.DirectionBackward,
+			Limit:     beforeLimit,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if len(earlier.Events) > 0 {
+			out.EventsBefore = reversed(earlier.Events)
+			out.Start = pduToken(earlier.Start).String()
+		}
+	}
+	if afterLimit > 0 {
+		later, err := s.rooms.Timeline(ctx, roomserver.TimelineRequest{
+			RoomID:    roomID,
+			User:      userID,
+			Low:       hit.Position,
+			Direction: storage.DirectionForward,
+			Limit:     afterLimit,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if len(later.Events) > 0 {
+			out.EventsAfter = later.Events
+			out.End = pduToken(later.End).String()
+		}
+	}
+
+	if criteria.EventContext.IncludeProfile {
+		senders := sendersOf(append(append([]roomserver.ClientEvent{hit.Event}, out.EventsBefore...), out.EventsAfter...))
+		members, err := s.rooms.MemberEvents(ctx, roomID, senders)
+		if err != nil {
+			return nil, err
+		}
+		out.ProfileInfo = map[string]SearchProfileInfo{}
+		for _, m := range members {
+			if m.StateKey == nil {
+				continue
+			}
+			var content struct {
+				DisplayName string `json:"displayname"`
+				AvatarURL   string `json:"avatar_url"`
+			}
+			_ = json.Unmarshal(m.Content, &content)
+			out.ProfileInfo[*m.StateKey] = SearchProfileInfo{DisplayName: content.DisplayName, AvatarURL: content.AvatarURL}
+		}
+	}
+	return out, nil
+}
+
+// searchGroups groups a page of hits by room or sender, in the order the
+// page ranked them.
+func searchGroups(found []roomserver.SearchResult, criteria *RoomEventsCriteria, nextBatch string) map[string]map[string]SearchGroupEntry {
+	out := map[string]map[string]SearchGroupEntry{}
+	for _, g := range criteria.Groupings.GroupBy {
+		var keyOf func(roomserver.SearchResult) string
+		switch g.Key {
+		case "room_id":
+			keyOf = func(r roomserver.SearchResult) string { return r.Event.RoomID }
+		case "sender":
+			keyOf = func(r roomserver.SearchResult) string { return r.Event.Sender }
+		default:
+			continue
+		}
+		groups := map[string]SearchGroupEntry{}
+		for _, hit := range found {
+			key := keyOf(hit)
+			entry, ok := groups[key]
+			if !ok {
+				entry.Order = len(groups) + 1
+				entry.NextBatch = nextBatch
+			}
+			entry.Results = append(entry.Results, hit.Event.EventID)
+			groups[key] = entry
+		}
+		out[g.Key] = groups
+	}
+	return out
+}
+
+// parseSearchToken reads a ?next_batch=.
+func parseSearchToken(raw string, order storage.SearchOrder) (offset int, before int64, err error) {
+	if raw == "" {
+		return 0, 0, nil
+	}
+	bad := errInvalidTokenf("%q is not a search token", raw)
+	switch order {
+	case storage.SearchByRecent:
+		body, ok := strings.CutPrefix(raw, recentTokenPrefix)
+		if !ok {
+			return 0, 0, bad
+		}
+		if before, err = strconv.ParseInt(body, 10, 64); err != nil {
+			return 0, 0, bad
+		}
+		return 0, before, nil
+	default:
+		body, ok := strings.CutPrefix(raw, rankTokenPrefix)
+		if !ok {
+			return 0, 0, bad
+		}
+		if offset, err = strconv.Atoi(body); err != nil || offset < 0 {
+			return 0, 0, fmt.Errorf("%w", bad)
+		}
+		return offset, 0, nil
+	}
+}

server/internal/syncapi/slidingconn.go

@@ -0,0 +1,219 @@
+package syncapi
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Sliding sync connections: what the server remembers about what each
+// client has already been sent.
+//
+// A classic sync token is a position in every stream and nothing else; the
+// client is sent everything after it. Sliding sync cannot work that way,
+// because what a client is sent depends on what it asked for — which rooms
+// are in the window of which list, with which pieces of state — and that
+// changes from request to request. So the server remembers, per connection,
+// which rooms it has sent and as of which position, and `pos` names one of
+// those memories.
+//
+// The memory is in process and is lost on restart. That is what the
+// protocol expects: an unknown `pos` is answered with M_UNKNOWN_POS, and
+// the client starts a new connection — one initial request, after which it
+// is in step again. Persisting it would buy a slightly cheaper restart for
+// a write on every request.
+
+// ErrUnknownPos is a `pos` this server does not remember: expired, from
+// before a restart, or never handed out.
+var ErrUnknownPos = errors.New("unknown sliding sync position")
+
+// Limits of the connection store.
+const (
+	// connectionTTL is how long an idle connection is remembered. A client
+	// that comes back later starts again.
+	connectionTTL = 30 * time.Minute
+
+	// maxConnectionsPerDevice caps the connections one device may hold
+	// open, each conn_id being one. The oldest goes first.
+	maxConnectionsPerDevice = 10
+
+	// keptGenerations is how many of a connection's latest positions stay
+	// usable. A client that never saw a response retries with the pos it
+	// sent, so the one before the latest has to still work.
+	keptGenerations = 3
+)
+
+// sentRoom is what a connection has been sent of one room.
+type sentRoom struct {
+	// pos is the event stream position the room was last sent up to.
+	pos int64
+
+	// config is the room's required_state and timeline_limit as of that
+	// send, rendered as a string: when it changes, the state the client
+	// holds is no longer what it asked for.
+	config string
+
+	// membership is the user's membership as of that send.
+	membership string
+
+	// typing is the typing notification last sent, as its JSON.
+	typing string
+}
+
+// connSnapshot is a connection as of one `pos`.
+type connSnapshot struct {
+	// token is the stream positions the response at this pos read up to.
+	token Token
+
+	// startPDU is the event stream position the connection began at: rooms
+	// the user left before it are not news to it.
+	startPDU int64
+
+	rooms map[string]sentRoom
+
+	// lists is each list's room count as last sent.
+	lists map[string]int
+
+	// toDevice is where the to_device extension has got to.
+	toDevice int64
+}
+
+// clone copies a snapshot so that a new generation can be built from it
+// without touching the one a retry might still need.
+func (c *connSnapshot) clone() *connSnapshot {
+	return &connSnapshot{
+		token:    c.token,
+		startPDU: c.startPDU,
+		rooms:    maps.Clone(c.rooms),
+		lists:    maps.Clone(c.lists),
+		toDevice: c.toDevice,
+	}
+}
+
+// slidingConn is one connection: a client device's conn_id.
+type slidingConn struct {
+	// nonce makes the connection's positions its own: a pos from another
+	// connection, or from before a restart, cannot name one of these
+	// generations by accident.
+	nonce string
+
+	generations map[int64]*connSnapshot
+	latest      int64
+	lastUsed    time.Time
+}
+
+type connKey struct {
+	user, device, connID string
+}
+
+// connStore holds every connection.
+type connStore struct {
+	mu    sync.Mutex
+	conns map[connKey]*slidingConn
+	now   func() time.Time
+}
+
+func newConnStore(now func() time.Time) *connStore {
+	return &connStore{conns: map[connKey]*slidingConn{}, now: now}
+}
+
+// load returns the snapshot a request's pos names, or a fresh one for a
+// request with none. A request without a pos starts the connection over,
+// which is what a client does after M_UNKNOWN_POS and on first launch.
+func (s *connStore) load(key connKey, pos string) (*connSnapshot, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.sweep()
+
+	if pos == "" {
+		delete(s.conns, key)
+		return nil, nil
+	}
+	conn, ok := s.conns[key]
+	if !ok {
+		return nil, fmt.Errorf("%w: the connection has expired", ErrUnknownPos)
+	}
+	nonce, raw, ok := strings.Cut(pos, ".")
+	gen, err := strconv.ParseInt(raw, 10, 64)
+	if !ok || err != nil || nonce != conn.nonce {
+		return nil, fmt.Errorf("%w: %q is not a position of this connection", ErrUnknownPos, pos)
+	}
+	snap, ok := conn.generations[gen]
+	if !ok {
+		return nil, fmt.Errorf("%w: %q is too old", ErrUnknownPos, pos)
+	}
+	conn.lastUsed = s.now()
+	return snap, nil
+}
+
+// save stores a new generation of a connection and returns its pos.
+func (s *connStore) save(key connKey, snap *connSnapshot) string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	conn, ok := s.conns[key]
+	if !ok {
+		conn = &slidingConn{nonce: randomNonce(), generations: map[int64]*connSnapshot{}, lastUsed: s.now()}
+		s.conns[key] = conn
+		s.capDevice(key)
+	}
+	conn.latest++
+	conn.generations[conn.latest] = snap
+	delete(conn.generations, conn.latest-keptGenerations)
+	conn.lastUsed = s.now()
+	return conn.nonce + "." + strconv.FormatInt(conn.latest, 10)
+}
+
+// sweep forgets idle connections. It runs under the lock on every load,
+// which keeps the store bounded without a goroutine of its own.
+func (s *connStore) sweep() {
+	cutoff := s.now().Add(-connectionTTL)
+	for key, conn := range s.conns {
+		if conn.lastUsed.Before(cutoff) {
+			delete(s.conns, key)
+		}
+	}
+}
+
+// capDevice drops a device's oldest connections beyond the cap. The
+// connection being saved is never one of them.
+func (s *connStore) capDevice(key connKey) {
+	var others []connKey
+	for k := range s.conns {
+		if k != key && k.user == key.user && k.device == key.device {
+			others = append(others, k)
+		}
+	}
+	slices.SortFunc(others, func(a, b connKey) int {
+		return s.conns[a].lastUsed.Compare(s.conns[b].lastUsed)
+	})
+	for len(others) > maxConnectionsPerDevice-1 {
+		delete(s.conns, others[0])
+		others = others[1:]
+	}
+}
+
+// connections reports how many connections are held, for tests and
+// metrics.
+func (s *connStore) connections() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.conns)
+}
+
+func randomNonce() string {
+	var b [6]byte
+	_, _ = rand.Read(b[:])
+	return hex.EncodeToString(b[:])
+}
+
+// SlidingConnections reports how many sliding sync connections the server
+// is holding state for.
+func (s *API) SlidingConnections() int { return s.conns.connections() }

server/internal/syncapi/sync.go

@@ -7,9 +7,12 @@ import (
 	"slices"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
+
 	"github.com/dariuszjaniak/warp/server/internal/matrix/event"
 	"github.com/dariuszjaniak/warp/server/internal/roomserver"
 	"github.com/dariuszjaniak/warp/server/internal/storage"
+	"github.com/dariuszjaniak/warp/server/internal/tracing"
 )
 
 // GET /sync: the endpoint a client spends its life in.
@@ -103,6 +106,12 @@ type JoinedRoom struct {
 	// it out rather than claiming zero: an absent count says nothing,
 	// where a zero says "you have read everything".
 	UnreadNotifications *UnreadNotifications `json:"unread_notifications,omitempty"`
+
+	// UnreadThreadNotifications is the counts of each thread with
+	// something unread, keyed by the thread root's event ID, for a filter
+	// that asks for them; UnreadNotifications then counts the main
+	// timeline alone.
+	UnreadThreadNotifications map[string]UnreadNotifications `json:"unread_thread_notifications,omitempty"`
 }
 
 // InvitedRoom is a room the user has been invited to. It carries no
@@ -266,7 +275,10 @@ func (s *API) Sync(ctx context.Context, req SyncRequest) (Response, error) {
 // blocking on a change it has just missed. The listener is returned rather
 // than closed here because the caller is the one that knows whether it is
 // about to wait on it.
-func (s *API) attempt(ctx context.Context, req SyncRequest, filter Filter, since Token) (Response, bool, *Listener, error) {
+func (s *API) attempt(ctx context.Context, req SyncRequest, filter Filter, since Token) (_ Response, _ bool, _ *Listener, err error) {
+	ctx, span := tracing.Start(ctx, "syncapi.sync", attribute.Bool("matrix.initial", since.IsInitial()))
+	defer func() { tracing.End(span, err) }()
+
 	rooms, err := s.rooms.UserMemberships(ctx, req.User, nil)
 	if err != nil {
 		return Response{}, false, nil, err
@@ -333,17 +345,18 @@ func (s *API) build(ctx context.Context, req SyncRequest, filter Filter, since T
 	if err := s.buildRooms(ctx, req, filter, since, to, rooms, &resp); err != nil {
 		return Response{}, false, err
 	}
-	if err := s.unreadNotifications(ctx, req.User, &resp); err != nil {
+	if err := s.unreadNotifications(ctx, req.User, filter.UnreadThreadNotifications(), &resp); err != nil {
 		return Response{}, false, err
 	}
 
-	accountData, err := s.globalAccountData(ctx, req, since.AccountData, to.AccountData, filter)
+	accountData, err := s.globalAccountData(ctx, req, since.AccountData, to.AccountData,
+		req.Since == "" || req.FullState, filter)
 	if err != nil {
 		return Response{}, false, err
 	}
 	resp.AccountData.Events = accountData
 
-	toDevice, toDevicePos, err := s.toDeviceMessages(ctx, req.Localpart, req.Device, since.ToDevice, to.ToDevice)
+	toDevice, toDevicePos, err := s.toDeviceMessages(ctx, req.Localpart, req.Device, since.ToDevice, to.ToDevice, 0)
 	if err != nil {
 		return Response{}, false, err
 	}
@@ -373,7 +386,7 @@ func (s *API) build(ctx context.Context, req SyncRequest, filter Filter, since T
 	}
 
 	resp.NextBatch = next.String()
-	return resp, isInteresting(resp, since), nil
+	return resp, isInteresting(resp, req.Since == ""), nil
 }
 
 // isInteresting reports whether a response is worth returning rather than
@@ -385,8 +398,13 @@ func (s *API) build(ctx context.Context, req SyncRequest, filter Filter, since T
 // into a busy loop. So the test is whether the response has anything in it,
 // with an initial sync as the exception: a client that has just logged in
 // needs an answer whatever is in it.
-func isInteresting(resp Response, since Token) bool {
-	if since.IsInitial() {
+//
+// Initial means the client sent no ?since=, not that the token it sent is
+// all zeroes. On a server whose streams have not moved yet — no events, no
+// receipts, nothing — "s0" is a client's every token, and a sync that took
+// that for a first one would never wait.
+func isInteresting(resp Response, initial bool) bool {
+	if initial {
 		return true
 	}
 	return len(resp.Rooms.Join) > 0 ||
@@ -665,7 +683,8 @@ func (s *API) roomEphemeral(ctx context.Context, userID, roomID string, window r
 }
 
 // globalAccountData reads the account data that is not about a room.
-func (s *API) globalAccountData(ctx context.Context, req SyncRequest, from, to int64, filter Filter) ([]AccountDataEvent, error) {
+// initial is whether this is the client's first read of it.
+func (s *API) globalAccountData(ctx context.Context, req SyncRequest, from, to int64, initial bool, filter Filter) ([]AccountDataEvent, error) {
 	entries, err := s.accountDataEntries(ctx, req.Localpart, from, to)
 	if err != nil {
 		return nil, err
@@ -673,7 +692,13 @@ func (s *API) globalAccountData(ctx context.Context, req SyncRequest, from, to i
 	// Every user has push rules, written or not: a user who has never
 	// changed one has the server defaults, and a client reads its rules
 	// from the account data of its first sync whichever it is.
-	if from == 0 && s.push != nil && !slices.ContainsFunc(entries, func(e storage.AccountDataEntry) bool {
+	//
+	// "First" is the client's first sync, not a token whose account data
+	// position is zero: on a server where nobody has written any account
+	// data yet every token has a zero there, and treating each sync as the
+	// first would send the rules every time — which makes every sync
+	// "interesting", and turns every long poll into a busy loop.
+	if initial && s.push != nil && !slices.ContainsFunc(entries, func(e storage.AccountDataEntry) bool {
 		return e.RoomID == "" && e.DataType == pushRulesType
 	}) {
 		entries = append(entries, storage.AccountDataEntry{DataType: pushRulesType})

server/internal/syncapi/syncapi.go

@@ -84,12 +84,14 @@ type Keys interface {
 }
 
 // Push is the part of the push server a sync reads: the push rules it
-// reports as account data, the unread counts of each room, and which
+// reports as account data, the unread counts of each room (and of each
+// thread in it, for a client that asks), and which
 // rooms' counts moved. It is also told about read receipts, which are what
 // clear the counts. *pushserver.API satisfies it.
 type Push interface {
 	Document(ctx context.Context, localpart, userID string) ([]byte, error)
 	NotificationCounts(ctx context.Context, userID string, roomIDs []string) (map[string]storage.NotificationCounts, error)
+	ThreadNotificationCounts(ctx context.Context, userID string, roomIDs []string) (map[string]map[string]storage.NotificationCounts, error)
 	NotificationCountChanges(ctx context.Context, userID string, from, to int64) ([]string, error)
 	ReceiptSet(ctx context.Context, userID, roomID, receiptType, threadID, eventID string)
 }
@@ -104,8 +106,11 @@ type API struct {
 	store      storage.SyncStore
 	notifier   *Notifier
 	typing     *typingTracker
-	log        *slog.Logger
-	now        func() time.Time
+
+	// conns remembers what each sliding sync connection has been sent.
+	conns *connStore
+	log   *slog.Logger
+	now   func() time.Time
 
 	// presenceEnabled is the config switch of PLAN.md open question 3.
 	// With presence off the endpoints answer M_FORBIDDEN and the presence
@@ -227,6 +232,7 @@ func New(opts Options) (*API, error) {
 		store:           opts.Store,
 		notifier:        opts.Notifier,
 		typing:          newTypingTracker(opts.TypingTimeout, opts.Notifier, opts.Now),
+		conns:           newConnStore(opts.Now),
 		log:             opts.Logger,
 		now:             opts.Now,
 		presenceEnabled: opts.PresenceEnabled,

server/internal/syncapi/todevice.go

@@ -140,14 +140,14 @@ type toDeviceEvent struct {
 // position of the last message *delivered* rather than the top of the
 // stream: a queue longer than one batch is drained over several syncs, and
 // a token naming the top would skip the rest.
-func (s *API) toDeviceMessages(ctx context.Context, localpart, deviceID string, from, to int64) ([]toDeviceEvent, int64, error) {
+func (s *API) toDeviceMessages(ctx context.Context, localpart, deviceID string, from, to int64, limit int) ([]toDeviceEvent, int64, error) {
 	if deviceID == "" {
 		// An access token always names a device, so this cannot happen
 		// through the API; a guest session later might.
 		return nil, from, nil
 	}
 
-	msgs, err := s.store.ToDeviceMessages(ctx, localpart, deviceID, from, to, 0)
+	msgs, err := s.store.ToDeviceMessages(ctx, localpart, deviceID, from, to, limit)
 	if err != nil {
 		return nil, 0, err
 	}