Silverfish/proton-bridge
1
0
Fork 1
mirror of https://github.com/ProtonMail/proton-bridge.git synced 2025-12-10 20:56:51 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(GODT-2327): Only start processing events once sync is finished

Browse Source
This commit is contained in:
James Houlahan
2023-02-06 10:32:13 +01:00
committed by Jakub
parent 232875d5cc
commit 63bc87cc86
2 changed files with 52 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
internal/user/user.go
View File

@ -187,12 +187,23 @@ func New(
user.log.Debug("Sync triggered")
user.abortable.Do(ctx, func(ctx context.Context) {
if user.vault.SyncStatus().IsComplete() {
user.log.Debug("Sync is already complete, skipping")
} else if err := user.doSync(ctx); err != nil {
user.log.WithError(err).Error("Failed to sync, will retry later")
time.AfterFunc(SyncRetryCooldown, user.goSync)
if !user.vault.SyncStatus().IsComplete() {
if err := user.doSync(ctx); err != nil {
user.log.WithError(err).Error("Failed to sync, will retry later")
go func() {
select {
case <-ctx.Done():
user.log.WithError(err).Warn("Aborting sync retry")
case <-time.After(SyncRetryCooldown):
user.goSync()
}
}()
}
}
// Once we know the sync has completed, we can start polling for API events.
user.startEvents(ctx)
})
})
@ -203,44 +214,6 @@ func (user *User) TriggerSync() {
user.goSync()
}
func (user *User) StartEvents() {
// Stream events from the API, logging any errors that occur.
// This does nothing until the sync has been marked as complete.
// When we receive an API event, we attempt to handle it.
// If successful, we update the event ID in the vault.
user.tasks.Once(func(ctx context.Context) {
ticker := proton.NewTicker(EventPeriod, EventJitter)
defer ticker.Stop()
for {
var doneCh chan struct{}
select {
case <-ctx.Done():
return
case doneCh = <-user.pollAPIEventsCh:
// ...
case <-ticker.C:
// ...
}
user.log.Debug("Event poll triggered")
if !user.vault.SyncStatus().IsComplete() {
user.log.Debug("Sync is incomplete, skipping event poll")
} else if err := user.doEventPoll(ctx); err != nil {
user.log.WithError(err).Error("Failed to poll events")
}
if doneCh != nil {
close(doneCh)
}
}
})
}
// ID returns the user's ID.
func (user *User) ID() string {
return safe.RLockRet(func() string {
@ -589,6 +562,42 @@ func (user *User) initUpdateCh(mode vault.AddressMode) {
}
}
// startEvents streams events from the API, logging any errors that occur.
// This does nothing until the sync has been marked as complete.
// When we receive an API event, we attempt to handle it.
// If successful, we update the event ID in the vault.
func (user *User) startEvents(ctx context.Context) {
ticker := proton.NewTicker(EventPeriod, EventJitter)
defer ticker.Stop()
for {
var doneCh chan struct{}
select {
case <-ctx.Done():
return
case doneCh = <-user.pollAPIEventsCh:
// ...
case <-ticker.C:
// ...
}
user.log.Debug("Event poll triggered")
if !user.vault.SyncStatus().IsComplete() {
user.log.Debug("Sync is incomplete, skipping event poll")
} else if err := user.doEventPoll(ctx); err != nil {
user.log.WithError(err).Error("Failed to poll events")
}
if doneCh != nil {
close(doneCh)
}
}
}
// doEventPoll is called whenever API events should be polled.
func (user *User) doEventPoll(ctx context.Context) error {
user.eventLock.Lock()