1

I am developing an event-sourced Electric Vehicle Charging Station Management System, which is connected to several Charging Stations. In this domain, I've come up with an aggregate for the Charging Station, which includes the internal state of the Charging Station(whether it is network-connected, if a car is charging using the station's connectors).

The station notifies me about its state through messages defined in a standardized protocol:

  • Heartbeat: whether the station is still "alive"
  • StatusNotification: whether the station has encountered an error(under voltage), or if everything is correct

And my server can send commands to this station:

  • RemoteStartTransaction: tells the station to unlock and reserve one of its connectors, for a car to charge using the connector.

I've developed an Aggregate for this Charging Station. It contains the internal entities of its connector, whether it's charging or not, if it has a problem in the power system, ...

And the Aggregate, which its memory representation resides in the server that I control, not in the Charging Station itself, has a StationClient service, which is responsible for sending these commands to the physical Charging Station(pseudocode):

class StationAggregate {
  stationClient: StationClient
  URL: string
  connector: Connector[]

  unlock(connectorId) {
    if this.connectors.find(connectorId).isAvailableToBeUnlocked() {
      return ErrorConnectorNotAvailable
    }
    error = this.stationClient.sendRemoteStartTransaction(this.URL, connectorId)
    if error {
      return ErrorStationRejectedUnlock
    }
    this.applyEvents([
      StationUnlockedEvent(connectorId, now())
    ])
    return Ok
  }

  receiveHeartbeat(timestamp) {
    this.applyEvents([
      StationSentHeartbeat(timestamp)
    ])
    return Ok
  }
}

I am using a optimistic concurrency, which means that, I load the Aggregate from a list of events, and I store the current version of the Aggregate in its memory representation: StationAggregate in version #2032, when a command is successfully processed and event(s) applied, it would the in version #2033, for example. In that way, I can put a unique constraint on the (StationID, Version) tuple on my persistence layer, and guarantee that only one event is persisted.

If by any chance, occurs a receival of a Heartbeat message, and the receival of a Unlock command. In both threads, they would load the StationAggregate and would be both in version X, in the case of the Heartbeat receival, there would be no side-effects, but in the case of the Unlock command, there would be a side-effect that tells the physical Charging Station to be unlocked. However as I'm using optimistic concurrency, that StationUnlocked event could be rejected from the persistence layer. I don't know how I could handle that, as I can't retry the command, because it its inherently not idempotent(as the physical Station would reject the second request)

I don't know if I'm modelling something wrong, or if it's really a hard domain to model.

5
  • Why are you storing the arrival of every Heartbeat message in your eventlog if they don't trigger a state change? How critical is it for restoring the state of your Aggregate that the events are recorded in exactly the order they arrived, or could you deal with the situation that some concurrent events are recorded in a different order? Jun 10, 2020 at 12:31
  • @BartvanIngenSchenau I am storing this event because it belongs to the Station's history, and it does can change its state, if the Station is offline for a while and it sends this Heartbeat message I know that the connection with the Station is established. Only specific events could "conflict" with one another, for example two StationUnlocked events would definitely conflict, but in this case a StationSentHeartbeat and StationUnlocked event would not conflict
    – Henke
    Jun 10, 2020 at 13:48
  • 1
    Rather than storing every heartbeat, why not store the events StationCameOnline and StationWentOffline. You will need the latter anyway to notice in the replay when there is a large enough gap in the hearbeats. And a true conflict would only arise when different orders of the events give a different end-result, which I only see if the Station allows fewer Transactions than it has Connectors. Jun 10, 2020 at 14:03
  • I would assume I need some part of my code to track when the Heartbeat has not arrived in, e.g., 5 minutes, to say the Station is Offline, right? Assuming the implementation is an in-memory timer, that resets itself when a Heartbeat message comes in(setTimeout(() => station.goOffline(), 5.minutes). How would I persist this state in case of a crash in the service? For now, I am persisting these Heartbeats in the event store, and in case of a restart of the service, I look the past events to see if I was in the middle of this timer countdown
    – Henke
    Jun 11, 2020 at 16:08
  • But, on the other subject, what if an event that is non-conflicting with the StationUnlock event comes also, I don't think the problem is specifically with the Heartbeat event being stored(although we can discuss if it is indeed a valid domain event), It could be any other event happening to this Station Aggregate. The thing is, how would I replay/retry the UnlockStation command, as it has side-effects and it's not idempotent?
    – Henke
    Jun 11, 2020 at 16:11

1 Answer 1

1

I load the Aggregate from a list of events, and I store the current version of the Aggregate in its memory representation

Rather than storing the current, versioned, memory representation of StationAggregate on every event, you should store the events themselves in an append-only repository and only store the current state of StationAggregate every X time or Y events.

When loading a StationAggregate, you would first load the last saved snapshot of the state and then the events that occurred after that snapshot was taken. Then you replay those events, where you only make the corresponding internal state changes but not any calls to outside systems.

When you now have two events hat occur at the same time, they both can be added to the datastore without any problems (as they are just appended in whatever order) and the next time the StationAggregate gets loaded, both events are reflected in the internal state.


With regard to how to deal with the Heartbeat events, that depends on where the StationAggregate instance lives most of the time.

If the StationAggregate exists mostly in RAM and only needs to be reloaded from the database when the system starts up (initial boot, reboot or crash), then I wouldn't even bother to persist the connection state. It is far easier to start with the assumption that the Station is offline and wait for (or request if the protocol supports it) the first Heartbeat message. The only reason for persisting connection status events would be for later reporting/analysis and that is best handled with CameOnline/WentOffline events.

If the StationAggregate exists mostly in the database and gets loaded for every request/event, then it depends on how the timers work in your system if you need to persist the Heartbeat events or just CameOnline/WentOffline events.

If your timers can be set to fire an event after a configurable time and they can be restarted before the timeout elapses, then the CameOnline/WentOffline events would be sufficient.

If your timers can't be restarted, then you would need to store the Heartbeat events to be able to periodically check how long ago the last one was received in order to detect that a Station went offline.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.