|
|
|
|
@ -6,11 +6,14 @@ import (
|
|
|
|
|
"crypto/sha512"
|
|
|
|
|
"encoding/binary"
|
|
|
|
|
"fmt"
|
|
|
|
|
"strings"
|
|
|
|
|
"reflect"
|
|
|
|
|
"sync/atomic"
|
|
|
|
|
"time"
|
|
|
|
|
"context"
|
|
|
|
|
|
|
|
|
|
badger "github.com/dgraph-io/badger/v3"
|
|
|
|
|
_ "modernc.org/sqlite"
|
|
|
|
|
"database/sql"
|
|
|
|
|
"github.com/google/uuid"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
@ -94,28 +97,31 @@ type PendingACLSignal struct {
|
|
|
|
|
// Default message channel size for nodes
|
|
|
|
|
// Nodes represent a group of extensions that can be collectively addressed
|
|
|
|
|
type Node struct {
|
|
|
|
|
// Set at creation time, cannot be changed
|
|
|
|
|
Key ed25519.PrivateKey `gv:"key"`
|
|
|
|
|
ID NodeID
|
|
|
|
|
Type NodeType `gv:"type"`
|
|
|
|
|
// TODO: move each extension to it's own db key, and extend changes to notify which extension was changed
|
|
|
|
|
Extensions map[ExtType]Extension
|
|
|
|
|
|
|
|
|
|
Policies []Policy `gv:"policies"`
|
|
|
|
|
|
|
|
|
|
// Keys set at time of creation, cannot be changed
|
|
|
|
|
Extensions map[ExtType]Extension
|
|
|
|
|
|
|
|
|
|
// Not serialized
|
|
|
|
|
PendingACLs map[uuid.UUID]PendingACL `gv:"pending_acls"`
|
|
|
|
|
PendingACLSignals map[uuid.UUID]PendingACLSignal `gv:"pending_signal"`
|
|
|
|
|
|
|
|
|
|
// Channel for this node to receive messages from the Context
|
|
|
|
|
MsgChan chan *Message
|
|
|
|
|
// Size of MsgChan
|
|
|
|
|
// Size of MsgChan, serialized
|
|
|
|
|
BufferSize uint32 `gv:"buffer_size"`
|
|
|
|
|
// Channel for this node to process delayed signals
|
|
|
|
|
TimeoutChan <-chan time.Time
|
|
|
|
|
|
|
|
|
|
Active atomic.Bool
|
|
|
|
|
|
|
|
|
|
// TODO: enhance WriteNode to write SignalQueue to a different key, and use writeSignalQueue to decide whether or not to update it
|
|
|
|
|
writeSignalQueue bool
|
|
|
|
|
// Written to a different table
|
|
|
|
|
AddedQueuedSignals []QueuedSignal
|
|
|
|
|
RemovedQueuedSignals []uuid.UUID
|
|
|
|
|
SignalQueue []QueuedSignal
|
|
|
|
|
NextSignal *QueuedSignal
|
|
|
|
|
}
|
|
|
|
|
@ -203,29 +209,11 @@ func (node *Node) QueueTimeout(reason WaitReason, dest NodeID, signal Signal, ti
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (node *Node) QueueSignal(time time.Time, signal Signal) {
|
|
|
|
|
node.SignalQueue = append(node.SignalQueue, QueuedSignal{signal, time})
|
|
|
|
|
node.NextSignal, node.TimeoutChan = SoonestSignal(node.SignalQueue)
|
|
|
|
|
node.writeSignalQueue = true
|
|
|
|
|
node.AddedQueuedSignals = append(node.AddedQueuedSignals, QueuedSignal{signal, time})
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (node *Node) DequeueSignal(id uuid.UUID) error {
|
|
|
|
|
idx := -1
|
|
|
|
|
for i, q := range(node.SignalQueue) {
|
|
|
|
|
if q.Signal.ID() == id {
|
|
|
|
|
idx = i
|
|
|
|
|
break
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
if idx == -1 {
|
|
|
|
|
return fmt.Errorf("%s is not in SignalQueue", id)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
node.SignalQueue[idx] = node.SignalQueue[len(node.SignalQueue)-1]
|
|
|
|
|
node.SignalQueue = node.SignalQueue[:len(node.SignalQueue)-1]
|
|
|
|
|
node.NextSignal, node.TimeoutChan = SoonestSignal(node.SignalQueue)
|
|
|
|
|
node.writeSignalQueue = true
|
|
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
func (node *Node) DequeueSignal(id uuid.UUID) {
|
|
|
|
|
node.RemovedQueuedSignals = append(node.RemovedQueuedSignals, id)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func SoonestSignal(signals []QueuedSignal) (*QueuedSignal, <-chan time.Time) {
|
|
|
|
|
@ -735,58 +723,93 @@ func WriteNodeExtList(ctx *Context, node *Node) error {
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func WriteNodeInitial(ctx *Context, node *Node) error {
|
|
|
|
|
ctx.Log.Logf("db", "Writing node initial data %s - %+v", node.ID, node)
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func GetExtField(ctx *Context, node *Node, ext_type ExtType, field string) (interface{}, error) {
|
|
|
|
|
ext_info, exists := ctx.Extensions[ext_type]
|
|
|
|
|
if exists == false {
|
|
|
|
|
return nil, fmt.Errorf("0x%x is not a know extension type", ext_type)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ext, has_ext := node.Extensions[ext_type]
|
|
|
|
|
if has_ext == false {
|
|
|
|
|
return nil, fmt.Errorf("0x%x is not an extension of %s", ext_type, node.ID)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
field_idx, has_field := ext_info.FieldMap[field]
|
|
|
|
|
if has_field == false {
|
|
|
|
|
return nil, fmt.Errorf("%s is not a field in %+v", field, ext_info.Type)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return reflect.ValueOf(ext).FieldByIndex(field_idx).Interface(), nil
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func WriteNodeChanges(ctx *Context, node *Node, changes Changes) error {
|
|
|
|
|
ctx.Log.Logf("db", "Writing changes for %s - %+v", node.ID, changes)
|
|
|
|
|
|
|
|
|
|
ext_serialized := map[ExtType]SerializedValue{}
|
|
|
|
|
for ext_type := range(changes) {
|
|
|
|
|
ext, ext_exists := node.Extensions[ext_type]
|
|
|
|
|
if ext_exists == false {
|
|
|
|
|
ctx.Log.Logf("db", "extension 0x%x does not exist for %s", ext_type, node.ID)
|
|
|
|
|
} else {
|
|
|
|
|
serialized_ext, err := SerializeAny(ctx, ext)
|
|
|
|
|
tx, err := ctx.DB.BeginTx(context.TODO(), &sql.TxOptions{
|
|
|
|
|
Isolation: sql.LevelSerializable,
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
ext_serialized[ext_type] = serialized_ext
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var sq_serialized *SerializedValue = nil
|
|
|
|
|
if node.writeSignalQueue == true {
|
|
|
|
|
node.writeSignalQueue = false
|
|
|
|
|
ser, err := SerializeAny(ctx, node.SignalQueue)
|
|
|
|
|
// Remove removed signals from DB, and from in memory list
|
|
|
|
|
new_signal_queue := []QueuedSignal{}
|
|
|
|
|
for _, signal_id := range(node.RemovedQueuedSignals) {
|
|
|
|
|
_, err = tx.Exec("DELETE FROM queued_signals WHERE node = ? AND signal_ID = ?", node.ID, signal_id)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
sq_serialized = &ser
|
|
|
|
|
|
|
|
|
|
idx := -1
|
|
|
|
|
for i, signal := range(node.SignalQueue) {
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
node_serialized, err := SerializeAny(ctx, node)
|
|
|
|
|
node.RemovedQueuedSignals = nil
|
|
|
|
|
|
|
|
|
|
for _, signal := range(node.AddedQueuedSignals) {
|
|
|
|
|
signal_serialized, err := SerializeAny(ctx, signal.Signal)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
id_bytes, err := node.ID.MarshalBinary()
|
|
|
|
|
return ctx.DB.Update(func(txn *badger.Txn) error {
|
|
|
|
|
err := txn.Set(id_bytes, node_serialized.Data)
|
|
|
|
|
_, err = tx.Exec("UPDATE queued_signals ADD node = ?, signal_id = ?, signal_data = ?, time = ?", node.ID, signal.ID(), signal_serialized.Data, signal.Time)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
if sq_serialized != nil {
|
|
|
|
|
err := txn.Set(append(id_bytes, signal_queue_suffix...), sq_serialized.Data)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
node.SignalQueue = append(node.SignalQueue, node.AddedQueuedSignals...)
|
|
|
|
|
node.AddedQueuedSignals = nil
|
|
|
|
|
|
|
|
|
|
for ext_type, changes := range(changes) {
|
|
|
|
|
change_strings := make([]string, len(changes))
|
|
|
|
|
change_values := make([]interface{}, len(changes))
|
|
|
|
|
|
|
|
|
|
for i, changed_value := range(changes) {
|
|
|
|
|
change_strings[i] = fmt.Sprintf("%s = ?", changed_value)
|
|
|
|
|
var err error
|
|
|
|
|
change_values[i], err = GetExtField(ctx, node, ext_type, changed_value)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
for ext_type, data := range(ext_serialized) {
|
|
|
|
|
err := txn.Set(append(id_bytes, ExtTypeSuffix(ext_type)...), data.Data)
|
|
|
|
|
|
|
|
|
|
set_string := strings.Join(change_strings, ", ")
|
|
|
|
|
_, err = tx.Exec(fmt.Sprintf("UPDATE extension_%x SET %s WHERE node = ?", ext_type, set_string), append(change_values, node.ID))
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
return nil
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
return tx.Commit()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func LoadNode(ctx *Context, id NodeID) (*Node, error) {
|
|
|
|
|
|
