gorp

package
v0.0.0-...-d441601 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 3, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AppendMessageIsNewer 

func AppendMessageIsNewer(state *State, msg *AppendMessage) bool

func AppendMessageIsUpToDate 

func AppendMessageIsUpToDate(state *State, msg *AppendMessage) bool

Compare the terms of the message and the machine.

func AppendMessageLogIsLonger 

func AppendMessageLogIsLonger(state *State, msg *AppendMessage) bool

when two leaders can't decide which one should be a leader, it resolves to which one has more committed messages

func ApplyData 

func ApplyData(entry LogEntry) error

func CanVoteFor 

func CanVoteFor(state *State, msg *RequestVoteMessage) bool

func ConfigureLog 

func ConfigureLog(logLevel *string)

func GenerateHash 

func GenerateHash(message json.RawMessage) (string, error)

func NumMajority 

func NumMajority(state *State) int

Get the number of machines needed for a majority, does remove the calling machine from the majority. AKA the number of separate machines needed for a majority.

func PrevLogsMatch 

func PrevLogsMatch(state *State, msg *AppendMessage) bool

If the log at the previous index does not contain the same term, then return false

func VoteMsgIsUpToDate 

func VoteMsgIsUpToDate(state *State, msg *RequestVoteMessage) bool

We may be facing issues with this function (maybe with the append message functions too), because of the following excerpt from the Raft paper: "Raft determines which of two logs is more up-to-date by comparing the index and term of the last entries in the logs. If the logs have last entries with different terms, then the log with the later term is more up-to-date. If the logs end with the same term, then whichever log is longer is more up-to-date."

Types

type AppendMessage 

type AppendMessage struct {
	Term     int
	LeaderId PeerAddress

	// -1 indicates that the log is empty
	PrevLogIndex int

	PrevLogTerm int
	Entry       LogEntry

	LeaderCommit int
}

type AppendMessageReply 

type AppendMessageReply struct {
	CommitTerm int
	Success    bool
}

type Broker 

type Broker struct {
	Role Role
	// contains filtered or unexported fields
}

A wrapper for the state/role so that RPC calls don't have to be specific about the concrete type it is making the call to. All its methods forward to the Role struct it has an instance of.

func FromState 

func FromState(state *State) Broker

For start up

func (*Broker) AppendMessage 

func (broker *Broker) AppendMessage(ap AppendMessage, apr *AppendMessageReply) error

func (*Broker) Execute 

func (broker *Broker) Execute(ctx context.Context)

func (*Broker) NextRole 

func (broker *Broker) NextRole(ctx context.Context) (Role, error)

func (*Broker) RequestVote 

func (broker *Broker) RequestVote(rvm RequestVoteMessage, rvp *RequestVoteReply) error

func (*Broker) StartClientServer 

func (broker *Broker) StartClientServer()

func (*Broker) StartRPCServer 

func (broker *Broker) StartRPCServer()

func (*Broker) StopClientServer 

func (broker *Broker) StopClientServer()

func (*Broker) StopRPCServer 

func (broker *Broker) StopRPCServer()

func (*Broker) SwitchRole 

func (broker *Broker) SwitchRole(role Role)

type Candidate 

type Candidate struct {
	State *State

	ChangeSignal chan *RoleTransition
}

func NewCandidate 

func NewCandidate() *Candidate

func (*Candidate) AppendMessage 

func (candidate *Candidate) AppendMessage(message AppendMessage, reply *AppendMessageReply) error

Another machine has turned into a leader. It has done so with a majority of votes, so now this machine should also accept the leader. Turn to follower if term is equal to or less than the message term

func (*Candidate) Execute 

func (candidate *Candidate) Execute(ctx context.Context)

func (*Candidate) GetChangeSignal 

func (candidate *Candidate) GetChangeSignal() chan *RoleTransition

func (*Candidate) GetState 

func (candidate *Candidate) GetState() *State

func (*Candidate) HandleClient 

func (candidate *Candidate) HandleClient(w http.ResponseWriter, r *http.Request)

func (*Candidate) Init 

func (candidate *Candidate) Init(state *State) *Candidate

func (*Candidate) RequestVote 

func (candidate *Candidate) RequestVote(msg RequestVoteMessage, rply *RequestVoteReply) error

func (*Candidate) SendRequest 

func (candidate *Candidate) SendRequest(ctx context.Context, vote_status chan bool, element string)

type ConfigData 

type ConfigData struct {
	Old []PeerAddress `json:"old"`
	New []PeerAddress `json:"new"`
}

type Follower 

type Follower struct {
	State *State

	ChangeSignal chan *RoleTransition
	// contains filtered or unexported fields
}

func NewFollower 

func NewFollower() *Follower

func (*Follower) AppendMessage 

func (follower *Follower) AppendMessage(message AppendMessage, reply *AppendMessageReply) error

func (*Follower) Apply 

func (follower *Follower) Apply()

func (*Follower) Execute 

func (follower *Follower) Execute(ctx context.Context)

Checks on the heartbeat and turns into a candidate if no pulse

func (*Follower) GetChangeSignal 

func (follower *Follower) GetChangeSignal() chan *RoleTransition

func (*Follower) GetState 

func (follower *Follower) GetState() *State

func (*Follower) HandleClient 

func (follower *Follower) HandleClient(w http.ResponseWriter, r *http.Request)

func (*Follower) Init 

func (follower *Follower) Init(state *State) *Follower

func (*Follower) RequestVote 

func (follower *Follower) RequestVote(msg RequestVoteMessage, rply *RequestVoteReply) error

type Leader 

type Leader struct {
	State *State

	// essentially a queue of messages that we have to try and append
	MessageQueue chan LogEntry

	ChangeSignal chan *RoleTransition
	// contains filtered or unexported fields
}

func NewLeader 

func NewLeader() *Leader

Factory functions to create new role instances

func (*Leader) AppendMessage 

func (leader *Leader) AppendMessage(msg AppendMessage, rply *AppendMessageReply) error

func (*Leader) Apply 

func (leader *Leader) Apply()

func (*Leader) Execute 

func (leader *Leader) Execute(ctx context.Context)

func (*Leader) GetChangeSignal 

func (leader *Leader) GetChangeSignal() chan *RoleTransition

func (*Leader) GetState 

func (leader *Leader) GetState() *State

func (*Leader) GetStatus 

func (leader *Leader) GetStatus(hash string) StatusResponse

func (*Leader) HandleClient 

func (leader *Leader) HandleClient(w http.ResponseWriter, r *http.Request)

HandleClient routes incoming client requests to the appropriate handler.

Endpoints: 

POST /api
  Submit a new log entry (write, update, delete) to the cluster.
  Request body:
    {
      "type": "data",
      "message": {
        "path": "documents/hello.txt",
        "blob": "hello world",
        "operation": "write"
      }
    }
  Response (200):
    {
      "hash": "abc123...",
      "timestamp": "2024-01-01T00:00:00.000000000Z"
    }

  Examples:

    Write a file:
      curl -X POST http://localhost:8080/api \
        -H "Content-Type: application/json" \
        -d '{"type":"data","message":{"path":"docs/hello.txt","blob":"hello world","operation":"write"}}'

    Update a file:
      curl -X POST http://localhost:8080/api \
        -H "Content-Type: application/json" \
        -d '{"type":"data","message":{"path":"docs/hello.txt","blob":"updated content","operation":"update"}}'

    Delete a file:
      curl -X POST http://localhost:8080/api \
        -H "Content-Type: application/json" \
        -d '{"type":"data","message":{"path":"docs/hello.txt","blob":"","operation":"delete"}}'

GET /status?hash=<hash>
  Poll for the status of a previously submitted log entry.
  Response (200):
    { "status": "success", "hash": "abc123..." }
    { "status": "pending", "hash": "abc123..." }
    { "status": "failed",  "hash": "abc123..." }

  Example:
    curl "http://localhost:8080/status?hash=abc123..."

GET /file?path=<path>
  Read a file from the committed state machine.
  Response (200): "file contents here"
  Response (404): "File not found."

  Example:
    curl "http://localhost:8080/file?path=docs/hello.txt"

GET /files
  List all files in the data directory.
  Response (200):
    ["docs/hello.txt", "docs/world.txt"]

func (*Leader) Init 

func (leader *Leader) Init(state *State) *Leader

func (*Leader) Replicate 

func (leader *Leader) Replicate(parent_ctx context.Context, msg LogEntry)

In its current form, this implementation may cause issues, cancelling normally functioning appends when a majority is achieved. This needs further testing, but it should still offer certainty that a majority has replicated any one log message.

func (*Leader) RequestVote 

func (leader *Leader) RequestVote(msg RequestVoteMessage, rply *RequestVoteReply) error

func (*Leader) SendHeartbeats 

func (leader *Leader) SendHeartbeats(ctx context.Context)

type LogEntry 

type LogEntry struct {
	Term int `json:"term"`

	// Type specifies what type of message this should be interpreted as. This
	// may be a config change, update to storage, or potential snapshot data
	Type    string          `json:"type"`
	Message json.RawMessage `json:"message"`

	// Hash is generated from the message content and used to track
	// the status of the entry. Persisted so a new leader can recover this.
	Hash      string    `json:"hash"`
	Timestamp time.Time `json:"timestamp"`
}

type MessageData 

type MessageData struct {
	Path      string `json:"path"`
	Blob      string `json:"blob"`
	Operation string `json:"operation"` // "write", "delete", "update"
}

type PeerAddress 

type PeerAddress struct {
	Host     string `json:"host"`
	RPCPort  int    `json:"rpcport"`
	HTTPPort int    `json:"httpport"`
}

func (PeerAddress) HTTPAddr 

func (p PeerAddress) HTTPAddr() string

func (PeerAddress) RPCAddr 

func (p PeerAddress) RPCAddr() string

type RequestVoteMessage 

type RequestVoteMessage struct {
	Term        int
	CandidateId PeerAddress

	LastLogIndex int
	LastLogTerm  int
}

type RequestVoteReply 

type RequestVoteReply struct {
	Term int

	VoteGranted bool
}

type Role 

type Role interface {
	// RPC methods
	RequestVote(RequestVoteMessage, *RequestVoteReply) error
	AppendMessage(AppendMessage, *AppendMessageReply) error

	// Used to implement the actual logic of the replicas/individual roles
	// Returns the next role to transition to
	Execute(context.Context)

	HandleClient(w http.ResponseWriter, r *http.Request)

	GetChangeSignal() chan *RoleTransition

	// Used as a type check to ensure that the role has a relation
	// to the underlying state object of the replica
	GetState() *State
}

type RoleTransition 

type RoleTransition struct {
	RoleName string // "leader", "candidate", "follower", or "" for shutdown
	State    *State
}

RoleTransition represents a request to transition to a new role

type State 

type State struct {
	// used to redirect clients to the appropriate leader
	Leader PeerAddress `json:"leader"`

	// the running replica's host/port
	PeerAddress PeerAddress `json:"address"`

	// mostly used for debugging and testing, stores a string representation of the current role
	Role string `json:"role"`

	// the set of servers participating in consensus
	Config []PeerAddress `json:"config"`

	// persistent state
	Log        []LogEntry  `json:"log"`
	CommitTerm int         `json:"commitTerm"`
	VotedFor   PeerAddress `json:"votedFor"`

	// volatile state
	CommitIndex int `json:"commitIndex"`
	LastApplied int `json:"lastApplied"`

	// timeout in milliseconds
	ElectionTimeout int `json:"electionTimeout"`
	// Index 0 is the minimum timeout, and index 1 is the maximum, used in
	// candidate replicas to calculate how long they should time out for after
	// losing an election
	RandomizedTimeout []int `json:"randomizedTimeout"`
}

func EmptyState 

func EmptyState() State

func (*State) Debug_Print 

func (state *State) Debug_Print()

type StatusResponse 

type StatusResponse struct {
	Status string `json:"status"` // "success", "pending", "failed"
	Hash   string `json:"hash"`
}

Response sent back when a client polls for status

type SubmitResponse 

type SubmitResponse struct {
	Hash      string `json:"hash"`
	Timestamp string `json:"timestamp"`
}

Response sent back to the client when they first submit a message

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.