generated from TBD54566975/tbd-project-template
-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
This is a typed version of dragonboat APIs. It also does some simplifications, namely - No immutable files in snapshots - No support for cancelling snapshot save / restore - No active sessions, which can result into duplicate events on timeouts / errors. Also, it does not support adding / removing members from the cluster yet. However, it supports - typed events - typed queries - snapshotting
- Loading branch information
Showing
6 changed files
with
985 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/binary" | ||
| "errors" | ||
| "fmt" | ||
| "io" | ||
| "log" | ||
| "os" | ||
| "os/signal" | ||
| "time" | ||
|
|
||
| "github.com/alecthomas/kong" | ||
| "github.com/block/ftl/internal/raft" | ||
| "github.com/lni/dragonboat/v4" | ||
| "golang.org/x/exp/rand" | ||
| "golang.org/x/sync/errgroup" | ||
| ) | ||
|
|
||
| var cli struct { | ||
| RaftConfig raft.RaftConfig `embed:"" prefix:"raft-"` | ||
| } | ||
|
|
||
| type IntStateMachine struct { | ||
| sum int64 | ||
| } | ||
|
|
||
| type IntEvent int64 | ||
|
|
||
| func (i *IntEvent) UnmarshalBinary(data []byte) error { //nolint:unparam | ||
| *i = IntEvent(binary.BigEndian.Uint64(data)) | ||
| return nil | ||
| } | ||
|
|
||
| func (i IntEvent) MarshalBinary() ([]byte, error) { //nolint:unparam | ||
| return binary.BigEndian.AppendUint64([]byte{}, uint64(i)), nil | ||
| } | ||
|
|
||
| var _ raft.StateMachine[int64, int64, IntEvent, *IntEvent] = &IntStateMachine{} | ||
|
|
||
| func (s IntStateMachine) Lookup(key int64) (int64, error) { | ||
| return s.sum, nil | ||
| } | ||
|
|
||
| func (s *IntStateMachine) Update(msg IntEvent) error { | ||
| s.sum += int64(msg) | ||
| return nil | ||
| } | ||
|
|
||
| func (s IntStateMachine) Close() error { | ||
| return nil | ||
| } | ||
|
|
||
| func (s IntStateMachine) Recover(reader io.Reader) error { | ||
| err := binary.Read(reader, binary.BigEndian, &s.sum) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to recover from snapshot: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s IntStateMachine) Save(writer io.Writer) error { | ||
| err := binary.Write(writer, binary.BigEndian, s.sum) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to save snapshot: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func main() { | ||
| kctx := kong.Parse(&cli) | ||
| ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt) | ||
|
|
||
| cluster := raft.New(&cli.RaftConfig) | ||
| shard := raft.AddShard(ctx, cluster, 1, &IntStateMachine{}) | ||
|
|
||
| wg, ctx := errgroup.WithContext(ctx) | ||
| messages := make(chan int) | ||
|
|
||
| wg.Go(func() error { | ||
| defer close(messages) | ||
| // send a random number every 10 seconds | ||
| ticker := time.NewTicker(10 * time.Second) | ||
| defer ticker.Stop() | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| messages <- rand.Intn(1000) | ||
| case <-ctx.Done(): | ||
| return nil | ||
| } | ||
| } | ||
| }) | ||
| wg.Go(func() error { | ||
| return cluster.Start(ctx, nil) | ||
| }) | ||
| wg.Go(func() error { | ||
| ticker := time.NewTicker(10 * time.Second) | ||
| for { | ||
| select { | ||
| case msg := <-messages: | ||
| ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) | ||
| defer cancel() | ||
|
|
||
| err := shard.Propose(ctx, IntEvent(msg)) | ||
| if errors.Is(err, dragonboat.ErrShardNotReady) { | ||
| log.Println("shard not ready") | ||
| } else if err != nil { | ||
| return fmt.Errorf("failed to propose event: %w", err) | ||
| } | ||
| case <-ticker.C: | ||
| ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) | ||
| defer cancel() | ||
|
|
||
| state, err := shard.Query(ctx, 1) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to query shard: %w", err) | ||
| } | ||
| log.Println("state: ", state) | ||
| case <-ctx.Done(): | ||
| return nil | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| if err := wg.Wait(); err != nil { | ||
| kctx.FatalIfErrorf(err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.