Remember multiple recent transaction IDs for deduplication

appservice/appservice.go

@@ -46,6 +46,7 @@ func Create() *AppService {
 		StateStore: NewBasicStateStore(),
 		Router:     mux.NewRouter(),
 		UserAgent:  mautrix.DefaultUserAgent,
+		txnIDC:     NewTransactionIDCache(128),
 		Live:       true,
 		Ready:      false,
 	}
@@ -90,7 +91,7 @@ type AppService struct {
 	Registration *Registration    `yaml:"-"`
 	Log          maulogger.Logger `yaml:"-"`
 
-	lastProcessedTransaction string
+	txnIDC *TransactionIDCache
 
 	Events       chan *event.Event         `yaml:"-"`
 	DeviceLists  chan *mautrix.DeviceLists `yaml:"-"`

appservice/http.go

@@ -116,7 +116,7 @@ func (as *AppService) PutTransaction(w http.ResponseWriter, r *http.Request) {
 		}.Write(w)
 		return
 	}
-	if as.lastProcessedTransaction == txnID {
+	if as.txnIDC.IsProcessed(txnID) {
 		// Duplicate transaction ID: no-op
 		WriteBlankOK(w)
 		return
@@ -134,8 +134,8 @@ func (as *AppService) PutTransaction(w http.ResponseWriter, r *http.Request) {
 	} else {
 		as.handleTransaction(&txn)
 		WriteBlankOK(w)
+		as.txnIDC.MarkProcessed(txnID)
 	}
-	as.lastProcessedTransaction = txnID
 }
 
 func (as *AppService) handleTransaction(txn *Transaction) {

appservice/txnid.go

@@ -0,0 +1,43 @@
+// Copyright (c) 2021 Tulir Asokan
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+package appservice
+
+import "sync"
+
+type TransactionIDCache struct {
+	array    []string
+	arrayPtr int
+	hash     map[string]struct{}
+	lock     sync.RWMutex
+}
+
+func NewTransactionIDCache(size int) *TransactionIDCache {
+	return &TransactionIDCache{
+		array: make([]string, size),
+		hash:  make(map[string]struct{}),
+	}
+}
+
+func (txnIDC *TransactionIDCache) IsProcessed(txnID string) bool {
+	txnIDC.lock.RLock()
+	_, exists := txnIDC.hash[txnID]
+	txnIDC.lock.RUnlock()
+	return exists
+}
+
+func (txnIDC *TransactionIDCache) MarkProcessed(txnID string) {
+	txnIDC.lock.Lock()
+	txnIDC.hash[txnID] = struct{}{}
+	if txnIDC.array[txnIDC.arrayPtr] != "" {
+		for i := 0; i < len(txnIDC.array)/8; i++ {
+			delete(txnIDC.hash, txnIDC.array[txnIDC.arrayPtr+i])
+			txnIDC.array[txnIDC.arrayPtr+i] = ""
+		}
+	}
+	txnIDC.array[txnIDC.arrayPtr] = txnID
+	txnIDC.lock.Unlock()
+}

version.go

@@ -1,5 +1,5 @@
 package mautrix
 
-const Version = "v0.9.21"
+const Version = "v0.9.22"
 
 var DefaultUserAgent = "mautrix-go/" + Version