implement transactions: Add Enhanced Authorization and Protocol Improvements

.github/workflows/go.yml

@@ -7,7 +7,6 @@ on:
   push:
     branches: [ "main" ]
   pull_request:
-    branches: [ "main" ]
 
 
 permissions:

atlas/authorization.go

@@ -21,14 +21,17 @@ package atlas
 import (
 	"go.uber.org/zap"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 	"zombiezen.com/go/sqlite"
 )
 
 type Authorizer struct {
-	boundTime time.Duration
-	mu        sync.RWMutex
+	boundTime     time.Duration
+	mu            sync.RWMutex
+	LastTables    map[string]struct{}
+	ForceReadonly bool
 }
 
 var writeOps []sqlite.OpType = []sqlite.OpType{
@@ -74,6 +77,7 @@ func (a *Authorizer) Reset() {
 	defer a.mu.Unlock()
 
 	a.boundTime = 0
+	a.LastTables = make(map[string]struct{})
 }
 
 func (a *Authorizer) isJournalModeChange(action sqlite.Action) bool {
@@ -91,6 +95,15 @@ func (a *Authorizer) isAtlasChange(action sqlite.Action) bool {
 func (a *Authorizer) Authorize(action sqlite.Action) sqlite.AuthResult {
 	Logger.Info("Auth", zap.Any("action", action), zap.String("table", action.Table()))
 
+	if a.ForceReadonly && a.isWrite(action) {
+		return sqlite.AuthResultDeny
+	}
+
+	if action.Table() != "" && action.Database() != "" && action.Table() != "sqlite_master" {
+		name := strings.ToUpper(action.Database()) + "." + strings.ToUpper(action.Table())
+		a.LastTables[name] = struct{}{}
+	}
+
 	if a.isJournalModeChange(action) || a.isAtlasChange(action) {
 		return sqlite.AuthResultDeny
 	}

atlas/bootstrap/client.go

@@ -357,6 +357,9 @@ SET address = :address, port = :port, region = :region, active = 1
 		return err
 	}
 
+	// add the node to the quorum manager
+	qm.AddNode(node)
+
 	atlas.CurrentOptions.ServerId = node.Id
 
 	return err

atlas/bootstrap/migrations.sql

@@ -76,7 +76,7 @@ create table tables
 );
 /* Insert the nodes table as a tracked table */
 insert into tables
-values ('atlas.nodes', 'global', null, current_timestamp, 0, '', '');
+values ('ATLAS.NODES', 'global', null, current_timestamp, 0, '', '');
 
 /* Migrations are defined as rows on this table.
    They are ultimately numbered by the version and batch_part columns.

atlas/consensus/majority-quorum.go

@@ -25,11 +25,21 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"sync"
+	"sync/atomic"
 )
 
 type majorityQuorum struct {
-	q1 []*QuorumNode
-	q2 []*QuorumNode
+	q1                   []*QuorumNode
+	q2                   []*QuorumNode
+	nextMigrationVersion atomic.Int64
+}
+
+func (m *majorityQuorum) SetNextMigrationVersion(version int64) {
+	m.nextMigrationVersion.Store(version)
+}
+
+func (m *majorityQuorum) GetNextMigrationVersion() int64 {
+	return m.nextMigrationVersion.Load()
 }
 
 func (m *majorityQuorum) Gossip(ctx context.Context, in *GossipMigration, opts ...grpc.CallOption) (*emptypb.Empty, error) {

atlas/consensus/quorum.go

@@ -73,6 +73,8 @@ func (q *defaultQuorumManager) AddNode(node *Node) {
 
 type Quorum interface {
 	ConsensusClient
+	SetNextMigrationVersion(version int64)
+	GetNextMigrationVersion() int64
 }
 
 type QuorumNode struct {

atlas/consensus/server.go

@@ -34,7 +34,7 @@ import (
 	"zombiezen.com/go/sqlite"
 )
 
-const NodeTable = "atlas.nodes"
+const NodeTable = "ATLAS.NODES"
 
 type Server struct {
 	UnimplementedConsensusServer
@@ -53,6 +53,9 @@ func (s *Server) StealTableOwnership(ctx context.Context, req *StealTableOwnersh
 	}()
 
 	_, err = atlas.ExecuteSQL(ctx, "BEGIN IMMEDIATE", conn, false)
+	if err != nil {
+		return nil, err
+	}
 
 	tr := GetDefaultTableRepository(ctx, conn)
 	existingTable, err := tr.GetTable(req.GetTable().GetName())
@@ -303,7 +306,7 @@ func (s *Server) AcceptMigration(ctx context.Context, req *WriteMigrationRequest
 	}()
 
 	commitConn := conn
-	if !strings.HasPrefix(req.GetMigration().GetVersion().GetTableName(), "atlas.") {
+	if !strings.HasPrefix(req.GetMigration().GetVersion().GetTableName(), "ATLAS.") {
 		commitConn, err = atlas.Pool.Take(ctx)
 		if err != nil {
 			return nil, err
@@ -388,7 +391,7 @@ func (s *Server) applyMigration(migrations []*Migration, commitConn *sqlite.Conn
 	return nil
 }
 
-func constructCurrentNode() *Node {
+func ConstructCurrentNode() *Node {
 	return &Node{
 		Id:      atlas.CurrentOptions.ServerId,
 		Address: atlas.CurrentOptions.AdvertiseAddress,
@@ -515,7 +518,7 @@ VALUES (:id, :address, :port, :region, 1, current_timestamp, 0)`, conn, false, a
 	}
 
 	mreq := &WriteMigrationRequest{
-		Sender:    constructCurrentNode(),
+		Sender:    ConstructCurrentNode(),
 		Migration: migration,
 	}
 
@@ -635,7 +638,7 @@ func SendGossip(ctx context.Context, req *GossipMigration, conn *sqlite.Conn) er
 		Table:             req.GetTable(),
 		PreviousMigration: req.GetPreviousMigration(),
 		Ttl:               req.GetTtl() - 1,
-		Sender:            constructCurrentNode(),
+		Sender:            ConstructCurrentNode(),
 	}
 
 	wg := sync.WaitGroup{}

atlas/ownership.go

@@ -42,12 +42,51 @@ type TableOwnerships struct {
 	mu            sync.RWMutex
 	subscriptions map[string][]chan<- TableOwnershipChange
 	commitTimes   map[string]time.Time
+	holds         map[string]int
 }
 
 var Ownership = &TableOwnerships{
 	own:           map[string]struct{}{},
 	subscriptions: map[string][]chan<- TableOwnershipChange{},
 	commitTimes:   map[string]time.Time{},
+	holds:         map[string]int{},
+}
+
+func (t *TableOwnerships) Hold(table string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	if c, ok := t.holds[table]; ok {
+		t.holds[table] = c + 1
+		return
+	}
+
+	t.holds[table] = 1
+}
+
+func (t *TableOwnerships) Release(table string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	if c, ok := t.holds[table]; ok {
+		if c == 1 {
+			delete(t.holds, table)
+		} else {
+			t.holds[table] = c - 1
+		}
+	}
+
+	if c, ok := t.holds[table]; ok && c <= 0 {
+		delete(t.holds, table)
+	}
+}
+
+func (t *TableOwnerships) IsHeld(table string) bool {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	_, ok := t.holds[table]
+	return ok
 }
 
 func (t *TableOwnerships) Add(table string, version int64) {

atlas/pool.go

@@ -23,7 +23,6 @@ import (
 	_ "embed"
 	"runtime"
 	"strings"
-	"sync"
 	"zombiezen.com/go/sqlite"
 	"zombiezen.com/go/sqlite/sqlitemigration"
 )
@@ -34,27 +33,18 @@ var migrations string
 var Pool *sqlitemigration.Pool
 var MigrationsPool *sqlitemigration.Pool
 
-var authorizers = map[*sqlite.Conn]*Authorizer{}
-
 // CreatePool creates a new connection pool for the database and the migrations database.
 func CreatePool(options *Options) {
 	if Pool != nil {
 		return
 	}
 
-	authorizers = make(map[*sqlite.Conn]*Authorizer)
-	amu := sync.Mutex{}
-
 	Pool = sqlitemigration.NewPool(options.DbFilename, sqlitemigration.Schema{}, sqlitemigration.Options{
 		Flags:    sqlite.OpenReadWrite | sqlite.OpenCreate | sqlite.OpenWAL,
 		PoolSize: runtime.NumCPU() * 2,
 		PrepareConn: func(conn *sqlite.Conn) (err error) {
 			auth := &Authorizer{}
 
-			amu.Lock()
-			authorizers[conn] = auth
-			amu.Unlock()
-
 			err = conn.SetAuthorizer(auth)
 			if err != nil {
 				return