Browse Source

Replace legacy message store with sqlite3 message store

sqliteserver
Sarah Jamie Lewis 5 months ago
parent
commit
32d81046cf
  1. 5
      .gitignore
  2. 1
      go.mod
  3. 8
      go.sum
  4. 9
      server.go
  5. 184
      storage/message_store.go
  6. 58
      storage/message_store_test.go

5
.gitignore

@ -3,4 +3,7 @@ serverConfig.json
coverage.out
tordir
.idea/
messages
messages
cwtch.messages
serverMonitorReport.txt
testcwtchmessages.db

1
go.mod

@ -8,6 +8,7 @@ require (
git.openprivacy.ca/openprivacy/connectivity v1.4.3
git.openprivacy.ca/openprivacy/log v1.0.2
github.com/gtank/ristretto255 v0.1.2
github.com/mattn/go-sqlite3 v1.14.7
github.com/struCoder/pidusage v0.1.3
golang.org/x/crypto v0.0.0-20201012173705-84dcc777aaee
)

8
go.sum

@ -16,16 +16,21 @@ github.com/gtank/merlin v0.1.1 h1:eQ90iG7K9pOhtereWsmyRJ6RAwcP4tHTDBHXNg+u5is=
github.com/gtank/merlin v0.1.1/go.mod h1:T86dnYJhcGOh5BjZFCJWTDeTK7XW8uE+E21Cy/bIQ+s=
github.com/gtank/ristretto255 v0.1.2 h1:JEqUCPA1NvLq5DwYtuzigd7ss8fwbYay9fi4/5uMzcc=
github.com/gtank/ristretto255 v0.1.2/go.mod h1:Ph5OpO6c7xKUGROZfWVLiJf9icMDwUeIvY4OmlYW69o=
github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
github.com/kr/pty v1.1.1 h1:VkoXIwSboBpnk99O/KFauAEILuNHv5DVFKZMBN/gUgw=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/mattn/go-sqlite3 v1.14.7 h1:fxWBnXkxfM6sRiuH3bqJ4CfzZojMOLVc0UTsTglEghA=
github.com/mattn/go-sqlite3 v1.14.7/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
github.com/mimoo/StrobeGo v0.0.0-20181016162300-f8f6d4d2b643 h1:hLDRPB66XQT/8+wG9WsDpiCvZf1yKO7sz7scAjSlBa0=
github.com/mimoo/StrobeGo v0.0.0-20181016162300-f8f6d4d2b643/go.mod h1:43+3pMjjKimDBf5Kr4ZFNGbLql1zKkbImw+fZbw3geM=
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@ -48,9 +53,12 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44 h1:Bli41pIlzTzf3KEY06n+xnzK/BESIg2ze4Pgfh/aI8c=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e h1:FDhOuMEY4JVRztM/gsbk+IKUQ8kj74bxZrgw87eMMVc=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

9
server.go

@ -3,9 +3,9 @@ package server
import (
"crypto/ed25519"
"cwtch.im/cwtch/model"
"cwtch.im/cwtch/server/metrics"
"cwtch.im/cwtch/server/storage"
"fmt"
"git.openprivacy.ca/cwtch.im/server/metrics"
"git.openprivacy.ca/cwtch.im/server/storage"
"git.openprivacy.ca/cwtch.im/tapir"
"git.openprivacy.ca/cwtch.im/tapir/applications"
tor2 "git.openprivacy.ca/cwtch.im/tapir/networks/tor"
@ -62,10 +62,9 @@ func (s *Server) Run(acn connectivity.ACN) error {
log.Infof("cwtch server running on cwtch:%s\n", addressIdentity+".onion:")
s.metricsPack.Start(service, s.config.ConfigDir, s.config.ServerReporting.LogMetricsToFile)
ms := new(storage.MessageStore)
err := ms.Init(s.config.ConfigDir, s.config.MaxBufferLines, s.metricsPack.MessageCounter)
ms, err := storage.InitializeSqliteMessageStore("cwtch.messages")
if err != nil {
return err
return fmt.Errorf("could not open database: %v", err)
}
// Needed because we only collect metrics on a per-session basis

184
storage/message_store.go

@ -1,21 +1,12 @@
package storage
import (
"bufio"
"cwtch.im/cwtch/protocol/groups"
"cwtch.im/cwtch/server/metrics"
"encoding/json"
"database/sql"
"encoding/base64"
"fmt"
"git.openprivacy.ca/openprivacy/log"
"os"
"path"
"sync"
)
const (
fileStorePartitions = 10
fileStoreFilename = "cwtch.messages"
directory = "messages"
_ "github.com/mattn/go-sqlite3" // sqlite3 driver
)
// MessageStoreInterface defines an interface to interact with a store of cwtch messages.
@ -24,129 +15,82 @@ type MessageStoreInterface interface {
FetchMessages() []*groups.EncryptedGroupMessage
}
// MessageStore is a file-backed implementation of MessageStoreInterface
type MessageStore struct {
activeLogFile *os.File
filePos int
storeDirectory string
lock sync.Mutex
messages []*groups.EncryptedGroupMessage
messageCounter metrics.Counter
maxBufferLines int
bufferPos int
bufferRotated bool
// SqliteMessageStore is an sqlite3 backed message store
type SqliteMessageStore struct {
database *sql.DB
}
// Close closes the message store and underlying resources.
func (ms *MessageStore) Close() {
ms.lock.Lock()
ms.messages = nil
ms.activeLogFile.Close()
ms.lock.Unlock()
// Close closes the underlying sqlite3 database to further changes
func (s *SqliteMessageStore) Close() {
s.database.Close()
}
func (ms *MessageStore) updateBuffer(gm *groups.EncryptedGroupMessage) {
ms.messages[ms.bufferPos] = gm
ms.bufferPos++
if ms.bufferPos == ms.maxBufferLines {
ms.bufferPos = 0
ms.bufferRotated = true
// AddMessage implements the MessageStoreInterface AddMessage for sqlite message store
func (s *SqliteMessageStore) AddMessage(message groups.EncryptedGroupMessage) {
tx, err := s.database.Begin()
if err != nil {
log.Errorf("%q", err)
return
}
}
func (ms *MessageStore) initAndLoadFiles() error {
ms.activeLogFile = nil
for i := fileStorePartitions - 1; i >= 0; i-- {
ms.filePos = 0
filename := path.Join(ms.storeDirectory, fmt.Sprintf("%s.%d", fileStoreFilename, i))
f, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0600)
if err != nil {
log.Errorf("MessageStore could not open: %v: %v", filename, err)
continue
}
ms.activeLogFile = f
scanner := bufio.NewScanner(f)
for scanner.Scan() {
gms := scanner.Text()
ms.filePos++
gm := &groups.EncryptedGroupMessage{}
err := json.Unmarshal([]byte(gms), gm)
if err == nil {
ms.updateBuffer(gm)
}
}
sqlStmt := `INSERT INTO messages(signature, ciphertext) values (?,?);`
stmt, err := s.database.Prepare(sqlStmt)
if err != nil {
log.Errorf("%q: %s", err, sqlStmt)
return
}
if ms.activeLogFile == nil {
return fmt.Errorf("Could not create log file to write to in %s", ms.storeDirectory)
defer stmt.Close()
_, err = stmt.Exec(base64.StdEncoding.EncodeToString(message.Signature), base64.StdEncoding.EncodeToString(message.Ciphertext))
if err != nil {
log.Errorf("%q: %s\n", err, sqlStmt)
return
}
return nil
tx.Commit()
}
func (ms *MessageStore) updateFile(gm *groups.EncryptedGroupMessage) {
s, err := json.Marshal(gm)
// FetchMessages implements the MessageStoreInterface FetchMessages for sqlite message store
func (s SqliteMessageStore) FetchMessages() []*groups.EncryptedGroupMessage {
rows, err := s.database.Query("SELECT id, signature,ciphertext from messages")
if err != nil {
log.Errorf("Failed to unmarshal group message %v\n", err)
log.Errorf("%v", err)
return nil
}
fmt.Fprintf(ms.activeLogFile, "%s\n", s)
ms.filePos++
if ms.filePos >= ms.maxBufferLines/fileStorePartitions {
ms.rotateFileStore()
defer rows.Close()
var messages []*groups.EncryptedGroupMessage
for rows.Next() {
var id int
var signature string
var ciphertext string
err = rows.Scan(&id, &signature, &ciphertext)
if err != nil {
log.Errorf("Error fetching row %v", err)
}
rawSignature, _ := base64.StdEncoding.DecodeString(signature)
rawCiphertext, _ := base64.StdEncoding.DecodeString(ciphertext)
messages = append(messages, &groups.EncryptedGroupMessage{
Signature: rawSignature,
Ciphertext: rawCiphertext,
})
}
return messages
}
func (ms *MessageStore) rotateFileStore() {
ms.activeLogFile.Close()
os.Remove(path.Join(ms.storeDirectory, fmt.Sprintf("%s.%d", fileStoreFilename, fileStorePartitions-1)))
for i := fileStorePartitions - 2; i >= 0; i-- {
os.Rename(path.Join(ms.storeDirectory, fmt.Sprintf("%s.%d", fileStoreFilename, i)), path.Join(ms.storeDirectory, fmt.Sprintf("%s.%d", fileStoreFilename, i+1)))
}
f, err := os.OpenFile(path.Join(ms.storeDirectory, fmt.Sprintf("%s.%d", fileStoreFilename, 0)), os.O_CREATE|os.O_APPEND|os.O_RDWR, 0600)
// InitializeSqliteMessageStore creates a database `dbfile` with the necessary tables (if it doesn't already exist)
// and returns an open database
func InitializeSqliteMessageStore(dbfile string) (*SqliteMessageStore, error) {
db, err := sql.Open("sqlite3", dbfile)
if err != nil {
log.Errorf("Could not open new message store file in: %s", ms.storeDirectory)
log.Errorf("database %v cannot be created or opened %v", dbfile, err)
return nil, fmt.Errorf("database %v cannot be created or opened: %v", dbfile, err)
}
ms.filePos = 0
ms.activeLogFile = f
}
// Init sets up a MessageStore of size maxBufferLines (# of messages) backed by filename
func (ms *MessageStore) Init(appDirectory string, maxBufferLines int, messageCounter metrics.Counter) error {
ms.storeDirectory = path.Join(appDirectory, directory)
os.Mkdir(ms.storeDirectory, 0700)
ms.bufferPos = 0
ms.maxBufferLines = maxBufferLines
ms.messages = make([]*groups.EncryptedGroupMessage, maxBufferLines)
ms.bufferRotated = false
ms.messageCounter = messageCounter
err := ms.initAndLoadFiles()
return err
}
// FetchMessages returns all messages from the backing file.
func (ms *MessageStore) FetchMessages() (messages []*groups.EncryptedGroupMessage) {
ms.lock.Lock()
if !ms.bufferRotated {
messages = make([]*groups.EncryptedGroupMessage, ms.bufferPos)
copy(messages, ms.messages[0:ms.bufferPos])
} else {
messages = make([]*groups.EncryptedGroupMessage, ms.maxBufferLines)
copy(messages, ms.messages[ms.bufferPos:ms.maxBufferLines])
copy(messages[ms.bufferPos:], ms.messages[0:ms.bufferPos])
sqlStmt := `CREATE TABLE IF NOT EXISTS messages (id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, signature TEXT UNIQUE NOT NULL, ciphertext TEXT NOT NULL);`
_, err = db.Exec(sqlStmt)
if err != nil {
db.Close()
log.Errorf("%q: %s", err, sqlStmt)
return nil, fmt.Errorf("%s: %q", sqlStmt, err)
}
ms.lock.Unlock()
return
}
// AddMessage adds a GroupMessage to the store
func (ms *MessageStore) AddMessage(gm groups.EncryptedGroupMessage) {
ms.messageCounter.Add(1)
ms.lock.Lock()
ms.updateBuffer(&gm)
ms.updateFile(&gm)
ms.lock.Unlock()
log.Infof("Database Initialized")
slms := new(SqliteMessageStore)
slms.database = db
return slms, nil
}

58
storage/message_store_test.go

@ -2,53 +2,35 @@ package storage
import (
"cwtch.im/cwtch/protocol/groups"
"cwtch.im/cwtch/server/metrics"
"git.openprivacy.ca/openprivacy/log"
"os"
"strconv"
"testing"
)
func TestMessageStore(t *testing.T) {
os.Remove("ms.test")
ms := new(MessageStore)
counter := metrics.NewCounter()
ms.Init("./", 1000, counter)
for i := 0; i < 499; i++ {
gm := groups.EncryptedGroupMessage{
Ciphertext: []byte("Hello this is a fairly average length message that we are writing here. " + strconv.Itoa(i)),
}
ms.AddMessage(gm)
}
if counter.Count() != 499 {
t.Errorf("Counter should be at 499 was %v", counter.Count())
}
ms.Close()
ms.Init("./", 1000, counter)
m := ms.FetchMessages()
if len(m) != 499 {
t.Errorf("Should have been 499 was %v", len(m))
os.Remove("../testcwtchmessages.db")
log.SetLevel(log.LevelDebug)
db, err := InitializeSqliteMessageStore("../testcwtchmessages.db")
if err != nil {
t.Fatalf("Error: %v", err)
}
counter.Reset()
db.AddMessage(groups.EncryptedGroupMessage{
Signature: []byte("Hello world 2"),
Ciphertext: []byte("Hello world"),
})
for i := 0; i < 1000; i++ {
gm := groups.EncryptedGroupMessage{
Ciphertext: []byte("Hello this is a fairly average length message that we are writing here. " + strconv.Itoa(i)),
}
ms.AddMessage(gm)
}
db.AddMessage(groups.EncryptedGroupMessage{
Signature: []byte("Hello world 1"),
Ciphertext: []byte("Hello world"),
})
m = ms.FetchMessages()
if len(m) != 1000 {
t.Errorf("Should have been 1000 was %v", len(m))
messages := db.FetchMessages()
for _, message := range messages {
t.Logf("Message: %v", message)
}
ms.Close()
ms.Init("./", 1000, counter)
m = ms.FetchMessages()
if len(m) != 999 {
t.Errorf("Should have been 999 was %v", len(m))
if len(messages) != 2 {
t.Fatalf("Incorrect number of messages returned")
}
ms.Close()
os.RemoveAll("./messages")
db.Close()
}
Loading…
Cancel
Save