Use TransactionWriter to reduce database lock issues on SQLite
@@ -36,6 +36,8 @@ type Database struct {
queueJSONStatements
sqlutil.PartitionOffsetStatements
db *sql.DB
+ queuePDUsWriter *sqlutil.TransactionWriter
+ queueJSONWriter *sqlutil.TransactionWriter
}
// NewDatabase opens a new database
@@ -74,6 +76,9 @@ func (d *Database) prepare() error {
return err
}
+ d.queuePDUsWriter = sqlutil.NewTransactionWriter()
+ d.queueJSONWriter = sqlutil.NewTransactionWriter()
+
return d.PartitionOffsetStatements.Prepare(d.db, "federationsender")
}
@@ -145,12 +150,16 @@ func (d *Database) GetJoinedHosts(
// metadata entries.
func (d *Database) StoreJSON(
ctx context.Context, js string,
-) (int64, error) {
- nid, err := d.insertQueueJSON(ctx, nil, js)
- if err != nil {
- return 0, fmt.Errorf("d.insertQueueJSON: %w", err)
+) (nid int64, err error) {
+ err = d.queueJSONWriter.Do(d.db, func(txn *sql.Tx) error {
+ n, e := d.insertQueueJSON(ctx, nil, js)
+ if e != nil {
+ return fmt.Errorf("d.insertQueueJSON: %w", e)
}
- return nid, nil
+ nid = n
+ return nil
+ })
+ return
}
// AssociatePDUWithDestination creates an association that the
@@ -162,7 +171,7 @@ func (d *Database) AssociatePDUWithDestination(
serverName gomatrixserverlib.ServerName,
nids []int64,
) error {
- return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+ return d.queuePDUsWriter.Do(d.db, func(txn *sql.Tx) error {
for _, nid := range nids {
if err := d.insertQueuePDU(
ctx, // context
@@ -230,18 +239,18 @@ func (d *Database) CleanTransactionPDUs(
serverName gomatrixserverlib.ServerName,
transactionID gomatrixserverlib.TransactionID,
) error {
- return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
- nids, err := d.selectQueuePDUs(ctx, txn, serverName, transactionID, 50)
+ var err error
+ var nids []int64
+ var deleteNIDs []int64
+ if err = d.queuePDUsWriter.Do(d.db, func(txn *sql.Tx) error {
+ nids, err = d.selectQueuePDUs(ctx, txn, serverName, transactionID, 50)
if err != nil {
return fmt.Errorf("d.selectQueuePDUs: %w", err)
}
-
if err = d.deleteQueueTransaction(ctx, txn, serverName, transactionID); err != nil {
return fmt.Errorf("d.deleteQueueTransaction: %w", err)
}
-
var count int64
- var deleteNIDs []int64
for _, nid := range nids {
count, err = d.selectQueueReferenceJSONCount(ctx, txn, nid)
if err != nil {
@@ -251,15 +260,19 @@ func (d *Database) CleanTransactionPDUs(
deleteNIDs = append(deleteNIDs, nid)
}
}
-
+ return nil
+ }); err != nil {
+ return err
+ }
+ err = d.queueJSONWriter.Do(d.db, func(txn *sql.Tx) error {
if len(deleteNIDs) > 0 {
if err = d.deleteQueueJSON(ctx, txn, deleteNIDs); err != nil {
return fmt.Errorf("d.deleteQueueJSON: %w", err)
}
}
-
return nil
})
+ return err
}
// GetPendingPDUCount returns the number of PDUs waiting to be
