feat(backend): set seek pointer at the end of batch for avoiding memory leaks

backend/cmd/assets/main.go

@@ -63,6 +63,7 @@ func main() {
 					}
 				}
 			}
+			iter.Close()
 		},
 		true,
 		cfg.MessageSizeLimit,

backend/cmd/db/main.go

@@ -112,6 +112,7 @@ func main() {
 				}
 			})
 		}
+		iter.Close()
 	}
 
 	// Init consumer

backend/cmd/ender/main.go

@@ -54,6 +54,7 @@ func main() {
 				statsLogger.Collect(sessionID, meta)
 				sessions.UpdateSession(sessionID, meta.Timestamp, iter.Message().Meta().Timestamp)
 			}
+			iter.Close()
 		},
 		false,
 		cfg.MessageSizeLimit,

backend/cmd/heuristics/main.go

@@ -64,6 +64,7 @@ func main() {
 				lastMessageID = msg.Meta().Index
 				builderMap.HandleMessage(sessionID, msg, iter.Message().Meta().Index)
 			}
+			iter.Close()
 		},
 		false,
 		cfg.MessageSizeLimit,

backend/cmd/sink/main.go

@@ -107,6 +107,7 @@ func main() {
 				messageSize.Record(context.Background(), float64(len(data)))
 				savedMessages.Add(context.Background(), 1)
 			}
+			iter.Close()
 		},
 		false,
 		cfg.MessageSizeLimit,

backend/pkg/messages/batch.go

@@ -11,6 +11,7 @@ type Iterator interface {
 	Next() bool       // Return true if we have next message
 	Type() int        // Return type of the next message
 	Message() Message // Return raw or decoded message
+	Close()
 }
 
 type iteratorImpl struct {
@@ -184,6 +185,13 @@ func (i *iteratorImpl) Message() Message {
 	return i.msg
 }
 
+func (i *iteratorImpl) Close() {
+	_, err := i.data.Seek(0, io.SeekEnd)
+	if err != nil {
+		log.Printf("can't set seek pointer at the end: %s", err)
+	}
+}
+
 func messageHasSize(msgType uint64) bool {
 	return !(msgType == 80 || msgType == 81 || msgType == 82)
 }