Skip to content

feat: Implement WebSocket-based real-time communication for orchestrator hosts with intelligent fallback #330

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sai-kumar-peddireddy merged 5 commits into from
Nov 28, 2025
Merged

feat: Implement WebSocket-based real-time communication for orchestrator hosts with intelligent fallback #330

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
80aa9a9
feat: Implement WebSocket-based real-time communication for orchestra…
sai-kumar-peddireddy Nov 24, 2025
ff6782d
Merge branch 'main' into orchestrator-websock
cjlapao Nov 26, 2025
aed0747
fix: Add periodic WebSocket ping and resolve race condition in persis…
sai-kumar-peddireddy Nov 27, 2025
4b03d5d
feat: Implement periodic host connection monitoring and automatic rec…
sai-kumar-peddireddy Nov 27, 2025
0108352
Refactor orchestrator database methods to improve thread safety and a…
sai-kumar-peddireddy Nov 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
447 changes: 447 additions & 0 deletions src/data/concurrency_test.go
View file
Open in desktop

Large diffs are not rendered by default.

13 changes: 12 additions & 1 deletion src/data/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ type JsonDatabase struct {
saveProcess chan bool
filename string
saveMutex sync.Mutex
dataMutex sync.RWMutex
cancel chan bool
data Data
}
Expand Down Expand Up @@ -289,7 +290,9 @@ func (j *JsonDatabase) processSave(ctx basecontext.ApiContext) error {

defer tempFile.Close()

j.dataMutex.RLock()
jsonString, err := json.MarshalIndent(j.data, "", " ")
j.dataMutex.RUnlock()
if err != nil {
ctx.LogDebugf("[Database] Error marshalling data to temp file: %v", err)
j.isSaving = false
Expand Down Expand Up @@ -466,13 +469,15 @@ func (j *JsonDatabase) recoverFromResidualSaveFiles(ctx basecontext.ApiContext,
ctx.LogInfof("[Database] Save file %s is empty, ignoring", latestSaveFile)
return false, nil
}
err = json.Unmarshal(content, &data)
if err != nil {
ctx.LogErrorf("[Database] Error unmarshalling save file %s: %v", latestSaveFile, err)
return false, err
}
j.dataMutex.Lock()
j.data = data
j.connected = true
j.dataMutex.Unlock()

ctx.LogInfof("[Database] Successfully recovered data from save file %s", latestSaveFile)
if err := j.SaveNow(ctx); err != nil {
ctx.LogErrorf("[Database] Error saving database: %v", err)
Expand Down Expand Up @@ -518,7 +523,9 @@ func (j *JsonDatabase) recoverFromBackupFile(ctx basecontext.ApiContext) error {
ctx.LogErrorf("[Database] Error unmarshalling backup file %s: %v", latestBackupFile, err)
return err
}
j.dataMutex.Lock()
j.data = data
j.dataMutex.Unlock()
ctx.LogInfof("[Database] Successfully recovered data from backup file %s", latestBackupFile)
return nil
}
Expand Down Expand Up @@ -568,13 +575,16 @@ func (j *JsonDatabase) loadFromFile(ctx basecontext.ApiContext) error {
}
}

j.dataMutex.Lock()
j.data = data
j.connected = true
j.dataMutex.Unlock()
return nil
}

func (j *JsonDatabase) loadFromEmpty(ctx basecontext.ApiContext) error {
ctx.LogInfof("[Database] Database file is empty, creating new file")
j.dataMutex.Lock()
j.data = Data{
Users: make([]models.User, 0),
Claims: make([]models.Claim, 0),
Expand All @@ -583,6 +593,7 @@ func (j *JsonDatabase) loadFromEmpty(ctx basecontext.ApiContext) error {
PackerTemplates: make([]models.PackerTemplate, 0),
ManifestsCatalog: make([]models.CatalogManifest, 0),
}
j.dataMutex.Unlock()

if j.Config.AutoRecover {
// Check if there are any backup files available
Expand Down
Loading