syncstate

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 15, 2026 License: MIT Imports: 8 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultDirtyThreshold uint64 = 500

DefaultDirtyThreshold defines the number of updates triggering an early asynchronous disk flush.

View Source 
const DefaultFlushInterval = 50 * time.Millisecond

DefaultFlushInterval defines the periodic disk checkpoint duration (50ms parity with TDLib).

View Source 
const MaxGapTimeout = 700 * time.Millisecond

MaxGapTimeout defines the ceiling timeout after which getDifference is mandatory (700ms).

View Source 
const MaxPendingUpdatesThreshold = 20

MaxPendingUpdatesThreshold defines the maximum unapplied out-of-order updates before forcing a sync.

View Source 
const MinGapTimeout = 50 * time.Millisecond

MinGapTimeout defines the short-gap speculative repair window (50ms).

Variables

This section is empty.

Functions

This section is empty.

Types

type BatchHandler 

type BatchHandler func(batch DifferenceBatch) error

BatchHandler defines the consumer callback signature for processing retrieved difference slices.

type CheckpointFlusher 

type CheckpointFlusher struct {
	// contains filtered or unexported fields
}

CheckpointFlusher governs periodic and synchronous disk flushes to prevent fsync bottlenecks.

func NewCheckpointFlusher 

func NewCheckpointFlusher(clientID int64, vector *StateVector, sink StateStorage, interval time.Duration) *CheckpointFlusher

NewCheckpointFlusher constructs a flusher bound to a StateVector and StateStorage.

func (*CheckpointFlusher) Close 

func (c *CheckpointFlusher) Close() error

Close gracefully terminates the flusher background loop and forces a final flush.

func (*CheckpointFlusher) Flush 

func (c *CheckpointFlusher) Flush(ctx context.Context) error

Flush synchronously writes the current vector snapshot to the persistence sink.

func (*CheckpointFlusher) MarkDirty 

func (c *CheckpointFlusher) MarkDirty(count uint64)

MarkDirty increments the count of uncommitted in-memory updates.

func (*CheckpointFlusher) Start 

func (c *CheckpointFlusher) Start(ctx context.Context)

Start launches the background flusher loop until ctx is canceled or Close is called.

type DifferenceAPI 

type DifferenceAPI interface {
	UpdatesGetDifference(ctx context.Context, request *tg.UpdatesGetDifferenceRequest) (tg.UpdatesDifferenceClass, error)
}

DifferenceAPI abstracts the MTProto RPC method for difference retrieval.

type DifferenceBatch 

type DifferenceBatch struct {
	Messages          []tg.MessageClass
	EncryptedMessages []tg.EncryptedMessageClass
	OtherUpdates      []tg.UpdateClass
	Users             []tg.UserClass
	Chats             []tg.ChatClass
	IsIntermediate    bool
}

DifferenceBatch represents an atomic set of messages and updates retrieved during a difference sync.

type DifferenceFetcher 

type DifferenceFetcher struct {
	// contains filtered or unexported fields
}

DifferenceFetcher coordinates sequential, pull-based getDifference pagination with O(1) streaming checkpoints.

func NewDifferenceFetcher 

func NewDifferenceFetcher(
	api DifferenceAPI,
	vector *StateVector,
	flusher *CheckpointFlusher,
	gapBuffer *GapBuffer,
	onTooLong func(),
) *DifferenceFetcher

NewDifferenceFetcher constructs a DifferenceFetcher bound to state registers and flusher.

func (*DifferenceFetcher) FetchDifference 

func (f *DifferenceFetcher) FetchDifference(ctx context.Context, handler BatchHandler) error

FetchDifference executes paginated difference recovery at the consumer's processing pace. Concurrent calls are strictly coalesced via singleflight to prevent server request flooding.

type GapBuffer 

type GapBuffer struct {
	// contains filtered or unexported fields
}

GapBuffer manages unsequenced updates with micro-repair and escalation timers.

func NewGapBuffer 

func NewGapBuffer(onShortGap ShortGapCallback, onTimeout GapTimeoutCallback) *GapBuffer

NewGapBuffer constructs a GapBuffer with short and maximum repair callback handlers.

func (*GapBuffer) Clear 

func (b *GapBuffer) Clear()

Clear resets the buffer and halts any active timers immediately.

func (*GapBuffer) Len 

func (b *GapBuffer) Len() int

Len returns the count of updates currently held in the buffer.

func (*GapBuffer) PopConsecutive 

func (b *GapBuffer) PopConsecutive(currentPTS int32) ([]PendingUpdate, int32)

PopConsecutive drains all consecutive updates from the buffer starting from currentPTS.

func (*GapBuffer) Push 

func (b *GapBuffer) Push(upd PendingUpdate, currentPTS int32)

Push adds an out-of-order update to the buffer, sorting by PTS ascending and initiating timers.

type GapCheckResult 

type GapCheckResult int

GapCheckResult defines the deterministic action required for an incoming update. 

const (
	// GapApply indicates the update is consecutive and can be processed immediately.
	GapApply GapCheckResult = iota
	// GapIgnore indicates the update is stale or duplicate and must be discarded.
	GapIgnore
	// GapRefetch indicates a gap was detected, requiring buffering or difference recovery.
	GapRefetch
)

func CheckGap 

func CheckGap(localPTS, remotePTS, ptsCount int32) GapCheckResult

CheckGap evaluates sequence mathematical invariants against local state.

type GapTimeoutCallback 

type GapTimeoutCallback func()

GapTimeoutCallback is invoked when max gap duration or update count threshold is exceeded.

type PendingUpdate 

type PendingUpdate struct {
	PTS         int32
	PTSCount    int32
	ReceiveTime time.Time
	Raw         tg.UpdateClass
	TraceID     [16]byte
	ClientID    int64
}

PendingUpdate encapsulates an out-of-order update awaiting gap resolution.

type ShortGapCallback 

type ShortGapCallback func(missingPTS int32)

ShortGapCallback is invoked when a single missing update can be targeted specifically.

type StateSnapshot 

type StateSnapshot struct {
	PTS  int32
	QTS  int32
	Date int32
	Seq  int32
}

StateSnapshot captures an immutable point-in-time state of the sequence counters.

type StateStorage 

type StateStorage interface {
	SaveState(ctx context.Context, clientID int64, snapshot StateSnapshot) error
	LoadState(ctx context.Context, clientID int64) (snapshot StateSnapshot, found bool, err error)
	SaveChannelPTS(ctx context.Context, clientID, channelID int64, pts int32) error
	LoadChannelPTS(ctx context.Context, clientID, channelID int64) (pts int32, found bool, err error)
	ForEachChannelPTS(ctx context.Context, clientID int64, fn func(channelID int64, pts int32) error) error
	SetDialogNotLoaded(ctx context.Context, clientID, dialogID int64, notLoaded bool) error
	IsDialogNotLoaded(ctx context.Context, clientID, dialogID int64) (bool, error)
}

StateStorage represents the persistence driver contract for committing vector checkpoints and channel state tracking, supporting multi-tenant isolation via clientID.

type StateVector 

type StateVector struct {
	// contains filtered or unexported fields
}

StateVector encapsulates atomic sequence registers for MTProto state management.

func NewStateVector 

func NewStateVector(initial StateSnapshot) *StateVector

NewStateVector instantiates a StateVector initialized from an initial snapshot.

func (*StateVector) Date 

func (s *StateVector) Date() int32

Date returns the current global date timestamp.

func (*StateVector) PTS 

func (s *StateVector) PTS() int32

PTS returns the current local PTS counter.

func (*StateVector) QTS 

func (s *StateVector) QTS() int32

QTS returns the current local QTS counter.

func (*StateVector) Seq 

func (s *StateVector) Seq() int32

Seq returns the current local sequence number.

func (*StateVector) SetDate 

func (s *StateVector) SetDate(val int32)

SetDate atomically sets the global date timestamp.

func (*StateVector) SetPTS 

func (s *StateVector) SetPTS(val int32)

SetPTS atomically updates the local PTS counter.

func (*StateVector) SetQTS 

func (s *StateVector) SetQTS(val int32)

SetQTS atomically updates the local QTS counter.

func (*StateVector) SetSeq 

func (s *StateVector) SetSeq(val int32)

SetSeq atomically updates the local sequence number.

func (*StateVector) Snapshot 

func (s *StateVector) Snapshot() StateSnapshot

Snapshot returns an immutable point-in-time copy of all four state counters.

func (*StateVector) UpdateDate 

func (s *StateVector) UpdateDate(val int32) bool

UpdateDate monotonically advances the date timestamp if val is strictly greater.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.