Fixed packet loss bug due to client not waiting before exiting for the

writes to fully proccess
This commit is contained in:
2025-05-23 10:30:24 +03:00
parent d6a2eda646
commit 8aa5cbb72e
7 changed files with 154 additions and 80 deletions

View File

@@ -124,8 +124,9 @@ type Model struct {
index int
selectedMessage *data.Message
editingMessage *data.Message
lastReadMsg *snowflake.ID
keepLastRead bool
previousLastReadMsg *snowflake.ID
keepPreviousLastRead bool
messagesHeight int
maxMessagesHeight int
@@ -142,27 +143,27 @@ func New() Model {
vi := viminput.New()
return Model{
vi: vi,
focus: false,
locked: false,
hasReadAccess: false,
hasWriteAccess: false,
networkIndex: -1,
receiverIndex: -1,
frequencyIndex: -1,
base: SnapToBottom,
index: Unselected,
selectedMessage: nil,
editingMessage: nil,
lastReadMsg: nil,
keepLastRead: false,
messagesHeight: 0,
maxMessagesHeight: -1,
messagesCache: nil,
prerender: "",
width: -1,
style: blurStyle(),
borderStyle: ViBlurredBorder(),
vi: vi,
focus: false,
locked: false,
hasReadAccess: false,
hasWriteAccess: false,
networkIndex: -1,
receiverIndex: -1,
frequencyIndex: -1,
base: SnapToBottom,
index: Unselected,
selectedMessage: nil,
editingMessage: nil,
previousLastReadMsg: nil,
keepPreviousLastRead: false,
messagesHeight: 0,
maxMessagesHeight: -1,
messagesCache: nil,
prerender: "",
width: -1,
style: blurStyle(),
borderStyle: ViBlurredBorder(),
}
}
@@ -215,16 +216,30 @@ func (m Model) Update(msg tea.Msg) (Model, tea.Cmd) {
}
lastMsg := state.GetLastMessage(frequency.ID)
if m.base != SnapToBottom {
m.keepLastRead = false
} else if lastMsg != nil {
lastReadMsg := state.State.LastReadMessages[frequency.ID]
if m.keepLastRead && m.lastReadMsg != nil && lastReadMsg != nil &&
*lastReadMsg != *lastMsg && *lastReadMsg != *m.lastReadMsg {
m.keepLastRead = false
}
if lastMsg != nil {
if m.base == SnapToBottom && lastMsg != nil {
// lastReadMsg := state.State.LastReadMessages[frequency.ID]
// if m.keepPreviousLastRead && m.previousLastReadMsg != nil && lastReadMsg != nil &&
// *lastReadMsg != *lastMsg && *lastReadMsg != *m.previousLastReadMsg {
// m.keepPreviousLastRead = false
// }
state.State.LastReadMessages[frequency.ID] = lastMsg
log.Println("HIT")
state.State.LastReadMessages[frequency.ID] = lastMsg
delete(state.State.RemoteNotifications, frequency.ID)
} else if m.base == SnapToBottom {
assert.Never("lastMsg is nil")
}
if m.base != SnapToBottom {
assert.Assert(
m.keepPreviousLastRead, "debug reached m.keep",
"m.base", m.base,
"lastMsg", lastMsg,
"lastReadMessage", state.State.LastReadMessages[frequency.ID],
"m.previousLastReadMessage", m.previousLastReadMsg,
)
m.keepPreviousLastRead = false
}
}
m.hasReadAccess = frequency.Perms != packet.PermNoAccess || member.IsAdmin
@@ -267,12 +282,12 @@ func (m Model) Update(msg tea.Msg) (Model, tea.Cmd) {
lastMsg := state.GetLastMessage(receiverId)
if m.base != SnapToBottom {
m.keepLastRead = false
m.keepPreviousLastRead = false
} else if lastMsg != nil {
lastReadMsg := state.State.LastReadMessages[receiverId]
if m.keepLastRead && m.lastReadMsg != nil && lastReadMsg != nil &&
*lastReadMsg != *lastMsg && *lastReadMsg != *m.lastReadMsg {
m.keepLastRead = false
if m.keepPreviousLastRead && m.previousLastReadMsg != nil && lastReadMsg != nil &&
*lastReadMsg != *lastMsg && *lastReadMsg != *m.previousLastReadMsg {
m.keepPreviousLastRead = false
}
state.State.LastReadMessages[receiverId] = lastMsg
@@ -820,7 +835,8 @@ func (m *Model) ResetBeforeSwitch() {
m.base = SnapToBottom
m.SetIndex(Unselected)
m.maxMessagesHeight = -1
m.lastReadMsg = nil
m.previousLastReadMsg = nil
m.keepPreviousLastRead = false
}()
networkId := state.NetworkId(m.networkIndex)
@@ -856,10 +872,10 @@ func (m *Model) RestoreAfterSwitch() tea.Cmd {
frequency := frequencies[m.frequencyIndex]
log.Println("Restoring frequency:", frequency.ID)
m.lastReadMsg = state.State.LastReadMessages[frequency.ID]
m.previousLastReadMsg = state.State.LastReadMessages[frequency.ID]
lastMsg := state.GetLastMessage(frequency.ID)
if m.lastReadMsg != nil && lastMsg != nil && *m.lastReadMsg != *lastMsg {
m.keepLastRead = true
if m.previousLastReadMsg != nil && lastMsg != nil && *m.previousLastReadMsg != *lastMsg {
m.keepPreviousLastRead = true
}
if val, ok := msgs[frequency.ID]; ok {
@@ -881,10 +897,10 @@ func (m *Model) RestoreAfterSwitch() tea.Cmd {
receiverId := state.Data.Signals[m.receiverIndex]
log.Println("Restoring signal:", receiverId)
m.lastReadMsg = state.State.LastReadMessages[receiverId]
m.previousLastReadMsg = state.State.LastReadMessages[receiverId]
lastMsg := state.GetLastMessage(receiverId)
if m.lastReadMsg != nil && lastMsg != nil && *m.lastReadMsg != *lastMsg {
m.keepLastRead = true
if m.previousLastReadMsg != nil && lastMsg != nil && *m.previousLastReadMsg != *lastMsg {
m.keepPreviousLastRead = true
}
if val, ok := msgs[receiverId]; ok {
@@ -1025,8 +1041,8 @@ func (m *Model) renderMessages(screenHeight int) string {
group := []data.Message{}
lastReadId := state.State.LastReadMessages[*id]
if m.keepLastRead {
lastReadId = m.lastReadMsg
if m.keepPreviousLastRead {
lastReadId = m.previousLastReadMsg
}
last := snowflake.ID(0)

View File

@@ -729,7 +729,7 @@ func calculateNotifications() {
}
pings := getSignalNotification(signal)
state.State.Notifications[signal] = pings
state.State.LocalNotifications[signal] = pings
}
for networkId := range state.State.Networks {
@@ -740,9 +740,9 @@ func calculateNotifications() {
pings, hasNotif := getFrequencyNotification(networkId, frequency.ID)
if hasNotif {
state.State.Notifications[frequency.ID] = pings
state.State.LocalNotifications[frequency.ID] = pings
} else {
delete(state.State.Notifications, frequency.ID)
delete(state.State.LocalNotifications, frequency.ID)
}
}
}

View File

@@ -114,7 +114,7 @@ func (m Model) View() string {
}
notif := ""
pings, hasNotif := state.State.Notifications[frequency.ID]
pings, hasNotif := state.MergedNotification(frequency.ID)
if hasNotif {
notifSymbol := lipgloss.NewStyle().
Foreground(colors.White).Render(notifSymbol)

View File

@@ -45,7 +45,8 @@ func (m Model) View() string {
pings := 0
for _, signal := range state.Data.Signals {
pings += state.State.Notifications[signal]
p, _ := state.MergedNotification(signal)
pings += p
}
var signalsIcon lipgloss.Style
if pings == 0 {
@@ -86,7 +87,7 @@ func (m Model) View() string {
pings, ok := 0, false
frequencies := state.State.Frequencies[networkId]
for _, frequency := range frequencies {
fpings, fok := state.State.Notifications[frequency.ID]
fpings, fok := state.MergedNotification(frequency.ID)
pings += fpings
ok = ok || fok
}

View File

@@ -84,7 +84,7 @@ func (m Model) View() string {
}
notif := ""
pings := state.State.Notifications[signal]
pings, _ := state.MergedNotification(signal)
if pings != 0 {
notif = notifs[min(pings, 10)-1]
notif = lipgloss.NewStyle().Inline(true).

View File

@@ -35,23 +35,25 @@ type state struct {
BlockedUsers map[snowflake.ID]struct{} // key is user id
BlockingUsers map[snowflake.ID]struct{} // key is user id
LastReadMessages map[snowflake.ID]*snowflake.ID // key is frequency id or receiver id
Notifications map[snowflake.ID]int // key is frequency id or receiver id
LastReadMessages map[snowflake.ID]*snowflake.ID // key is frequency id or receiver id
RemoteNotifications map[snowflake.ID]int // key is frequency id or receiver id
LocalNotifications map[snowflake.ID]int // key is frequency id or receiver id
}
var State state = state{
ChatState: map[snowflake.ID]ChatState{},
LastFrequency: map[snowflake.ID]snowflake.ID{},
Messages: map[snowflake.ID]*btree.BTreeG[data.Message]{},
Networks: map[snowflake.ID]data.Network{},
Frequencies: map[snowflake.ID][]data.Frequency{},
Members: map[snowflake.ID]map[snowflake.ID]data.Member{},
Users: map[snowflake.ID]data.User{},
TrustedUsers: map[snowflake.ID]ed25519.PublicKey{},
BlockedUsers: map[snowflake.ID]struct{}{},
BlockingUsers: map[snowflake.ID]struct{}{},
LastReadMessages: map[snowflake.ID]*snowflake.ID{},
Notifications: map[snowflake.ID]int{},
ChatState: map[snowflake.ID]ChatState{},
LastFrequency: map[snowflake.ID]snowflake.ID{},
Messages: map[snowflake.ID]*btree.BTreeG[data.Message]{},
Networks: map[snowflake.ID]data.Network{},
Frequencies: map[snowflake.ID][]data.Frequency{},
Members: map[snowflake.ID]map[snowflake.ID]data.Member{},
Users: map[snowflake.ID]data.User{},
TrustedUsers: map[snowflake.ID]ed25519.PublicKey{},
BlockedUsers: map[snowflake.ID]struct{}{},
BlockingUsers: map[snowflake.ID]struct{}{},
LastReadMessages: map[snowflake.ID]*snowflake.ID{},
RemoteNotifications: map[snowflake.ID]int{},
LocalNotifications: map[snowflake.ID]int{},
}
type UserData struct {
@@ -292,15 +294,13 @@ func UpdateNotifications(info *packet.NotificationsInfo) []snowflake.ID {
ping := info.Pings[i]
if ping != nil {
// log.Println(source, *ping)
State.Notifications[source] = int(*ping)
State.RemoteNotifications[source] = int(*ping)
if !IsFrequency(source) && !slices.Contains(Data.Signals, source) {
signals = append(signals, source)
// log.Println("Signals:", Data.Signals)
}
} else {
// log.Println(source, "deleted")
delete(State.Notifications, info.Source[i])
delete(State.RemoteNotifications, info.Source[i])
}
}
@@ -309,10 +309,10 @@ func UpdateNotifications(info *packet.NotificationsInfo) []snowflake.ID {
func SendFinalData() {
data := JsonUserData()
ch1 := gateway.SendAsync(&packet.SetUserData{
Data: &data,
User: nil,
})
// ch1 := gateway.SendAsync(&packet.SetUserData{
// Data: &data,
// User: nil,
// })
sources := []snowflake.ID{}
@@ -332,6 +332,10 @@ func SendFinalData() {
}
}
ch1 := gateway.SendAsync(&packet.SetUserData{
Data: &data,
User: nil,
})
ch2 := gateway.SendAsync(&packet.SetLastReadMessages{
Source: sources,
LastRead: lastReads,
@@ -340,15 +344,55 @@ func SendFinalData() {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
select {
case <-ctx.Done():
case <-ch1:
}
done := make(chan struct{})
go func() {
<-ch1
<-ch2
close(done)
}()
select {
case <-ctx.Done():
case <-ch2:
log.Println("Timeout while waiting for final writes to complete")
case <-done:
log.Println("All final writes completed successfully")
}
// HACK: Give a small grace period for the writes to be processed
// Tweak this value as needed
time.Sleep(20 * time.Millisecond)
// TODO:
// I think the issue is that it's random which of these 2 requests goes
// first (bcz it only does the first request after the client disconnects)
// I could fix it server side but eh maybe not
// this should instead use a method that blocks until a response was
// received, which may need a new gateway method to do that as currently
// it just always sends to the prograg
// If this is tedious enough it might be worth it to just do it on the
// server side
// Also later on I should probably remove the calculate notifs in core
// it can be replaced with just diff-ing incoming notifs
// this will most likely work fine although there are some issues with
// scopes like becoming an admin/no longer being admin or gaining
// or losing access to frequencies and of course msg deletions
// But it's probably the right approach (also WAYYYYYY faster)
// Then there is also the issue of when switching to a frequency
// not yet receiving the history so it says "no keep" bcz it's not loaded
// but with history it would've said "yes keep" so the solutin would
// be to rework it quite a bit to make it stateless or smthing
// Then that should be most issues when it comes to notifications
// just need to make sure local/remote notifs reset properly when
// reaching the bottom
// log.Println("BLOCKING...")
// <-ctx.Done()
// log.Println("CTX DANZO")
// TODO: remove this before release
// assert.NoError(ctx.Err(), "context has ran out of time!")
}
func UpdateBlockedUsers(info *packet.BlockInfo) {
@@ -375,3 +419,9 @@ func UpdateUsersInfo(info *packet.UsersInfo) {
State.Users[user.ID] = user
}
}
func MergedNotification(id snowflake.ID) (pings int, ok bool) {
remotePings, remoteOk := State.RemoteNotifications[id]
localPings, localOk := State.LocalNotifications[id]
return remotePings + localPings, remoteOk || localOk
}

View File

@@ -216,6 +216,11 @@ func (server *server) handleConnection(conn net.Conn) {
}
log.Println(addr, "sending packet:", packet)
if _, err := packet.Into(conn); err != nil {
// TODO: probably should add this to prevent the
// "use of closed connection" error, as it's intended to happen
// if !errors.Is(err, net.ErrClosed) {
// log.Println(addr, err)
// }
log.Println(addr, err)
return
}
@@ -384,6 +389,8 @@ func processRequest(ctx context.Context, sess *session.Session, request packet.P
response = &packet.Error{Error: "use of disallowed packet type for request"}
}
// TODO: Isn't this weird? shouldn't it always be packet.ErrorPacket type?
// rather than copying the pkt type of the request?
if response, ok := response.(*packet.Error); ok {
response.PktType = request.Type()
}